303 lines
11 KiB
Python
303 lines
11 KiB
Python
"""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.run(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.run(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.run(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.run(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.run(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.run(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.run(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.run(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.run(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.run(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.run(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.run(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.run(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)
|