Video-Tree-TRM5/tests/unit/test_ocr_adapter.py

"""MonkeyOCRClient 适配器单元测试。

覆盖范围：Protocol 合规性、单帧转录、失败降级、健康检查、
多端点轮询、行去重过滤。使用 responses 库 mock HTTP。
"""

from __future__ import annotations

import asyncio
from pathlib import Path  # noqa: TC003 — pytest tmp_path fixture 类型注解需运行时

import pytest
import requests
import responses

from adapters.ocr import MonkeyOCRClient
from app.ports import OCRProvider

# ---------------------------------------------------------------------------
# Protocol 合规性
# ---------------------------------------------------------------------------


class TestOCRProviderProtocol:
    """MonkeyOCRClient 必须满足 OCRProvider Protocol。"""

    def test_is_runtime_checkable_instance(self) -> None:
        """MonkeyOCRClient 实例应通过 isinstance(_, OCRProvider) 检查。"""
        client = MonkeyOCRClient(urls=["http://localhost:7866"])
        assert isinstance(client, OCRProvider)

    def test_has_transcribe_frames(self) -> None:
        """MonkeyOCRClient 必须暴露 transcribe_frames 方法。"""
        assert hasattr(MonkeyOCRClient, "transcribe_frames")


# ---------------------------------------------------------------------------
# 构造函数校验
# ---------------------------------------------------------------------------


class TestConstructor:
    """构造函数参数校验。"""

    def test_empty_urls_raises_value_error(self) -> None:
        """空端点列表应抛 ValueError（P5：不用 assert）。"""
        with pytest.raises(ValueError, match="端点列表不能为空"):
            MonkeyOCRClient(urls=[])

    def test_trailing_slash_stripped(self) -> None:
        """URL 尾部斜杠应被去除。"""
        client = MonkeyOCRClient(urls=["http://host:7866/"])
        assert client._urls == ["http://host:7866"]


# ---------------------------------------------------------------------------
# 健康检查
# ---------------------------------------------------------------------------


class TestCheckHealth:
    """check_health 预检所有端点。"""

    @responses.activate
    def test_healthy_endpoints(self) -> None:
        """所有端点返回 200 → 无异常。"""
        url = "http://10.0.0.1:7866"
        responses.add(responses.GET, f"{url}/health", status=200)
        client = MonkeyOCRClient(urls=[url])
        asyncio.get_event_loop().run_until_complete(client.check_health())

    @responses.activate
    def test_unhealthy_endpoint_raises(self) -> None:
        """端点返回 500 → RuntimeError。"""
        url = "http://10.0.0.1:7866"
        responses.add(responses.GET, f"{url}/health", status=500)
        client = MonkeyOCRClient(urls=[url])
        with pytest.raises(RuntimeError, match="健康检查失败"):
            asyncio.get_event_loop().run_until_complete(client.check_health())

    @responses.activate
    def test_unreachable_endpoint_raises(self) -> None:
        """端点连接失败 → RuntimeError。"""
        url = "http://10.0.0.1:7866"
        responses.add(
            responses.GET,
            f"{url}/health",
            body=requests.ConnectionError("refused"),
        )
        client = MonkeyOCRClient(urls=[url])
        with pytest.raises(RuntimeError, match="端点不可达"):
            asyncio.get_event_loop().run_until_complete(client.check_health())

    @responses.activate
    def test_multiple_endpoints_all_checked(self) -> None:
        """多端点时全部预检，任一失败即报错。"""
        url_a = "http://10.0.0.1:7866"
        url_b = "http://10.0.0.2:7866"
        responses.add(responses.GET, f"{url_a}/health", status=200)
        responses.add(responses.GET, f"{url_b}/health", status=503)
        client = MonkeyOCRClient(urls=[url_a, url_b])
        with pytest.raises(RuntimeError, match="健康检查失败"):
            asyncio.get_event_loop().run_until_complete(client.check_health())


# ---------------------------------------------------------------------------
# 单帧转录
# ---------------------------------------------------------------------------


class TestTranscribeFrames:
    """transcribe_frames 核心逻辑。"""

    @responses.activate
    def test_single_frame(self, tmp_path: Path) -> None:
        """单帧正常转录 → '帧1: line_a | line_b' 格式。"""
        url = "http://ocr:7866"
        responses.add(
            responses.POST,
            f"{url}/ocr/text",
            json={"content": "Hello World\nOCR Test"},
            status=200,
        )
        frame = tmp_path / "frame_001.jpg"
        frame.write_bytes(b"\xff\xd8\xff\xe0fake")
        client = MonkeyOCRClient(urls=[url])
        result = asyncio.get_event_loop().run_until_complete(client.transcribe_frames([frame]))
        assert result == "帧1: Hello World | OCR Test"

    @responses.activate
    def test_multiple_frames(self, tmp_path: Path) -> None:
        """多帧转录 → 每帧一行，帧号递增。"""
        url = "http://ocr:7866"
        responses.add(
            responses.POST,
            f"{url}/ocr/text",
            json={"content": "Line A"},
            status=200,
        )
        responses.add(
            responses.POST,
            f"{url}/ocr/text",
            json={"content": "Line B"},
            status=200,
        )
        frames = []
        for i in range(2):
            f = tmp_path / f"frame_{i}.jpg"
            f.write_bytes(b"\xff\xd8data")
            frames.append(f)
        client = MonkeyOCRClient(urls=[url])
        result = asyncio.get_event_loop().run_until_complete(client.transcribe_frames(frames))
        assert "帧1: Line A" in result
        assert "帧2: Line B" in result

    @responses.activate
    def test_empty_frames_returns_empty(self) -> None:
        """空帧列表 → 空串。"""
        client = MonkeyOCRClient(urls=["http://ocr:7866"])
        result = asyncio.get_event_loop().run_until_complete(client.transcribe_frames([]))
        assert result == ""


