"""MonkeyOcrTransport 测试(M3 设计 §4): 双端点协议、两段解析、数值防御、错误翻译。

fixtures 按 2026-07-21 真实服务取证响应**脱敏二次构造**(设计 §11):
保留结构骨架/数值/元素类型,OCR 识别文本一律替换为合成占位(零业务假设 +
P5 医疗数据不入库)。
"""

import asyncio
import io
import json
import zipfile

import httpx
import pytest

from polygateway.errors import (
    RequestRejectedError,
    ResultInvalidError,
    SourceDeadError,
    TransientError,
)
from polygateway.transports.monkey_ocr import (
    MonkeyOcrTransport,
    _classify_status,
    _parse_middle_json,
)
from polygateway.types import SourceConfig


def _source(**overrides):
    base = {
        "name": "monkey_1",
        "provider": "monkey",
        "base_url": "http://10.77.0.20:7866",
        "api_key": "none",  # MonkeyOCR 无鉴权,占位惯例(CHS .env 同款)
        "model": "monkey-ocr",
        "timeout_s": 120.0,
    }
    base.update(overrides)
    return SourceConfig(**base)


def _text_body(content="LINE-1\nLINE-2"):
    # 真实形态: {"success":true,"task_type":"text","content":...,"message":...}
    return {
        "success": True,
        "task_type": "text",
        "content": content,
        "message": "Text extraction completed successfully",
    }


def _parse_body(download_url="/static/sample_parsed_1.zip"):
    # 真实形态含 output_dir/files;解析只消费 success 与 download_url
    return {
        "success": True,
        "message": "image parsing (standard) completed successfully",
        "output_dir": "/app/tmp/x",
        "files": ["u_middle.json"],
        "download_url": download_url,
    }


def _page(para_blocks, page_size=(759.0, 540.0), page_idx=0):
    return {
        "page_idx": page_idx,
        "page_size": list(page_size),
        "para_blocks": para_blocks,
        "tables": [b for b in para_blocks if b.get("type") == "table"],
    }


def _block(type_, bbox):
    body = {"type": type_, "bbox": list(bbox), "index": 0}
    if type_ == "text":
        body["lines"] = [{"spans": [{"content": "LINE-SYNTH"}]}]
    return body


_DEFAULT_PAGES = [
    _page(
        [
            _block("table", (41, 48, 218, 282)),
            _block("image", (240, 30, 700, 500)),
            _block("text", (50, 300, 200, 340)),
        ]
    )
]


def _middle_bytes(pages=_DEFAULT_PAGES, top=None):
    payload = {"pdf_info": pages, "_parse_type": "ocr", "_version_name": "0.0.1"}
    if top is not None:
        payload = top
    return json.dumps(payload).encode()


def _zip_bytes(middle=None, member="sample_middle.json"):
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, "w") as zf:
        zf.writestr(member, middle if middle is not None else _middle_bytes())
        zf.writestr("sample.md", "# synth")
    return buf.getvalue()


def _transport_for(handler):
    mock = httpx.MockTransport(handler)
    return MonkeyOcrTransport(
        client_factory=lambda src: httpx.AsyncClient(base_url=src.base_url, transport=mock)
    )


def _routes(text_resp=None, parse_resp=None, zip_resp=None, health_resp=None):
    """按路径分发的 mock handler;None 项返回 500 便于暴露误路由。"""

    def handler(request: httpx.Request) -> httpx.Response:
        path = request.url.path
        if path == "/ocr/text":
            return text_resp or httpx.Response(500)
        if path == "/parse":
            return parse_resp or httpx.Response(500)
        if path.startswith("/static/"):
            return zip_resp or httpx.Response(500)
        if path == "/health":
            return health_resp or httpx.Response(500)
        return httpx.Response(500)

    return handler


class TestRecognizeText:
    async def test_normal(self):
        t = _transport_for(_routes(text_resp=httpx.Response(200, json=_text_body())))
        r = await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
        assert r.text == "LINE-1\nLINE-2"
        assert r.raw["task_type"] == "text"

    async def test_empty_content_legal(self):
        t = _transport_for(_routes(text_resp=httpx.Response(200, json=_text_body(""))))
        r = await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
        assert r.text == ""

    async def test_content_missing_result_invalid(self):
        body = _text_body()
        del body["content"]
        t = _transport_for(_routes(text_resp=httpx.Response(200, json=body)))
        with pytest.raises(ResultInvalidError):
            await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")

    async def test_content_non_str_result_invalid(self):
        t = _transport_for(_routes(text_resp=httpx.Response(200, json=_text_body(123))))
        with pytest.raises(ResultInvalidError):
            await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")

    async def test_multipart_shape(self):
        seen = {}

        def handler(request: httpx.Request) -> httpx.Response:
            seen["content_type"] = request.headers.get("content-type", "")
            seen["body"] = request.read()
            return httpx.Response(200, json=_text_body())

        t = _transport_for(handler)
        await t.recognize_text(image=b"IMAGE-BYTES", source=_source(), call_id="c1")
        assert seen["content_type"].startswith("multipart/form-data")
        assert b"IMAGE-BYTES" in seen["body"]
        assert b'filename="image.jpg"' in seen["body"]  # CHS invokers.py:496 固定名


