"""ports.py 端口冻结测试(M1 设计 §4): Protocol 结构性检查 + Gate 快照校验。""" from typing import Any import pytest from polygateway.ports import ( CacheBackend, GateDecision, GateState, GateUpdate, Middleware, Permit, ProviderGate, RateLimiter, SourceSelector, StructuredOutputStrategy, TelemetryRecorder, Transport, ) from polygateway.types import LLMResponse, SourceStats def _resp() -> LLMResponse: return LLMResponse("c", "t", "m", "p", 1, 2, 3, None, None, False, "cid") class _DummyPermit: async def release(self) -> None: ... async def settle(self, actual_tokens: int) -> None: ... class _DummyLimiter: async def try_acquire(self, source_key: str, est_tokens: int): return _DummyPermit() async def acquire(self, source_key: str, est_tokens: int): return _DummyPermit() async def source_stats(self, source_key: str): return SourceStats(0, 0, 0) async def mark_progress(self) -> None: ... async def progress_age_s(self) -> float: return 0.0 class _DummyGate: async def try_enter(self, source_name: str, owner: str): raise NotImplementedError async def record_success(self, entry): raise NotImplementedError async def record_failure(self, entry, reason: str, force_open: bool): raise NotImplementedError async def release_probe(self, entry): raise NotImplementedError async def retry_after_s(self, sources): return 0.0 class _DummyMw: async def __call__(self, request, call_next): return await call_next(request) class _DummyTransport: async def complete(self, *, messages, source, stream, overlay, call_id): raise NotImplementedError class _DummyCache: async def get(self, key: str): return None async def set(self, key: str, value: str, ttl_s: int) -> None: ... class _DummySelector: def order(self, sources, stats): return list(sources) class _DummyStrategy: def request_overlay(self, schema): return {} def parse(self, text: str) -> Any: return {} class _DummyRecorder: async def record_llm_call( self, *, call_id, parent_call_id, session_id, model, provider, source_name, messages, response, thinking, prompt_tokens, completion_tokens, usage_source, latency_ms, ttft_ms, max_inter_token_ms, cache_hit, error, cost, cached_prompt_tokens, model_reported, ) -> None: ... @pytest.mark.parametrize( ("impl", "protocol"), [ (_DummyPermit(), Permit), (_DummyLimiter(), RateLimiter), (_DummyGate(), ProviderGate), (_DummyMw(), Middleware), (_DummyTransport(), Transport), (_DummyCache(), CacheBackend), (_DummySelector(), SourceSelector), (_DummyStrategy(), StructuredOutputStrategy), (_DummyRecorder(), TelemetryRecorder), ], ) def test_protocols_are_runtime_checkable(impl, protocol): assert isinstance(impl, protocol) def _decision(**overrides) -> GateDecision: base = { "source_name": "qwen_1", "allowed": True, "state": GateState.CLOSED, "epoch": 0, "is_probe": False, "probe_owner": None, "retry_after_s": 0.0, } base.update(overrides) return GateDecision(**base) class TestGateDecisionInvariants: """校验逐条移植 CHS ports.py:405-440。""" def test_valid_probe_decision(self): d = _decision(state=GateState.HALF_OPEN, is_probe=True, probe_owner="w1") assert d.is_probe and d.probe_owner == "w1" def test_empty_source_rejected(self): with pytest.raises(ValueError): _decision(source_name=" ") def test_negative_epoch_and_retry_after_rejected(self): with pytest.raises(ValueError): _decision(epoch=-1) with pytest.raises(ValueError): _decision(retry_after_s=-0.1) def test_open_state_cannot_allow(self): with pytest.raises(ValueError): _decision(state=GateState.OPEN, allowed=True) def test_half_open_admission_must_be_probe(self): with pytest.raises(ValueError): _decision(state=GateState.HALF_OPEN, is_probe=False) def test_probe_requires_owner_and_half_open(self): with pytest.raises(ValueError): _decision(state=GateState.HALF_OPEN, is_probe=True, probe_owner=None) with pytest.raises(ValueError): _decision(state=GateState.CLOSED, is_probe=True, probe_owner="w1") def test_non_probe_cannot_carry_owner(self): with pytest.raises(ValueError): _decision(probe_owner="w1") class TestGateUpdate: def test_bounds(self): u = GateUpdate( applied=True, state=GateState.CLOSED, epoch=0, failure_count=0, retry_after_s=0.0 ) assert u.applied with pytest.raises(ValueError): GateUpdate( applied=True, state=GateState.CLOSED, epoch=-1, failure_count=0, retry_after_s=0.0 ) with pytest.raises(ValueError): GateUpdate( applied=True, state=GateState.CLOSED, epoch=0, failure_count=-1, retry_after_s=0.0 ) class TestOcrPorts: """M3 三个 OCR Protocol(设计 §3.2): runtime_checkable 结构判定。""" def test_ocr_ports_runtime_checkable(self): from polygateway.ports import OcrLayoutPort, OcrTextPort, OcrTransport class GoodClient: async def recognize_text(self, image): ... async def parse_layout(self, image): ... class GoodTransport: async def recognize_text(self, *, image, source, call_id): ... async def parse_layout(self, *, image, source, call_id): ... async def check_health(self, *, source): ... assert isinstance(GoodClient(), OcrTextPort) assert isinstance(GoodClient(), OcrLayoutPort) assert isinstance(GoodTransport(), OcrTransport) def test_missing_method_rejected(self): from polygateway.ports import OcrTransport class NoHealth: async def recognize_text(self, *, image, source, call_id): ... async def parse_layout(self, *, image, source, call_id): ... assert not isinstance(NoHealth(), OcrTransport)