openhop_repeater/tests/test_plugin_progress.py

import json
import sys
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch

import pytest

from repeater.plugins.ipc import PluginIPCClient, PluginIPCServer
from repeater.plugins.manager import (
    PROGRESS_MAX_LINES,
    PROGRESS_MAX_LOGS,
    OperationProgress,
    PluginManager,
    PluginManagerError,
)
from repeater.plugins.runtime import PluginRuntime
from repeater.plugins.storage import PluginStorage
from repeater.web.plugin_endpoints import PluginAPIEndpoints


def test_runtime_hands_each_output_line_to_the_listener(tmp_path: Path):
    runtime = PluginRuntime(PluginStorage(tmp_path / "plugins"))
    heard: list[str] = []
    runtime.on_output = heard.append
    script = "import sys; print('Collecting demo'); print('Installing', end=''); sys.stdout.flush()"
    result = runtime._run_install_command([sys.executable, "-c", script], timeout=30)
    assert result.returncode == 0
    assert heard == ["Collecting demo", "Installing"]
    assert "Collecting demo" in result.stdout

    def broken(_line: str) -> None:
        raise RuntimeError("listener bug")

    runtime.on_output = broken
    assert (
        runtime._run_install_command([sys.executable, "-c", "print('ok')"], timeout=30).returncode
        == 0
    )


def test_progress_log_is_bounded_and_pages_by_cursor():
    log = OperationProgress("openhop.demo", "update")
    for i in range(PROGRESS_MAX_LINES + 20):
        log.append(f"line {i}")
    snap = log.snapshot()
    assert len(snap["lines"]) == PROGRESS_MAX_LINES
    assert snap["lines"][0] == "line 20"
    assert snap["next"] == PROGRESS_MAX_LINES + 20
    log.append("tail")
    assert log.snapshot(since=snap["next"])["lines"] == ["tail"]
    log.finish("boom")
    assert (log.snapshot()["state"], log.snapshot()["error"]) == ("error", "boom")


def test_manager_records_the_outcome_and_restores_the_listener(tmp_path: Path):
    mgr = PluginManager(PluginStorage(tmp_path / "plugins"))
    assert mgr.progress("missing.plugin")["state"] == "idle"
    with pytest.raises(PluginManagerError):
        mgr.update_plugin("missing.plugin")
    snap = mgr.progress("missing.plugin")
    assert (snap["operation"], snap["state"]) == ("update", "error")
    assert "plugin not found" in snap["error"]
    assert mgr.runtime.on_output is None


def test_manager_keeps_a_bounded_number_of_logs(tmp_path: Path):
    mgr = PluginManager(PluginStorage(tmp_path / "plugins"))
    for i in range(PROGRESS_MAX_LOGS + 10):
        with pytest.raises(PluginManagerError):
            mgr.update_plugin(f"nobody.{i:03d}")
    assert len(mgr._progress) == PROGRESS_MAX_LOGS
    assert mgr.progress("nobody.000")["state"] == "idle"


def test_ipc_round_trips_progress(tmp_path: Path):
    storage = PluginStorage(tmp_path / "plugins")
    manager = PluginManager(storage, PluginRuntime(storage, popen_factory=lambda *a, **k: None))
    sock_path = Path(f"/tmp/oh-pm-progress-{tmp_path.name}.sock")
    server = PluginIPCServer(sock_path, manager)
    server.start()
    try:
        client = PluginIPCClient(sock_path)
        assert client.progress("openhop.demo")["state"] == "idle"
        with pytest.raises(PluginManagerError):
            manager.update_plugin("openhop.demo")
        assert client.progress("openhop.demo", since=0)["state"] == "error"
    finally:
        server.stop()


def _snap(state, lines=(), nxt=0, started=None, error=None):
    return {
        "state": state,
        "operation": "update",
        "lines": list(lines),
        "next": nxt,
        "error": error,
        "started": started,
    }


def _events(tmp_path, snapshots, *, ticks=None, fresh=False):
    api = PluginAPIEndpoints({"storage": {"storage_dir": str(tmp_path)}})
    answers = iter(snapshots)
    client = SimpleNamespace(progress=lambda _id, since=0: next(answers, snapshots[-1]))
    clock = iter(ticks or [0.0] * 1000)
    with patch.object(api, "_client_or_raise", return_value=client):
        chunks = api._progress_events(
            "openhop.demo", 0, fresh=fresh, sleep=lambda _s: None, clock=lambda: next(clock)
        )
        return [json.loads(chunk[len("data: ") : -2]) for chunk in chunks]


def test_stream_plays_lines_status_and_one_done(tmp_path):
    events = _events(
        tmp_path,
        [_snap("running", ["a"], 1), _snap("running", ["b", "c"], 3), _snap("complete", nxt=3)],
    )
    assert [e["type"] for e in events] == [
        "connected",
        "line",
        "status",
        "line",
        "line",
        "status",
        "done",
    ]
    assert [e["line"] for e in events if e["type"] == "line"] == ["a", "b", "c"]
    assert events[-1] == {"type": "done", "state": "complete", "error": None, "started": None}

    events = _events(tmp_path, [_snap("error", ["pip failed"], 1, error="pip")])
    assert events[-1]["state"] == "error" and events[-1]["error"] == "pip"


def test_idle_stream_keeps_alive_then_ends(tmp_path):
    events = _events(tmp_path, [_snap("idle")], ticks=[0.0] + [1.0] * 12 + [100.0])
    types = [e["type"] for e in events]
    assert types[:2] == ["connected", "status"]
    assert types.count("keepalive") == 2
    assert events[-1] == {"type": "done", "state": "idle", "error": None}

    events = _events(tmp_path, [_snap("running", started=5.0)], ticks=[0.0, 100.0, 500.0, 901.0])
    assert events[-1]["state"] == "timeout"


def test_fresh_stream_ignores_a_previous_operations_log(tmp_path):
    previous = _snap("complete", ["old"], 1, started=100.0)
    current = _snap("running", ["new"], 1, started=200.0)
    events = _events(
        tmp_path, [previous, previous, current, _snap("complete", nxt=1, started=200.0)], fresh=True
    )
    assert [e["type"] for e in events] == [
        "connected",
        "status",
        "line",
        "status",
        "status",
        "done",
    ]
    assert events[1]["state"] == "idle"
    assert events[2]["line"] == "new"
    assert events[-1]["started"] == 200.0

    events = _events(tmp_path, [previous])
    assert [e["type"] for e in events] == ["connected", "line", "status", "done"]