tests/test_capacity_worker_render.py

from __future__ import annotations

import io
import json
import queue
import threading
from types import SimpleNamespace

from api import bridge
from api import capacity_slide_worker as worker


def test_render_metrics_only_true_returns_expected_metrics():
    payload = {
        "text": "[CHK-001] synthetic calibration sentence.",
        "width": 800,
        "height": 520,
        "with_image": False,
        "page_index": 0,
        "metrics_only": True,
    }
    out = worker._render(payload)
    assert out["success"] is True
    assert out["total_pages"] >= 1
    assert out["chars_on_page"] >= 1
    assert "page_text" in out


def test_render_metrics_only_false_returns_data_url():
    payload = {
        "text": "[CHK-001] synthetic calibration sentence to render output image.",
        "width": 900,
        "height": 560,
        "with_image": True,
        "page_index": 0,
        "metrics_only": False,
    }
    out = worker._render(payload)
    assert out["success"] is True
    assert out["data_url"].startswith("data:image/png;base64,")


def test_main_emits_error_json_on_invalid_input(monkeypatch):
    monkeypatch.setattr("sys.stdin", io.StringIO("{"))
    out_stream = io.StringIO()
    monkeypatch.setattr("sys.stdout", out_stream)
    code = worker.main()
    assert code == 1
    payload = json.loads(out_stream.getvalue())
    assert payload["success"] is False
    assert "error" in payload


# ── Bridge worker reliability tests ──────────────────────────────


def _capture_logger(monkeypatch):
    """Replace bridge.logger with a recording stub and return the captured list."""
    captured = []

    def _record(level: str):
        def _log(fmt, *args):
            try:
                captured.append((level, fmt % args if args else fmt))
            except (TypeError, ValueError):
                captured.append((level, fmt))
        return _log

    fake_logger = SimpleNamespace(
        warning=_record("warning"),
        error=_record("error"),
        info=_record("info"),
        debug=_record("debug"),
    )
    monkeypatch.setattr(bridge, "logger", fake_logger)
    return captured


def test_drain_stream_to_logger_forwards_lines_and_handles_close(monkeypatch):
    captured = _capture_logger(monkeypatch)
    stream = io.StringIO("first warning\nsecond warning\n\n")
    bridge._drain_stream_to_logger(stream, "render-worker-test")
    warnings = [msg for level, msg in captured if level == "warning"]
    assert any("first warning" in m for m in warnings)
    assert any("second warning" in m for m in warnings)
    assert stream.closed


def test_drain_stream_to_logger_tolerates_none_stream(monkeypatch):
    _capture_logger(monkeypatch)
    bridge._drain_stream_to_logger(None, "render-worker-test")


def test_read_stdout_to_queue_forwards_then_signals_eof():
    q: "queue.Queue" = queue.Queue()
    stream = io.StringIO('{"success": true}\n')
    bridge._read_stdout_to_queue(stream, q)
    first = q.get_nowait()
    sentinel = q.get_nowait()
    assert first == '{"success": true}\n'
    assert sentinel is None
    assert stream.closed


def test_read_stdout_to_queue_signals_eof_on_none_stream():
    q: "queue.Queue" = queue.Queue()
    bridge._read_stdout_to_queue(None, q)
    assert q.get_nowait() is None


def _make_test_api(monkeypatch):
    """Build a CalculusAPI shell with bridge state set up but no real worker."""
    api = bridge.CalculusAPI.__new__(bridge.CalculusAPI)
    api._render_worker = None
    api._render_worker_lock = threading.Lock()
    api._render_response_queue = queue.Queue()
    api._render_worker_stopping = threading.Event()
    api._render_worker_restart_failures = 0
    return api


def test_run_render_task_times_out_when_worker_silent(monkeypatch):
    monkeypatch.setattr(bridge, "_RENDER_TIMEOUT_SEC", 0.05)
    captured = _capture_logger(monkeypatch)

    api = _make_test_api(monkeypatch)

    written: list = []

    class _FakeStdin:
        def write(self, payload):
            written.append(payload)

        def flush(self):
            pass

    proc = SimpleNamespace(
        stdin=_FakeStdin(),
        returncode=None,
    )
    proc.poll = lambda: None
    proc.kill = lambda: setattr(proc, "returncode", -9)
    proc.wait = lambda timeout=None: None
    api._render_worker = proc

    out = api._run_render_task({"slide_title": "x"})

    assert out["success"] is False
    assert "timeout" in out["error"].lower()
    assert written, "payload should have been written before the timeout"
    assert any("timed out" in msg for level, msg in captured if level == "error")
    # Worker was killed and dropped so the next call would respawn.
    assert api._render_worker is None


def test_run_render_task_returns_error_when_worker_signals_eof(monkeypatch):
    monkeypatch.setattr(bridge, "_RENDER_TIMEOUT_SEC", 1.0)
    _capture_logger(monkeypatch)

    api = _make_test_api(monkeypatch)

    class _FakeStdin:
        def write(self, payload):
            pass

        def flush(self):
            pass

    proc = SimpleNamespace(
        stdin=_FakeStdin(),
        returncode=-1,
    )
    proc.poll = lambda: -1
    proc.kill = lambda: None
    proc.wait = lambda timeout=None: None
    api._render_worker = proc
    # Pre-stage the EOF sentinel that the stdout reader would push on worker exit
    # and bypass restart by keeping proc.poll alive at write time.
    proc.poll = lambda: None
    api._render_response_queue.put(None)

    out = api._run_render_task({"slide_title": "y"})
    assert out["success"] is False
    assert "exited" in out["error"].lower()


def test_run_render_task_surfaces_unavailable_when_restart_fails(monkeypatch):
    _capture_logger(monkeypatch)
    api = _make_test_api(monkeypatch)

    def _failed_restart():
        api._render_worker = None

    monkeypatch.setattr(api, "_start_render_worker_locked", _failed_restart)

    out = api._run_render_task({"slide_title": "z"})
    assert out["success"] is False
    assert out["error"] == "Render worker unavailable"


def test_capacity_test_slide_returns_capability_unavailable():
    api = bridge.CalculusAPI.__new__(bridge.CalculusAPI)
    out = json.loads(api.capacity_test_slide("hello"))
    assert out["success"] is False
    assert out["error"] == "capability_unavailable"
    assert "reason" in out


def test_capacity_metrics_only_returns_capability_unavailable():
    api = bridge.CalculusAPI.__new__(bridge.CalculusAPI)
    out = api._capacity_metrics_only("hello")
    assert out["success"] is False
    assert out["error"] == "capability_unavailable"


def test_run_capacity_worker_returns_capability_unavailable():
    api = bridge.CalculusAPI.__new__(bridge.CalculusAPI)
    out = api._run_capacity_worker("hello")
    assert out["success"] is False
    assert out["error"] == "capability_unavailable"