124 lines
3.2 KiB
Python
124 lines
3.2 KiB
Python
"""core/protocols.py 单元测试 — 验证 Protocol 可 runtime_checkable。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from core.protocols import LLMProvider, TelemetryRecorder, VLMProvider
|
|
from core.types import LLMResponse
|
|
|
|
|
|
class _FakeLLM:
|
|
async def chat(
|
|
self,
|
|
messages: list[dict[str, Any]],
|
|
*,
|
|
session_id: str | None = None,
|
|
parent_call_id: str | None = None,
|
|
) -> LLMResponse:
|
|
return LLMResponse(
|
|
content="ok",
|
|
thinking="",
|
|
model="m",
|
|
provider="p",
|
|
prompt_tokens=1,
|
|
completion_tokens=1,
|
|
latency_ms=1,
|
|
ttft_ms=None,
|
|
max_inter_token_ms=None,
|
|
cache_hit=False,
|
|
call_id="c",
|
|
)
|
|
|
|
|
|
class _FakeVLM:
|
|
async def chat_with_images(
|
|
self,
|
|
messages: list[dict[str, Any]],
|
|
images: list[str | Path],
|
|
*,
|
|
session_id: str | None = None,
|
|
parent_call_id: str | None = None,
|
|
) -> LLMResponse:
|
|
return LLMResponse(
|
|
content="ok",
|
|
thinking="",
|
|
model="m",
|
|
provider="p",
|
|
prompt_tokens=1,
|
|
completion_tokens=1,
|
|
latency_ms=1,
|
|
ttft_ms=None,
|
|
max_inter_token_ms=None,
|
|
cache_hit=False,
|
|
call_id="c",
|
|
)
|
|
|
|
|
|
class _FakeTelemetry:
|
|
async def record_llm_call(
|
|
self,
|
|
*,
|
|
call_id: str,
|
|
parent_call_id: str | None,
|
|
session_id: str | None,
|
|
model_name: str,
|
|
provider: str,
|
|
messages: str,
|
|
response: str,
|
|
thinking: str,
|
|
prompt_tokens: int,
|
|
completion_tokens: int,
|
|
latency_ms: int,
|
|
ttft_ms: float | None,
|
|
max_inter_token_ms: float | None,
|
|
cache_hit: bool,
|
|
error: str | None,
|
|
) -> None:
|
|
pass
|
|
|
|
|
|
def test_fake_llm_satisfies_protocol() -> None:
|
|
assert isinstance(_FakeLLM(), LLMProvider)
|
|
|
|
|
|
def test_fake_vlm_satisfies_protocol() -> None:
|
|
assert isinstance(_FakeVLM(), VLMProvider)
|
|
|
|
|
|
def test_fake_telemetry_satisfies_protocol() -> None:
|
|
assert isinstance(_FakeTelemetry(), TelemetryRecorder)
|
|
|
|
|
|
def test_plain_object_does_not_satisfy() -> None:
|
|
assert not isinstance(object(), LLMProvider)
|
|
assert not isinstance(object(), VLMProvider)
|
|
assert not isinstance(object(), TelemetryRecorder)
|
|
|
|
|
|
from app.ports import PoolStrategy
|
|
|
|
|
|
class TestPoolStrategyProtocol:
|
|
"""PoolStrategy Protocol runtime_checkable 验证。"""
|
|
|
|
def test_pool_strategy_is_runtime_checkable(self) -> None:
|
|
"""PoolStrategy 支持 isinstance 检查。"""
|
|
from app.harness.pools import Pools
|
|
|
|
class FakeStrategy:
|
|
def build(self, questions, correctness, config):
|
|
return Pools(
|
|
diagnosis=[],
|
|
validation=[],
|
|
test=[],
|
|
baseline_run_id="",
|
|
baseline_val_accuracy=0.0,
|
|
)
|
|
|
|
def build_incremental(self, new_task_types, questions, correctness, config):
|
|
return {}
|
|
|
|
assert isinstance(FakeStrategy(), PoolStrategy)
|