replica-omnisciente/telminal/tests/test_process.py

"""Tests for the PTY-backed process engine (real pexpect)."""

from __future__ import annotations

import asyncio
import time
from pathlib import Path

import pytest

from telminal.process import TProcess


def _drive(coro_factory):
    """Run a coroutine factory under a fresh event loop and return its result.

    The factory is responsible for calling ``proc.run(stream=True)`` (which uses
    ``asyncio.create_task`` and therefore requires a running loop).
    """

    async def _wrapper():
        return await coro_factory()

    try:
        loop = asyncio.get_event_loop()
        if loop.is_running():
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
    except RuntimeError:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
    return loop.run_until_complete(_wrapper())


def test_process_captures_output():
    def factory():
        proc = TProcess("echo hello-world", 1)
        proc.set_temp_path(Path("/tmp"))
        proc.run(stream=True)

        async def wait():
            deadline = time.time() + 5
            while proc.is_running and time.time() < deadline:
                await asyncio.sleep(0.05)
            await asyncio.sleep(0.15)
            return proc.full_output

        return wait()

    assert "hello-world" in _drive(factory)


def test_process_exit_code_and_done():
    def factory():
        proc = TProcess("exit 7", 2)
        proc.set_temp_path(Path("/tmp"))
        proc.run(stream=True)

        async def wait():
            deadline = time.time() + 5
            while proc.is_running and time.time() < deadline:
                await asyncio.sleep(0.05)
            await asyncio.sleep(0.1)
            return proc.is_running

        return wait()

    assert _drive(factory) is False


def test_process_streaming_updates_buffer():
    def factory():
        proc = TProcess("printf 'line1\\nline2\\n'", 3)
        proc.set_temp_path(Path("/tmp"))
        proc.run(stream=True)

        async def wait():
            deadline = time.time() + 5
            while proc.is_running and time.time() < deadline:
                await asyncio.sleep(0.05)
            await asyncio.sleep(0.1)
            return proc.full_output

        return wait()

    out = _drive(factory)
    assert "line1" in out
    assert "line2" in out


def test_push_sends_plain_text_and_newline():
    def factory():
        proc = TProcess("cat", 4)
        proc.set_temp_path(Path("/tmp"))
        proc.run(stream=True)
        proc.push("ping\n")

        async def wait():
            # let the streaming task copy pty bytes into the buffer
            await asyncio.sleep(0.4)
            sent = "ping" in proc.full_output
            proc.terminate()
            return sent

        return wait()

    assert _drive(factory) is True


def test_push_control_char():
    # `push("^c")` must route to sendcontrol (transmit Ctrl-C) without error.
    # We assert the control char is delivered to the pty (echoed as ^C) rather
    # than relying on the child's SIGINT handler, which depends on real TTY job
    # control not present in this non-controlling pty.
    def factory():
        proc = TProcess("cat", 5)
        proc.set_temp_path(Path("/tmp"))
        proc.run(stream=True)
        proc.push("^c")

        async def wait():
            await asyncio.sleep(0.3)
            delivered = "^C" in proc.full_output or "\x03" in proc.full_output
            proc.terminate()
            return delivered

        return wait()

    assert _drive(factory) is True

def test_media_strip_truncates_long_output():
    proc = TProcess("true", 6)
    proc.set_temp_path(Path("/tmp"))
    proc._buffer.write("x" * 2000)
    stripped = proc.media_output
    assert len(stripped) == 1024
    assert stripped.endswith("x")


def test_html_render_contains_output():
    import tempfile

    d = Path(tempfile.mkdtemp())

    def factory():
        proc = TProcess("echo abc", 7)
        proc.set_temp_path(d)
        proc.run(stream=True)

        async def wait():
            try:
                proc._process.expect("abc", timeout=5)
            except Exception:
                pass
            deadline = time.time() + 5
            while proc.is_running and time.time() < deadline:
                await asyncio.sleep(0.05)
            await asyncio.sleep(0.1)
            return proc.html

        return wait()

    html_path = _drive(factory)
    content = Path(html_path).read_text()
    assert "abc" in content
    assert "<html" in content


def test_update_buttons_changes_for_running_vs_done():
    def factory():
        proc = TProcess("sleep 0.2", 8)
        proc.set_temp_path(Path("/tmp"))
        proc.run(stream=True)

        async def wait():
            first = proc.update_buttons()
            first_len = len(proc.buttons)
            deadline = time.time() + 5
            while proc.is_running and time.time() < deadline:
                await asyncio.sleep(0.05)
            await asyncio.sleep(0.1)
            second = proc.update_buttons()
            second_len = len(proc.buttons)
            return first, first_len, second, second_len

        return wait()

    first, first_len, second, second_len = _drive(factory)
    assert first is True
    assert first_len == 5
    assert second is True
    assert second_len == 2