From d2138e535fddf23ee3612fcb0460c38d793a1dc3 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 21 Jul 2026 23:08:33 -0400 Subject: [PATCH] test: add MonkeyOCR live integration suite --- tests/integration/test_monkey_ocr_live.py | 138 ++++++++++++++++++++++ 1 file changed, 138 insertions(+) create mode 100644 tests/integration/test_monkey_ocr_live.py diff --git a/tests/integration/test_monkey_ocr_live.py b/tests/integration/test_monkey_ocr_live.py new file mode 100644 index 0000000..51ea58e --- /dev/null +++ b/tests/integration/test_monkey_ocr_live.py @@ -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