Dirk Riemann
ebefee4184
Task 4.3 closes the loop on Chunk 4: the FastAPI lifespan now selects fake vs real clients via IX_TEST_MODE (new AppConfig field), wires /healthz probes to the live selfcheck() on OllamaClient / SuryaOCRClient, and spawns the worker with a production Pipeline factory that builds SetupStep -> OCRStep -> GenAIStep -> ReliabilityStep -> ResponseHandler over the injected clients. Factories: - make_genai_client(cfg) -> FakeGenAIClient | OllamaClient - make_ocr_client(cfg) -> FakeOCRClient | SuryaOCRClient (spec §6.2) Probes run the async selfcheck on a fresh event loop in a short-lived thread so they're safe to call from either sync callers or a live FastAPI handler without stalling the request loop. Drops the worker-loop spawn_worker_task stub — the app module owns the production spawn directly. Tests: +11 unit tests (5 factories + 6 app-wiring / probe adapter / pipeline build). Full suite: 236 passed. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
"""Tests for ``ix.app`` lifespan / probe wiring (Task 4.3).
|
|
|
|
The lifespan selects fake clients when ``IX_TEST_MODE=fake`` and exposes
|
|
their probes via the route DI hook. These tests exercise the probe
|
|
adapter in isolation — no DB, no real Ollama/Surya.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Literal
|
|
|
|
from ix.app import _make_ocr_probe, _make_ollama_probe, build_pipeline
|
|
from ix.config import AppConfig
|
|
from ix.genai.fake import FakeGenAIClient
|
|
from ix.ocr.fake import FakeOCRClient
|
|
from ix.pipeline.genai_step import GenAIStep
|
|
from ix.pipeline.ocr_step import OCRStep
|
|
from ix.pipeline.pipeline import Pipeline
|
|
from ix.pipeline.reliability_step import ReliabilityStep
|
|
from ix.pipeline.response_handler_step import ResponseHandlerStep
|
|
from ix.pipeline.setup_step import SetupStep
|
|
|
|
|
|
def _cfg(**overrides: object) -> AppConfig:
|
|
return AppConfig(_env_file=None, **overrides) # type: ignore[call-arg]
|
|
|
|
|
|
class _SelfcheckOllamaClient:
|
|
async def invoke(self, *a: object, **kw: object) -> object:
|
|
raise NotImplementedError
|
|
|
|
async def selfcheck(
|
|
self, expected_model: str
|
|
) -> Literal["ok", "degraded", "fail"]:
|
|
self.called_with = expected_model
|
|
return "ok"
|
|
|
|
|
|
class _SelfcheckOCRClient:
|
|
async def ocr(self, *a: object, **kw: object) -> object:
|
|
raise NotImplementedError
|
|
|
|
async def selfcheck(self) -> Literal["ok", "fail"]:
|
|
return "ok"
|
|
|
|
|
|
class _BrokenSelfcheckOllama:
|
|
async def invoke(self, *a: object, **kw: object) -> object:
|
|
raise NotImplementedError
|
|
|
|
async def selfcheck(
|
|
self, expected_model: str
|
|
) -> Literal["ok", "degraded", "fail"]:
|
|
raise RuntimeError("boom")
|
|
|
|
|
|
class TestOllamaProbe:
|
|
def test_fake_client_without_selfcheck_reports_ok(self) -> None:
|
|
cfg = _cfg(test_mode="fake", default_model="gpt-oss:20b")
|
|
probe = _make_ollama_probe(FakeGenAIClient(parsed=None), cfg)
|
|
assert probe() == "ok"
|
|
|
|
def test_real_selfcheck_returns_its_verdict(self) -> None:
|
|
cfg = _cfg(default_model="gpt-oss:20b")
|
|
client = _SelfcheckOllamaClient()
|
|
probe = _make_ollama_probe(client, cfg) # type: ignore[arg-type]
|
|
assert probe() == "ok"
|
|
assert client.called_with == "gpt-oss:20b"
|
|
|
|
def test_selfcheck_exception_falls_back_to_fail(self) -> None:
|
|
cfg = _cfg(default_model="gpt-oss:20b")
|
|
probe = _make_ollama_probe(_BrokenSelfcheckOllama(), cfg) # type: ignore[arg-type]
|
|
assert probe() == "fail"
|
|
|
|
|
|
class TestOCRProbe:
|
|
def test_fake_client_without_selfcheck_reports_ok(self) -> None:
|
|
from ix.contracts.response import OCRDetails, OCRResult
|
|
|
|
probe = _make_ocr_probe(FakeOCRClient(canned=OCRResult(result=OCRDetails())))
|
|
assert probe() == "ok"
|
|
|
|
def test_real_selfcheck_returns_its_verdict(self) -> None:
|
|
probe = _make_ocr_probe(_SelfcheckOCRClient()) # type: ignore[arg-type]
|
|
assert probe() == "ok"
|
|
|
|
|
|
class TestBuildPipeline:
|
|
def test_assembles_all_five_steps_in_order(self) -> None:
|
|
from ix.contracts.response import OCRDetails, OCRResult
|
|
|
|
genai = FakeGenAIClient(parsed=None)
|
|
ocr = FakeOCRClient(canned=OCRResult(result=OCRDetails()))
|
|
cfg = _cfg(test_mode="fake")
|
|
pipeline = build_pipeline(genai, ocr, cfg)
|
|
assert isinstance(pipeline, Pipeline)
|
|
steps = pipeline._steps # type: ignore[attr-defined]
|
|
assert [type(s) for s in steps] == [
|
|
SetupStep,
|
|
OCRStep,
|
|
GenAIStep,
|
|
ReliabilityStep,
|
|
ResponseHandlerStep,
|
|
]
|