PolyGateway/tests/integration/test_monkey_ocr_live.py

"""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",
            # 服务在 LAN,开发机若开着系统代理(httpx trust_env 读 macOS 系统配置,
            # 不是环境变量),代理会对内网地址回 403 —— 与本文件 raw httpx 用例
            # 显式传 trust_env=False 同因
            f"OCR__MONKEY__{n}__TRUST_ENV": "false",
        }
    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