test: add MonkeyOCR live integration suite

2026-07-21 23:08:33 -04:00
parent 26e0e72926
commit d2138e535f
1 changed files with 138 additions and 0 deletions
@@ -0,0 +1,138 @@
+"""MonkeyOCR 真实服务集成测试(M3 计划 T7;ROADMAP §4 验收出口)。
+
+前置: LAN 服务 10.77.0.20:7866/7867(不可达即 FAIL——验收必打真实,不 skip);
+语料 `data/soak/chs_images/`(仅实验室机器有,缺失 skip);
+Redis 组合用例沿既有约定: 缺 REDIS_URL 时 skip。
+与 soak 不并跑(Redis db3 会被 FLUSHDB)。
+"""
+
+import os
+import time
+from pathlib import Path
+
+import httpx
+import pytest
+from dotenv import dotenv_values
+
+from polygateway.ocr import OcrClient
+from polygateway.transports.monkey_ocr import _parse_middle_json
+
+_IMAGES = Path("data/soak/chs_images")
+_TABLE_IMAGE = _IMAGES / "chs_0001.jpg"  # 取证已核: 有表
+_PLAIN_IMAGE = _IMAGES / "chs_0002.jpg"  # 取证已核: 无表
+_PRIMARY = "http://10.77.0.20:7866"
+_SECONDARY = "http://10.77.0.20:7867"
+_BLACKHOLE = "http://10.255.255.1:7866"
+_REDIS_URL = os.environ.get("REDIS_URL") or dotenv_values(".env").get("REDIS_URL")
+
+pytestmark = pytest.mark.skipif(
+    not _TABLE_IMAGE.exists(), reason="需实验室语料 data/soak/chs_images(见 findings §6)"
+)
+
+
+def _env(sources: dict[int, str], **extra: str) -> dict[str, str]:
+    env = {
+        "LLM_MAX_RETRIES": "3",
+        "LLM_RETRY_BASE_DELAY": "1.0",
+        "LLM_RETRY_MAX_DELAY": "5.0",
+        "LLM_CIRCUIT_BREAKER_THRESHOLD": "2",
+        "LLM_CIRCUIT_BREAKER_COOLDOWN": "30",
+        "PGW_CACHE_BACKEND": "none",
+        "PGW_TELEMETRY_BACKEND": "none",
+    }
+    for n, base_url in sources.items():
+        env |= {
+            f"OCR__MONKEY__{n}__BASE_URL": base_url,
+            f"OCR__MONKEY__{n}__API_KEY": "none",
+            f"OCR__MONKEY__{n}__MODEL": "monkey-ocr",
+            f"OCR__MONKEY__{n}__TIMEOUT_S": "300",
+        }
+    env.update(extra)
+    return env
+
+
+class TestLiveEndpoints:
+    async def test_recognize_text_returns_content(self):
+        async with OcrClient.from_env("OCR", env=_env({1: _PRIMARY})) as client:
+            result = await client.recognize_text(_TABLE_IMAGE.read_bytes())
+        assert result.text.strip()  # 真实护理记录图必有文字
+        assert result.source_name == "monkey_1"
+        assert result.usage.prompt_tokens == 0 and result.latency_ms > 0
+
+    async def test_parse_layout_table_and_plain(self):
+        async with OcrClient.from_env("OCR", env=_env({1: _PRIMARY})) as client:
+            with_table = await client.parse_layout(_TABLE_IMAGE.read_bytes())
+            plain = await client.parse_layout(_PLAIN_IMAGE.read_bytes())
+        tables = [e for e in with_table.elements if e.type == "table"]
+        assert tables, "取证样本 chs_0001 应检出表格"
+        x1, y1, x2, y2 = tables[0].bbox
+        assert x2 > x1 and y2 > y1 and with_table.page_sizes[0][0] > 0
+        # 无表样本: elements 无 table 且不抛异常(合法"无表"语义)
+        assert all(e.type != "table" for e in plain.elements)
+
+    async def test_middle_json_consistency_guard(self):
+        """设计 §1.2 持续护栏: tables 列表与库全元素提取的 table bbox 一致。"""
+        async with httpx.AsyncClient(base_url=_PRIMARY, trust_env=False, timeout=300) as raw:
+            resp = await raw.post(
+                "/parse", files={"file": ("image.jpg", _TABLE_IMAGE.read_bytes(), "image/jpeg")}
+            )
+            resp.raise_for_status()
+            zip_resp = await raw.get(resp.json()["download_url"])
+            zip_resp.raise_for_status()
+        elements, _ = _parse_middle_json(zip_resp.content)
+        lib_tables = {e.bbox for e in elements if e.type == "table"}
+
+        import io
+        import json
+        import zipfile
+
+        with zipfile.ZipFile(io.BytesIO(zip_resp.content)) as archive:
+            middle = next(n for n in archive.namelist() if n.endswith("_middle.json"))
+            payload = json.load(archive.open(middle))
+        raw_tables = {
+            tuple(float(v) for v in t["bbox"])
+            for page in payload["pdf_info"]
+            for t in page.get("tables", [])
+        }
+        assert lib_tables == raw_tables and raw_tables
+
+    async def test_check_health_per_source(self):
+        env = _env({1: _PRIMARY, 2: _SECONDARY, 3: _BLACKHOLE})
+        started = time.monotonic()
+        async with OcrClient.from_env("OCR", env=env) as client:
+            health = await client.check_health()
+        elapsed = time.monotonic() - started
+        assert health["monkey_1"] is True
+        assert health["monkey_2"] is True
+        assert health["monkey_3"] is False  # 黑洞: 5s 探测超时判 False
+        assert elapsed < 30  # 走探测常量 5s,不等 TIMEOUT_S=300
+
+
+@pytest.mark.skipif(not _REDIS_URL, reason="需实验室远程 Redis: 在 .env 设置 REDIS_URL")
+class TestRedisGovernanceCombo:
+    def _redis_env(self, sources):
+        return _env(
+            sources,
+            PGW_LIMITER_BACKEND="redis",
+            PGW_BREAKER_BACKEND="redis",
+            REDIS_URL=_REDIS_URL,
+        )
+
+    async def test_real_call_through_redis_backends(self):
+        async with OcrClient.from_env("OCR", env=self._redis_env({1: _PRIMARY})) as client:
+            result = await client.recognize_text(_PLAIN_IMAGE.read_bytes())
+        assert result.text is not None
+
+    async def test_blackhole_pool_circuit_opens_with_retry_after(self):
+        """G1 真实后端联调: 全池熔断后 retry_after_s 来自 Redis gate。"""
+        from polygateway.errors import GatewayUnavailableError
+
+        env = self._redis_env({1: _BLACKHOLE})
+        env["OCR__MONKEY__1__TIMEOUT_S"] = "3"
+        env["OCR__RETRY__MAX_ATTEMPTS"] = "2"
+        async with OcrClient.from_env("OCR", env=env) as client:
+            with pytest.raises(GatewayUnavailableError) as ei:
+                await client.recognize_text(_PLAIN_IMAGE.read_bytes())
+        exc = ei.value
+        assert exc.per_source_reasons  # G1: 逐源原因非空
+        assert exc.retry_after_s >= 0