class TestParseLayout:
    async def test_full_chain_relative_url(self):
        t = _transport_for(
            _routes(
                parse_resp=httpx.Response(200, json=_parse_body()),
                zip_resp=httpx.Response(200, content=_zip_bytes()),
            )
        )
        r = await t.parse_layout(image=b"jpg", source=_source(), call_id="c1")
        types = [e.type for e in r.elements]
        assert types == ["table", "image", "text"]
        assert r.elements[0].bbox == (41.0, 48.0, 218.0, 282.0)
        assert r.elements[0].page_index == 0
        assert r.page_sizes == [(759.0, 540.0)]

    async def test_absolute_download_url(self):
        url = "http://10.77.0.20:7866/static/abs.zip"
        t = _transport_for(
            _routes(
                parse_resp=httpx.Response(200, json=_parse_body(url)),
                zip_resp=httpx.Response(200, content=_zip_bytes()),
            )
        )
        r = await t.parse_layout(image=b"jpg", source=_source(), call_id="c1")
        assert len(r.elements) == 3

    async def test_success_false_rejected_with_status_200(self):
        body = {"success": False, "message": "unsupported"}
        t = _transport_for(_routes(parse_resp=httpx.Response(200, json=body)))
        with pytest.raises(RequestRejectedError) as ei:
            await t.parse_layout(image=b"jpg", source=_source(), call_id="c1")
        # 设计 §10.2 有意修复: 200 响应确证服务活着,熔断按"响应即健康"记成功
        assert ei.value.status_code == 200

    async def test_download_url_missing_transient(self):
        body = _parse_body()
        del body["download_url"]
        t = _transport_for(_routes(parse_resp=httpx.Response(200, json=body)))
        with pytest.raises(TransientError):
            await t.parse_layout(image=b"jpg", source=_source(), call_id="c1")

    async def test_non_json_response_transient(self):
        t = _transport_for(_routes(parse_resp=httpx.Response(200, content=b"<html>")))
        with pytest.raises(TransientError):
            await t.parse_layout(image=b"jpg", source=_source(), call_id="c1")

    async def test_no_table_page_legal(self):
        pages = [_page([_block("text", (1, 2, 30, 40))])]
        t = _transport_for(
            _routes(
                parse_resp=httpx.Response(200, json=_parse_body()),
                zip_resp=httpx.Response(200, content=_zip_bytes(_middle_bytes(pages))),
            )
        )
        r = await t.parse_layout(image=b"jpg", source=_source(), call_id="c1")
        assert [e.type for e in r.elements] == ["text"]  # 无 table = 合法,不抛


class TestMiddleJsonDefense:
    """数值防御纯函数(CHS invokers.py:427-479 全量下沉)。"""

    def _expect_invalid(self, zip_bytes):
        with pytest.raises(ResultInvalidError):
            _parse_middle_json(zip_bytes)

    def test_bad_zip(self):
        self._expect_invalid(b"not-a-zip")

    def test_missing_middle_member(self):
        self._expect_invalid(_zip_bytes(member="other.json"))

    def test_pdf_info_not_list(self):
        self._expect_invalid(_zip_bytes(_middle_bytes(top={"pdf_info": {}})))

    def test_middle_not_json(self):
        self._expect_invalid(_zip_bytes(b"{broken"))

    @pytest.mark.parametrize(
        "page_size", [[-1, 540], [float("nan"), 540], [True, 540], [759], "bad"]
    )
    def test_page_size_invalid(self, page_size):
        page = _page([_block("table", (1, 2, 3, 4))])
        page["page_size"] = page_size
        self._expect_invalid(_zip_bytes(_middle_bytes([page])))

    def test_bbox_order_invalid(self):
        self._expect_invalid(
            _zip_bytes(_middle_bytes([_page([_block("table", (218, 48, 41, 282))])]))
        )

    def test_bbox_int_degenerate(self):
        # float 合法但 int() 后宽度为零: 专为 CHS shim 的裁剪路径兜底
        self._expect_invalid(
            _zip_bytes(_middle_bytes([_page([_block("table", (1.2, 1.2, 1.8, 5))])]))
        )

    def test_bbox_non_finite(self):
        self._expect_invalid(
            _zip_bytes(_middle_bytes([_page([_block("table", (1, 2, float("inf"), 4))])]))
        )

    def test_type_missing(self):
        block = {"bbox": [1, 2, 30, 40], "index": 0}
        self._expect_invalid(_zip_bytes(_middle_bytes([_page([block])])))

    def test_para_blocks_absent_defaults_empty(self):
        page = _page([])
        del page["para_blocks"]
        elements, page_sizes = _parse_middle_json(_zip_bytes(_middle_bytes([page])))
        assert elements == [] and page_sizes == [(759.0, 540.0)]

    def test_page_idx_missing_falls_back_to_enumeration(self):
        page = _page([_block("text", (1, 2, 30, 40))])
        del page["page_idx"]
        elements, _ = _parse_middle_json(_zip_bytes(_middle_bytes([page])))
        assert elements[0].page_index == 0

    @pytest.mark.parametrize("bad_idx", [True, -3, "2", 1.5])
    def test_page_idx_non_int_falls_back(self, bad_idx):
        # bool/负数/非 int 一律回退枚举序(与 _finite_number 拒 bool 的防御口径一致)
        page = _page([_block("text", (1, 2, 30, 40))])
        page["page_idx"] = bad_idx
        elements, _ = _parse_middle_json(_zip_bytes(_middle_bytes([page])))
        assert elements[0].page_index == 0