# ---------------------------------------------------------------------------
# 失败降级
# ---------------------------------------------------------------------------


class TestFailureDegradation:
    """单帧失败跳过，不影响其余帧。"""

    @responses.activate
    def test_single_frame_failure_returns_empty(self, tmp_path: Path) -> None:
        """唯一帧请求失败 → 返回空串（不抛异常）。"""
        url = "http://ocr:7866"
        responses.add(responses.POST, f"{url}/ocr/text", status=500)
        frame = tmp_path / "frame.jpg"
        frame.write_bytes(b"\xff\xd8data")
        client = MonkeyOCRClient(urls=[url])
        result = asyncio.get_event_loop().run_until_complete(client.transcribe_frames([frame]))
        assert result == ""

    @responses.activate
    def test_partial_failure_skips_bad_frame(self, tmp_path: Path) -> None:
        """第一帧失败、第二帧成功 → 仅输出第二帧。"""
        url = "http://ocr:7866"
        responses.add(responses.POST, f"{url}/ocr/text", status=500)
        responses.add(
            responses.POST,
            f"{url}/ocr/text",
            json={"content": "Good"},
            status=200,
        )
        frames = []
        for i in range(2):
            f = tmp_path / f"frame_{i}.jpg"
            f.write_bytes(b"\xff\xd8data")
            frames.append(f)
        client = MonkeyOCRClient(urls=[url])
        result = asyncio.get_event_loop().run_until_complete(client.transcribe_frames(frames))
        # 帧1 失败被跳过，帧2 成功但输出为 "帧2: Good"
        assert "帧1" not in result
        assert "帧2: Good" in result


# ---------------------------------------------------------------------------
# 行去重过滤
# ---------------------------------------------------------------------------


class TestLineDedup:
    """帧内行级去重与短行过滤。"""

    @responses.activate
    def test_duplicate_lines_removed(self, tmp_path: Path) -> None:
        """帧内重复行只保留首次出现。"""
        url = "http://ocr:7866"
        responses.add(
            responses.POST,
            f"{url}/ocr/text",
            json={"content": "重复行\n重复行\n不同行"},
            status=200,
        )
        frame = tmp_path / "frame.jpg"
        frame.write_bytes(b"\xff\xd8data")
        client = MonkeyOCRClient(urls=[url])
        result = asyncio.get_event_loop().run_until_complete(client.transcribe_frames([frame]))
        assert result == "帧1: 重复行 | 不同行"

    @responses.activate
    def test_single_char_lines_filtered(self, tmp_path: Path) -> None:
        """单字符行被过滤（长度 <= 1）。"""
        url = "http://ocr:7866"
        responses.add(
            responses.POST,
            f"{url}/ocr/text",
            json={"content": "A\nAB\n.\nCD"},
            status=200,
        )
        frame = tmp_path / "frame.jpg"
        frame.write_bytes(b"\xff\xd8data")
        client = MonkeyOCRClient(urls=[url])
        result = asyncio.get_event_loop().run_until_complete(client.transcribe_frames([frame]))
        # "A" 和 "." 被过滤（长度 <= 1），保留 "AB" 和 "CD"
        assert result == "帧1: AB | CD"

    @responses.activate
    def test_empty_content_skipped(self, tmp_path: Path) -> None:
        """OCR 返回空 content → 该帧跳过。"""
        url = "http://ocr:7866"
        responses.add(
            responses.POST,
            f"{url}/ocr/text",
            json={"content": ""},
            status=200,
        )
        frame = tmp_path / "frame.jpg"
        frame.write_bytes(b"\xff\xd8data")
        client = MonkeyOCRClient(urls=[url])
        result = asyncio.get_event_loop().run_until_complete(client.transcribe_frames([frame]))
        assert result == ""


# ---------------------------------------------------------------------------
# 多端点轮询
# ---------------------------------------------------------------------------


class TestRoundRobin:
    """多端点轮询行为。"""

    @responses.activate
    def test_round_robin_alternation(self, tmp_path: Path) -> None:
        """两端点交替使用。"""
        url_a = "http://host-a:7866"
        url_b = "http://host-b:7866"
        # 为两个端点各注册响应
        responses.add(
            responses.POST,
            f"{url_a}/ocr/text",
            json={"content": "From A"},
            status=200,
        )
        responses.add(
            responses.POST,
            f"{url_b}/ocr/text",
            json={"content": "From B"},
            status=200,
        )
        frames = []
        for i in range(2):
            f = tmp_path / f"frame_{i}.jpg"
            f.write_bytes(b"\xff\xd8data")
            frames.append(f)
        client = MonkeyOCRClient(urls=[url_a, url_b])
        result = asyncio.get_event_loop().run_until_complete(client.transcribe_frames(frames))
        assert "帧1: From A" in result
        assert "帧2: From B" in result
        # 验证两个端点都被调用
        called_urls = [c.request.url for c in responses.calls]
        assert any(url_a in u for u in called_urls)
        assert any(url_b in u for u in called_urls)