Spaces:
Sleeping
Sleeping
File size: 6,899 Bytes
fe0c99f ef2b969 fe0c99f ef2b969 fe0c99f ef2b969 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 | 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"
|