class TestErrorTranslation:
    @pytest.mark.parametrize(
        ("status", "exc_type"),
        [
            (502, TransientError),
            (429, TransientError),
            (401, SourceDeadError),
            (403, SourceDeadError),
            (404, RequestRejectedError),
        ],
    )
    async def test_http_status(self, status, exc_type):
        t = _transport_for(_routes(text_resp=httpx.Response(status)))
        with pytest.raises(exc_type) as ei:
            await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
        assert ei.value.status_code == status

    @pytest.mark.parametrize(
        ("status", "exc_type"),
        [(502, TransientError), (401, SourceDeadError), (404, RequestRejectedError)],
    )
    async def test_body_survives_every_branch(self, status, exc_type):
        """issue #10: OCR 侧 message 原本只有 HTTP 状态码,拒绝理由同样丢失。"""
        body = '{"detail":"unsupported image mode CMYK"}'
        t = _transport_for(_routes(text_resp=httpx.Response(status, content=body.encode())))
        with pytest.raises(exc_type) as ei:
            await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
        assert ei.value.body_text == body
        assert str(ei.value).endswith(f" | {body}")

    def test_unread_body_degrades_without_changing_class(self):
        """取不到 body 时降级空串: 绝不能让 ResponseNotRead 逃出错误四分类。

        直接测纯函数而非走 MockTransport——真实客户端对非 stream 请求总会读完
        响应,未读态只可能在将来给 OCR 加 stream 时出现,而那正是要防的场景。
        """

        class _Unread(httpx.SyncByteStream):
            def __iter__(self):
                yield b"body"

        exc = httpx.HTTPStatusError(
            "404",
            request=httpx.Request("POST", "http://ocr.example/ocr/text"),
            response=httpx.Response(404, stream=_Unread()),
        )
        err = _classify_status(exc, "monkey_1", "text")
        assert isinstance(err, RequestRejectedError)
        assert err.body_text == ""
        assert str(err) == "monkey_1 OCR text HTTP 404"

    async def test_connect_error_transient(self):
        def handler(request):
            raise httpx.ConnectError("refused", request=request)

        t = _transport_for(handler)
        with pytest.raises(TransientError) as ei:
            await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
        assert ei.value.status_code is None

    async def test_read_timeout_transient_with_cause(self):
        def handler(request):
            raise httpx.ReadTimeout("slow", request=request)

        t = _transport_for(handler)
        with pytest.raises(TransientError) as ei:
            await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
        # __cause__ 链保留,RetryMW._failure_reason 依赖它归类 timeout
        assert isinstance(ei.value.__cause__, httpx.TimeoutException)


class TestCheckHealth:
    async def test_healthy(self):
        t = _transport_for(_routes(health_resp=httpx.Response(200, json={"status": "healthy"})))
        assert await t.check_health(source=_source()) is True

    async def test_500_unhealthy(self):
        t = _transport_for(_routes(health_resp=httpx.Response(500)))
        assert await t.check_health(source=_source()) is False

    async def test_3xx_unhealthy(self):
        # 设计 §10.1 有意收紧: VT 的 resp.ok 含 3xx,库只认 2xx
        t = _transport_for(_routes(health_resp=httpx.Response(302)))
        assert await t.check_health(source=_source()) is False

    async def test_connect_failure_false(self):
        def handler(request):
            raise httpx.ConnectError("refused", request=request)

        t = _transport_for(handler)
        assert await t.check_health(source=_source()) is False

    async def test_cancellation_passes_through(self):
        def handler(request):
            raise asyncio.CancelledError()

        t = _transport_for(handler)
        with pytest.raises(asyncio.CancelledError):
            await t.check_health(source=_source())


class TestLifecycle:
    async def test_aclose_idempotent(self):
        t = _transport_for(_routes(health_resp=httpx.Response(200)))
        await t.check_health(source=_source())
        await t.aclose()
        await t.aclose()