From 93b88db2f5fc4abc247b34151d1fcbd179f8e286 Mon Sep 17 00:00:00 2001 From: xsl Date: Fri, 14 Aug 2026 00:26:39 +0800 Subject: [PATCH] test: unit suite for the gateway invariants plus an integration slice Covers routing, failover, the circuit breaker, structured-output retries, the cache-prefix validator, cost computation including cached-token pricing, the daily budget breaker, the quota protocol, the task state machine and log redaction. HTTP is mocked -- no unit test spends money. The integration test runs the real slice against PostgreSQL and Redis with a faked provider. --- tests/README.md | 48 +++ tests/conftest.py | 36 ++ tests/integration/test_vertical_slice.py | 102 +++++ tests/unit/test_circuit_breaker.py | 85 ++++ tests/unit/test_cost.py | 96 +++++ tests/unit/test_daily_budget.py | 84 ++++ tests/unit/test_deepseek_provider.py | 162 ++++++++ tests/unit/test_gateway.py | 469 +++++++++++++++++++++++ tests/unit/test_ids.py | 55 +++ tests/unit/test_log_redaction.py | 75 ++++ tests/unit/test_prompt_cache_prefix.py | 89 +++++ tests/unit/test_prompt_registry.py | 58 +++ tests/unit/test_quota_domain.py | 111 ++++++ tests/unit/test_response_envelope.py | 96 +++++ tests/unit/test_routing.py | 77 ++++ tests/unit/test_task_state.py | 66 ++++ 16 files changed, 1709 insertions(+) create mode 100755 tests/README.md create mode 100755 tests/conftest.py create mode 100755 tests/integration/test_vertical_slice.py create mode 100755 tests/unit/test_circuit_breaker.py create mode 100755 tests/unit/test_cost.py create mode 100755 tests/unit/test_daily_budget.py create mode 100755 tests/unit/test_deepseek_provider.py create mode 100755 tests/unit/test_gateway.py create mode 100755 tests/unit/test_ids.py create mode 100755 tests/unit/test_log_redaction.py create mode 100755 tests/unit/test_prompt_cache_prefix.py create mode 100755 tests/unit/test_prompt_registry.py create mode 100755 tests/unit/test_quota_domain.py create mode 100755 tests/unit/test_response_envelope.py create mode 100755 tests/unit/test_routing.py create mode 100755 tests/unit/test_task_state.py diff --git a/tests/README.md b/tests/README.md new file mode 100755 index 0000000..3a0fbb3 --- /dev/null +++ b/tests/README.md @@ -0,0 +1,48 @@ +# `tests/` + +| 目录 | 内容 | 依赖 | +| --- | --- | --- | +| `unit/` | 领域逻辑、LLM 网关、脱敏、错误码 | 无。不许连库、不许发网络请求 | +| `integration/` | 带 PostgreSQL / Redis 的链路 | `make up`，或 testcontainers | +| `golden/` | 黄金集回归（检索 / 规则 / 报告） | 尚未建立，见下 | + +```bash +make test-unit # 默认只跑 unit（integration 自动跳过） +MEIDA_RUN_INTEGRATION=1 uv run pytest -m integration +``` + +## 覆盖率门槛 + +`pytest` 对 `packages/llm` 与 `packages/domain` 统计覆盖率，低于阈值即失败。 +这两个包是 mypy strict 的范围，也是业务正确性的核心，其他层靠集成测试覆盖。 + +## 绝对不要做的事 + +- **不要在单测里调用真实的 DeepSeek**。`scripts/smoke.sh` 是唯一会真的花钱的 + 入口。单测用 `httpx.MockTransport` 或 `tests/unit/test_gateway.py` 里的 + `FakeProvider`。 +- 不要 `time.sleep()`。熔断器、预算闸门都支持注入时钟。 +- 不要断言日志文本。断言行为。 + +## 黄金集（还没建，但这是最重要的一道防线） + +架构 §16.2： + +> AI 改代码时最容易破坏的是那些「没有断言保护的隐性行为」。检索链路和规则引擎 +> 必须有一套固定输入 → 期望输出的样本集。 + +M2 落地知识库时必须同步建立（FR-KB-11，≥ 100 条用例）： +检索 Top-5 命中率 ≥ 80%、硬过滤零漏、规则命中正确性、报告结构完整性、 +**硬性禁忌违反数 = 0**。指标退化即阻断合并。 + +## 已覆盖的关键不变量 + +| 测试 | 保护的是什么 | +| --- | --- | +| `test_prompt_cache_prefix.py` | 固定前缀不含易变 token，否则输入成本翻数倍 | +| `test_daily_budget.py` | 日成本超限时**拒绝**非关键调用，不是只告警 | +| `test_gateway.py` | 路由、故障转移、熔断、结构化重试、每次调用都记成本 | +| `test_quota_domain.py` | 预扣三段式；退还幂等；服务端硬校验 | +| `test_task_state.py` | 任务状态机没有非法跃迁 | +| `test_log_redaction.py` | 手机号 / 令牌 / 密钥 / 诊断结论不进日志 | +| `test_response_envelope.py` | 每个错误码都有文案；配额文案不含价格字样 | diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100755 index 0000000..7055258 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +import os +from collections.abc import Iterator +from pathlib import Path + +import pytest + +# Unit tests never touch a real database, Redis or LLM provider. Setting these +# before any import keeps `packages.shared.config` deterministic. +os.environ.setdefault("APP_ENV", "local") +os.environ.setdefault("JWT_SECRET", "unit-test-secret-value-not-used-in-any-real-env") +os.environ.setdefault("DEEPSEEK_API_KEY", "unit-test-placeholder") + +REPO_ROOT = Path(__file__).resolve().parents[1] + + +@pytest.fixture +def repo_root() -> Path: + return REPO_ROOT + + +@pytest.fixture +def prompts_dir(tmp_path: Path) -> Iterator[Path]: + """A minimal on-disk prompt tree with a clean, cacheable system prefix.""" + root = tmp_path / "prompts" + (root / "agent_chat").mkdir(parents=True) + (root / "agent_chat" / "v1.txt").write_text( + "[system]\n你是穿搭顾问助手。只回答穿搭相关问题。\n\n[user]\n{user_input}\n", + encoding="utf-8", + ) + (root / "agent_chat" / "v2.txt").write_text( + "[system]\n你是穿搭顾问助手。回答要具体可执行。\n\n[user]\n{user_input}\n", + encoding="utf-8", + ) + yield root diff --git a/tests/integration/test_vertical_slice.py b/tests/integration/test_vertical_slice.py new file mode 100755 index 0000000..670616a --- /dev/null +++ b/tests/integration/test_vertical_slice.py @@ -0,0 +1,102 @@ +"""End-to-end vertical slice against real PostgreSQL + Redis. + +Run with ``make up && make migrate`` first, or let ``scripts/smoke.sh`` do it. +Skipped automatically when the infrastructure is not reachable, so ``make test`` +stays fast and hermetic. + +The LLM provider is faked here — ``scripts/smoke.sh`` is the one that proves the +real DeepSeek path. +""" + +from __future__ import annotations + +import os +import uuid +from decimal import Decimal +from typing import Any + +import pytest + +pytestmark = pytest.mark.integration + + +def _infra_available() -> bool: + return bool(os.getenv("MEIDA_RUN_INTEGRATION")) + + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif( + not _infra_available(), + reason="set MEIDA_RUN_INTEGRATION=1 with PostgreSQL and Redis running", + ), +] + + +@pytest.fixture +async def client() -> Any: + import httpx + + from apps.api.main import create_app + from packages.llm.types import ProviderResponse, Usage + from packages.shared.config import get_settings + + app = create_app(get_settings()) + + class FakeProvider: + name = "deepseek" + default_model = "deepseek-v4-flash" + + async def complete(self, request: Any) -> ProviderResponse: + return ProviderResponse( + text="建议白衬衫配西裤。", + model="deepseek-v4-flash", + usage=Usage(tokens_in=800, tokens_out=120, tokens_cached=600), + ) + + async def aclose(self) -> None: + return None + + transport = httpx.ASGITransport(app=app) + async with ( + httpx.AsyncClient(transport=transport, base_url="http://test") as http, + app.router.lifespan_context(app), + ): + app.state.llm_gateway._providers["deepseek"] = FakeProvider() + yield http + + +async def test_chat_slice_reserves_commits_and_records_cost(client: Any) -> None: + token_resp = await client.post( + "/api/v1/auth/dev-token", json={"device_key": f"it-{uuid.uuid4().hex[:8]}"} + ) + assert token_resp.json()["code"] == 0 + headers = {"Authorization": f"Bearer {token_resp.json()['data']['access_token']}"} + + before = (await client.get("/api/v1/quota", headers=headers)).json()["data"] + + chat = await client.post("/api/v1/agent/chat", json={"message": "面试穿什么"}, headers=headers) + body = chat.json() + assert body["code"] == 0 + assert body["data"]["reply"] + assert body["data"]["prompt_version"] == "v1" + assert Decimal(body["data"]["usage"]["cost_cents"]) > 0 + + after = (await client.get("/api/v1/quota", headers=headers)).json()["data"] + assert after["used"] == before["used"] + 1 + + ledger = (await client.get("/api/v1/quota/ledger", headers=headers)).json()["data"] + assert ledger[0]["tx_type"] == "consume" + assert ledger[0]["status"] == "committed" + + +async def test_alias_path_v1_matches_api_v1(client: Any) -> None: + assert (await client.get("/healthz")).status_code == 200 + token = await client.post("/v1/auth/dev-token", json={"device_key": "alias-check"}) + assert token.json()["code"] == 0 + + +async def test_unauthenticated_requests_get_the_unified_envelope(client: Any) -> None: + resp = await client.post("/api/v1/agent/chat", json={"message": "hi"}) + assert resp.status_code == 401 + assert resp.json()["code"] == 40100 diff --git a/tests/unit/test_circuit_breaker.py b/tests/unit/test_circuit_breaker.py new file mode 100755 index 0000000..a275d32 --- /dev/null +++ b/tests/unit/test_circuit_breaker.py @@ -0,0 +1,85 @@ +"""Circuit breaker behaviour (架构文档 §8.1 故障转移).""" + +from __future__ import annotations + +from packages.llm.circuit import CircuitBreaker, CircuitBreakerRegistry, CircuitState + + +class FakeClock: + def __init__(self) -> None: + self.now = 0.0 + + def __call__(self) -> float: + return self.now + + def advance(self, seconds: float) -> None: + self.now += seconds + + +def test_starts_closed_and_allows_traffic() -> None: + breaker = CircuitBreaker("p", failure_threshold=3, clock=FakeClock()) + assert breaker.state is CircuitState.CLOSED + assert breaker.allow() + + +def test_opens_only_after_threshold_consecutive_failures() -> None: + breaker = CircuitBreaker("p", failure_threshold=3, clock=FakeClock()) + breaker.record_failure() + breaker.record_failure() + assert breaker.state is CircuitState.CLOSED + breaker.record_failure() + assert breaker.state is CircuitState.OPEN + assert not breaker.allow() + + +def test_success_resets_the_failure_streak() -> None: + breaker = CircuitBreaker("p", failure_threshold=3, clock=FakeClock()) + breaker.record_failure() + breaker.record_failure() + breaker.record_success() + breaker.record_failure() + assert breaker.state is CircuitState.CLOSED + + +def test_half_open_probe_after_cooldown_then_closes_on_success() -> None: + clock = FakeClock() + breaker = CircuitBreaker("p", failure_threshold=1, cooldown_seconds=30.0, clock=clock) + breaker.record_failure() + assert breaker.state is CircuitState.OPEN + + clock.advance(29.0) + assert not breaker.allow() + + clock.advance(2.0) + assert breaker.allow() + assert breaker.state is CircuitState.HALF_OPEN + # only one probe is admitted + assert not breaker.allow() + + breaker.record_success() + assert breaker.state is CircuitState.CLOSED + assert breaker.allow() + + +def test_half_open_failure_reopens_and_restarts_cooldown() -> None: + clock = FakeClock() + breaker = CircuitBreaker("p", failure_threshold=1, cooldown_seconds=10.0, clock=clock) + breaker.record_failure() + clock.advance(11.0) + assert breaker.allow() + + breaker.record_failure() + assert breaker.state is CircuitState.OPEN + assert not breaker.allow() + + clock.advance(11.0) + assert breaker.allow() + + +def test_registry_returns_one_breaker_per_provider() -> None: + registry = CircuitBreakerRegistry(failure_threshold=2, cooldown_seconds=5.0) + assert registry.get("deepseek") is registry.get("deepseek") + assert registry.get("deepseek") is not registry.get("zhipu") + registry.get("deepseek").record_failure() + registry.get("deepseek").record_failure() + assert registry.snapshot() == {"deepseek": "open", "zhipu": "closed"} diff --git a/tests/unit/test_cost.py b/tests/unit/test_cost.py new file mode 100755 index 0000000..2a3522f --- /dev/null +++ b/tests/unit/test_cost.py @@ -0,0 +1,96 @@ +"""Cost computation and attribution (架构文档 §13, FR-BILL-08).""" + +from __future__ import annotations + +from decimal import Decimal + +from packages.llm.cost import ( + AICallRecord, + CompositeAICallRecorder, + InMemoryAICallRecorder, + ModelRate, + NullAICallRecorder, + RateTable, + compute_cost_cents, +) +from packages.llm.recording import PrometheusAICallRecorder +from packages.llm.types import Usage + +RATE = ModelRate( + input_cents_per_mtok=Decimal("100"), + cached_input_cents_per_mtok=Decimal("10"), + output_cents_per_mtok=Decimal("200"), +) + + +def test_cached_tokens_are_billed_at_the_cheaper_rate() -> None: + usage = Usage(tokens_in=1_000_000, tokens_out=0, tokens_cached=1_000_000) + assert compute_cost_cents(usage, RATE) == Decimal("10.000000") + + +def test_uncached_input_is_billed_at_the_full_rate() -> None: + usage = Usage(tokens_in=1_000_000, tokens_out=0, tokens_cached=0) + assert compute_cost_cents(usage, RATE) == Decimal("100.000000") + + +def test_mixed_usage_splits_cached_and_uncached_input() -> None: + # 3.7k input of which 3k cached, 800 output — the §7.2 Agent token budget. + usage = Usage(tokens_in=3700, tokens_out=800, tokens_cached=3000) + expected = ( + Decimal(700) * RATE.input_cents_per_mtok + + Decimal(3000) * RATE.cached_input_cents_per_mtok + + Decimal(800) * RATE.output_cents_per_mtok + ) / Decimal(1_000_000) + assert compute_cost_cents(usage, RATE) == expected.quantize(Decimal("0.000001")) + + +def test_zero_usage_costs_nothing() -> None: + assert compute_cost_cents(Usage(), RATE) == Decimal("0.000000") + + +def test_billable_input_never_goes_negative() -> None: + usage = Usage(tokens_in=100, tokens_cached=500) + assert usage.billable_input == 0 + + +def test_rate_table_falls_back_so_unknown_models_are_never_free() -> None: + default = ModelRate(Decimal("1"), Decimal("1"), Decimal("1")) + table = RateTable({"deepseek-v4-flash": RATE}, default=default) + assert table.for_model("deepseek-v4-flash") is RATE + assert table.for_model("some-new-model") is default + assert table.known_models() == ("deepseek-v4-flash",) + + +def _record(success: bool = True) -> AICallRecord: + return AICallRecord( + call_id=1, + scene="agent_chat", + provider="deepseek", + model="deepseek-v4-flash", + success=success, + usage=Usage(tokens_in=10, tokens_out=5, tokens_cached=2), + latency_ms=120, + cost_cents=Decimal("0.5"), + ) + + +async def test_with_defaults_stamps_created_at() -> None: + stamped = _record().with_defaults() + assert stamped.created_at is not None + assert stamped.with_defaults() is stamped + + +async def test_in_memory_recorder_collects_records() -> None: + recorder = InMemoryAICallRecorder() + await recorder.record(_record()) + assert len(recorder.records) == 1 + + +async def test_null_recorder_is_a_no_op() -> None: + assert await NullAICallRecorder().record(_record()) is None + + +async def test_composite_recorder_fans_out() -> None: + first, second = InMemoryAICallRecorder(), InMemoryAICallRecorder() + await CompositeAICallRecorder(first, second, PrometheusAICallRecorder()).record(_record()) + assert len(first.records) == len(second.records) == 1 diff --git a/tests/unit/test_daily_budget.py b/tests/unit/test_daily_budget.py new file mode 100755 index 0000000..436f7c3 --- /dev/null +++ b/tests/unit/test_daily_budget.py @@ -0,0 +1,84 @@ +"""Global daily cost circuit breaker (架构文档 §13 三道成本闸门). + +The requirement is enforcement, not alerting: once the platform-wide daily +spend crosses the threshold, non-critical AI calls are rejected. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from decimal import Decimal + +import pytest + +from packages.llm.budget import DailyBudgetGuard, InMemoryCostCounter +from packages.llm.errors import DailyBudgetExceededError + + +def make_guard( + budget: str = "100", *, enabled: bool = True, now: datetime | None = None +) -> tuple[DailyBudgetGuard, InMemoryCostCounter]: + counter = InMemoryCostCounter() + guard = DailyBudgetGuard( + counter, + daily_budget_cents=Decimal(budget), + enabled=enabled, + clock=(lambda: now) if now else None, + ) + return guard, counter + + +async def test_allows_calls_below_the_threshold() -> None: + guard, _ = make_guard("100") + await guard.add_cost(Decimal("99.9")) + await guard.ensure_within_budget(scene="agent_chat") + + +async def test_rejects_non_critical_scenes_once_the_budget_is_reached() -> None: + guard, _ = make_guard("100") + await guard.add_cost(Decimal("100")) + with pytest.raises(DailyBudgetExceededError) as excinfo: + await guard.ensure_within_budget(scene="agent_chat") + assert "agent_chat" in str(excinfo.value) + + +async def test_critical_scenes_bypass_the_breaker() -> None: + guard, _ = make_guard("100") + await guard.add_cost(Decimal("500")) + # Degrade the product, do not break it: 意图识别/越界拒答 must keep working. + await guard.ensure_within_budget(scene="intent_classify", critical=True) + + +async def test_disabled_guard_never_blocks_and_never_counts() -> None: + guard, counter = make_guard("1", enabled=False) + await guard.add_cost(Decimal("999")) + await guard.ensure_within_budget(scene="agent_chat") + assert await counter.get(guard.current_key()) == Decimal("0") + + +async def test_cost_accumulates_across_calls() -> None: + guard, _ = make_guard("100") + await guard.add_cost(Decimal("1.5")) + await guard.add_cost(Decimal("2.25")) + assert await guard.spent_today() == Decimal("3.75") + + +async def test_zero_cost_is_not_counted() -> None: + guard, _ = make_guard("100") + await guard.add_cost(Decimal("0")) + assert await guard.spent_today() == Decimal("0") + + +async def test_budget_day_boundary_follows_utc_plus_8() -> None: + """FR-BILL-01: 跨时区按 UTC+8 统一。17:00Z already belongs to the next day.""" + guard, _ = make_guard(now=datetime(2026, 8, 13, 17, 0, tzinfo=UTC)) + assert guard.current_key().endswith("2026-08-14") + + guard_before, _ = make_guard(now=datetime(2026, 8, 13, 15, 0, tzinfo=UTC)) + assert guard_before.current_key().endswith("2026-08-13") + + +async def test_budget_properties_are_exposed_for_dashboards() -> None: + guard, _ = make_guard("42") + assert guard.enabled is True + assert guard.budget_cents == Decimal("42") diff --git a/tests/unit/test_deepseek_provider.py b/tests/unit/test_deepseek_provider.py new file mode 100755 index 0000000..11a3c42 --- /dev/null +++ b/tests/unit/test_deepseek_provider.py @@ -0,0 +1,162 @@ +"""DeepSeek provider against a mocked transport. No network access.""" + +from __future__ import annotations + +import json +from typing import Any + +import httpx +import pytest + +from packages.llm.errors import ( + ProviderAuthError, + ProviderBadRequestError, + ProviderRateLimitedError, + ProviderServerError, + ProviderTimeoutError, +) +from packages.llm.providers.deepseek import DeepSeekProvider +from packages.llm.types import Message, ProviderRequest + +CHAT_RESPONSE: dict[str, Any] = { + "id": "chatcmpl-1", + "model": "deepseek-v4-flash", + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": "白衬衫配西裤"}, + "finish_reason": "stop", + } + ], + "usage": { + "prompt_tokens": 1200, + "completion_tokens": 300, + "prompt_cache_hit_tokens": 1000, + "prompt_cache_miss_tokens": 200, + }, +} + + +def make_provider(handler: Any) -> DeepSeekProvider: + return DeepSeekProvider( + api_key="test-key", + client=httpx.AsyncClient(transport=httpx.MockTransport(handler)), + ) + + +def request(json_mode: bool = False) -> ProviderRequest: + return ProviderRequest( + model="deepseek-v4-flash", + messages=(Message(role="system", content="前缀"), Message(role="user", content="问题")), + temperature=0.7, + max_tokens=800, + timeout_seconds=10.0, + json_mode=json_mode, + ) + + +async def test_successful_completion_maps_usage_including_cache_hits() -> None: + provider = make_provider(lambda _r: httpx.Response(200, json=CHAT_RESPONSE)) + result = await provider.complete(request()) + + assert result.text == "白衬衫配西裤" + assert result.model == "deepseek-v4-flash" + assert result.usage.tokens_in == 1200 + assert result.usage.tokens_out == 300 + assert result.usage.tokens_cached == 1000 + assert result.usage.billable_input == 200 + await provider.aclose() + + +async def test_openai_style_cached_tokens_are_also_understood() -> None: + payload = { + **CHAT_RESPONSE, + "usage": { + "prompt_tokens": 500, + "completion_tokens": 10, + "prompt_tokens_details": {"cached_tokens": 400}, + }, + } + provider = make_provider(lambda _r: httpx.Response(200, json=payload)) + result = await provider.complete(request()) + assert result.usage.tokens_cached == 400 + + +async def test_request_shape_is_openai_compatible() -> None: + captured: dict[str, Any] = {} + + def handler(req: httpx.Request) -> httpx.Response: + captured["url"] = str(req.url) + captured["auth"] = req.headers["authorization"] + captured["body"] = json.loads(req.content) + return httpx.Response(200, json=CHAT_RESPONSE) + + provider = make_provider(handler) + await provider.complete(request(json_mode=True)) + + assert captured["url"] == "https://api.deepseek.com/v1/chat/completions" + assert captured["auth"] == "Bearer test-key" + assert captured["body"]["model"] == "deepseek-v4-flash" + assert captured["body"]["response_format"] == {"type": "json_object"} + assert [m["role"] for m in captured["body"]["messages"]] == ["system", "user"] + + +@pytest.mark.parametrize( + ("status", "expected"), + [ + (429, ProviderRateLimitedError), + (401, ProviderAuthError), + (403, ProviderAuthError), + (500, ProviderServerError), + (503, ProviderServerError), + (400, ProviderBadRequestError), + (404, ProviderBadRequestError), + ], +) +async def test_http_errors_map_to_the_taxonomy(status: int, expected: type[Exception]) -> None: + provider = make_provider(lambda _r: httpx.Response(status, json={"error": "boom"})) + with pytest.raises(expected): + await provider.complete(request()) + + +async def test_only_4xx_bad_requests_are_non_retryable() -> None: + provider = make_provider(lambda _r: httpx.Response(400, json={"error": "bad"})) + with pytest.raises(ProviderBadRequestError) as excinfo: + await provider.complete(request()) + assert excinfo.value.retryable is False + + +async def test_timeouts_become_provider_timeout_errors() -> None: + def handler(req: httpx.Request) -> httpx.Response: + raise httpx.ReadTimeout("too slow", request=req) + + provider = make_provider(handler) + with pytest.raises(ProviderTimeoutError): + await provider.complete(request()) + + +async def test_transport_errors_are_retryable_server_errors() -> None: + def handler(req: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("dns failure", request=req) + + provider = make_provider(handler) + with pytest.raises(ProviderServerError) as excinfo: + await provider.complete(request()) + assert excinfo.value.retryable is True + + +async def test_malformed_response_body_is_a_server_error() -> None: + provider = make_provider(lambda _r: httpx.Response(200, json={"unexpected": True})) + with pytest.raises(ProviderServerError): + await provider.complete(request()) + + +async def test_zhipu_placeholder_explains_how_to_enable_it() -> None: + from packages.llm.providers.zhipu import ZhipuGLMProvider + + provider = ZhipuGLMProvider() + assert provider.name == "zhipu" + assert provider.default_model == "glm-4-flash" + with pytest.raises(NotImplementedError, match="备案"): + await provider.complete(request()) + await provider.aclose() diff --git a/tests/unit/test_gateway.py b/tests/unit/test_gateway.py new file mode 100755 index 0000000..5124ecd --- /dev/null +++ b/tests/unit/test_gateway.py @@ -0,0 +1,469 @@ +"""LLM gateway: routing, failover, circuit breaking, structured output, cost. + +No test here touches the network — providers are fakes. +""" + +from __future__ import annotations + +import itertools +from collections.abc import Sequence +from decimal import Decimal +from pathlib import Path + +import pytest +from pydantic import BaseModel + +from packages.llm.budget import DailyBudgetGuard, InMemoryCostCounter +from packages.llm.circuit import CircuitBreakerRegistry, CircuitState +from packages.llm.cost import InMemoryAICallRecorder, ModelRate, RateTable +from packages.llm.errors import ( + AllProvidersFailedError, + DailyBudgetExceededError, + ProviderBadRequestError, + ProviderRateLimitedError, + ProviderServerError, + ProviderTimeoutError, + SceneNotRoutedError, + StructuredOutputError, +) +from packages.llm.gateway import LLMGateway, extract_json +from packages.llm.prompts import PromptRegistry +from packages.llm.providers.zhipu import ZhipuGLMProvider +from packages.llm.routing import SceneRouter, build_scene_configs +from packages.llm.types import Message, ProviderRequest, ProviderResponse, Scene, Usage + +RATE = ModelRate( + input_cents_per_mtok=Decimal("100"), + cached_input_cents_per_mtok=Decimal("10"), + output_cents_per_mtok=Decimal("200"), +) + + +class FakeProvider: + """Returns queued responses/exceptions in order; repeats the last forever.""" + + def __init__( + self, + name: str, + outcomes: Sequence[ProviderResponse | Exception], + default_model: str = "fake-model", + ) -> None: + self._name = name + self._outcomes = list(outcomes) + self._default_model = default_model + self.requests: list[ProviderRequest] = [] + self.closed = False + + @property + def name(self) -> str: + return self._name + + @property + def default_model(self) -> str: + return self._default_model + + async def complete(self, request: ProviderRequest) -> ProviderResponse: + self.requests.append(request) + outcome = self._outcomes[min(len(self.requests) - 1, len(self._outcomes) - 1)] + if isinstance(outcome, Exception): + raise outcome + return outcome + + async def aclose(self) -> None: + self.closed = True + + +def response(text: str, model: str = "fake-model") -> ProviderResponse: + return ProviderResponse( + text=text, + model=model, + usage=Usage(tokens_in=1000, tokens_out=200, tokens_cached=800), + finish_reason="stop", + ) + + +def build_gateway( + prompts_dir: Path, + providers: dict[str, FakeProvider], + *, + primary: str = "p1", + backup: str | None = "p2", + budget_cents: str = "100000", + budget_enabled: bool = True, + failure_threshold: int = 3, + structured_max_retries: int = 2, + recorder: InMemoryAICallRecorder | None = None, +) -> tuple[LLMGateway, InMemoryAICallRecorder, DailyBudgetGuard]: + recorder = recorder or InMemoryAICallRecorder() + guard = DailyBudgetGuard( + InMemoryCostCounter(), + daily_budget_cents=Decimal(budget_cents), + enabled=budget_enabled, + ) + gateway = LLMGateway( + providers=dict(providers), + router=SceneRouter( + build_scene_configs( + primary_provider=primary, + backup_provider=backup, + structured_max_retries=structured_max_retries, + ) + ), + prompts=PromptRegistry(prompts_dir), + rates=RateTable({"fake-model": RATE}, default=RATE), + recorder=recorder, + budget=guard, + breakers=CircuitBreakerRegistry(failure_threshold=failure_threshold), + id_factory=itertools.count(1).__next__, + ) + return gateway, recorder, guard + + +# -------------------------------------------------------------------------- +# routing +# -------------------------------------------------------------------------- + + +async def test_routes_to_the_primary_provider(prompts_dir: Path) -> None: + primary = FakeProvider("p1", [response("好的")]) + backup = FakeProvider("p2", [response("backup")]) + gateway, _, _ = build_gateway(prompts_dir, {"p1": primary, "p2": backup}) + + result = await gateway.complete(scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "面试"}) + + assert result.provider == "p1" + assert result.text == "好的" + assert backup.requests == [] + + +async def test_scene_config_drives_model_and_sampling(prompts_dir: Path) -> None: + primary = FakeProvider("p1", [response("ok")], default_model="deepseek-v4-flash") + gateway, _, _ = build_gateway(prompts_dir, {"p1": primary}, backup=None) + + await gateway.complete(scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "hi"}) + + request = primary.requests[0] + assert request.model == "deepseek-v4-flash" + assert request.max_tokens == 800 # 架构文档 §7.2 输出预算 + assert request.timeout_seconds == 30.0 + + +async def test_cacheable_system_prefix_is_the_first_message(prompts_dir: Path) -> None: + primary = FakeProvider("p1", [response("ok")]) + gateway, _, _ = build_gateway(prompts_dir, {"p1": primary}, backup=None) + + await gateway.complete( + scene=Scene.AGENT_CHAT, + messages=[Message(role="user", content="历史消息")], + prompt_vars={"user_input": "现在的问题"}, + ) + + messages = primary.requests[0].messages + assert messages[0].role == "system" + assert "穿搭顾问助手" in messages[0].content + assert messages[-1].content == "现在的问题" + + +async def test_unrouted_scene_raises(prompts_dir: Path) -> None: + gateway = LLMGateway( + providers={}, + router=SceneRouter({}), + prompts=PromptRegistry(prompts_dir), + rates=RateTable({}, default=RATE), + recorder=InMemoryAICallRecorder(), + budget=DailyBudgetGuard( + InMemoryCostCounter(), daily_budget_cents=Decimal("1"), enabled=False + ), + ) + with pytest.raises(SceneNotRoutedError): + await gateway.complete(scene=Scene.AGENT_CHAT) + + +async def test_unregistered_provider_is_reported_as_a_failure(prompts_dir: Path) -> None: + gateway, _, _ = build_gateway(prompts_dir, {}, primary="ghost", backup=None) + with pytest.raises(AllProvidersFailedError): + await gateway.complete(scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "hi"}) + + +# -------------------------------------------------------------------------- +# failover +# -------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "failure", + [ + ProviderTimeoutError("p1", "timed out"), + ProviderServerError("p1", "502", status_code=502), + ProviderRateLimitedError("p1", "429", status_code=429), + ], +) +async def test_timeout_5xx_and_rate_limit_fail_over_to_the_backup( + prompts_dir: Path, failure: Exception +) -> None: + primary = FakeProvider("p1", [failure]) + backup = FakeProvider("p2", [response("备用回答")]) + gateway, recorder, _ = build_gateway(prompts_dir, {"p1": primary, "p2": backup}) + + result = await gateway.complete(scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "hi"}) + + assert result.provider == "p2" + assert result.text == "备用回答" + # Both the failed and the successful attempt are attributable. + assert [(r.provider, r.success) for r in recorder.records] == [("p1", False), ("p2", True)] + + +async def test_bad_request_does_not_fail_over(prompts_dir: Path) -> None: + """A malformed request would fail on the backup too — burning money twice.""" + primary = FakeProvider("p1", [ProviderBadRequestError("p1", "400", status_code=400)]) + backup = FakeProvider("p2", [response("should not be used")]) + gateway, _, _ = build_gateway(prompts_dir, {"p1": primary, "p2": backup}) + + with pytest.raises(ProviderBadRequestError): + await gateway.complete(scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "hi"}) + assert backup.requests == [] + + +async def test_all_providers_failing_raises_with_every_cause(prompts_dir: Path) -> None: + primary = FakeProvider("p1", [ProviderTimeoutError("p1", "timeout")]) + backup = FakeProvider("p2", [ProviderServerError("p2", "500", status_code=500)]) + gateway, _, _ = build_gateway(prompts_dir, {"p1": primary, "p2": backup}) + + with pytest.raises(AllProvidersFailedError) as excinfo: + await gateway.complete(scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "hi"}) + assert len(excinfo.value.failures) == 2 + + +async def test_zhipu_placeholder_participates_in_the_real_routing_path( + prompts_dir: Path, +) -> None: + """The backup slot is structural, but the code path through it is real.""" + primary = FakeProvider("p1", [ProviderTimeoutError("p1", "timeout")]) + gateway, recorder, _ = build_gateway( + prompts_dir, {"p1": primary, "zhipu": ZhipuGLMProvider()}, backup="zhipu" + ) + + with pytest.raises(AllProvidersFailedError) as excinfo: + await gateway.complete(scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "hi"}) + + assert isinstance(excinfo.value.failures[1], NotImplementedError) + assert {r.provider for r in recorder.records} == {"p1", "zhipu"} + + +# -------------------------------------------------------------------------- +# circuit breaker +# -------------------------------------------------------------------------- + + +async def test_open_circuit_skips_the_primary_entirely(prompts_dir: Path) -> None: + primary = FakeProvider("p1", [ProviderServerError("p1", "500", status_code=500)]) + backup = FakeProvider("p2", [response("备用")]) + gateway, _, _ = build_gateway(prompts_dir, {"p1": primary, "p2": backup}, failure_threshold=1) + + first = await gateway.complete(scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "a"}) + assert first.provider == "p2" + assert gateway.breaker_states()["p1"] == CircuitState.OPEN.value + + second = await gateway.complete(scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "b"}) + assert second.provider == "p2" + assert len(primary.requests) == 1 # not attempted the second time + + +async def test_success_keeps_the_circuit_closed(prompts_dir: Path) -> None: + primary = FakeProvider("p1", [response("ok")]) + gateway, _, _ = build_gateway(prompts_dir, {"p1": primary}, backup=None) + await gateway.complete(scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "a"}) + assert gateway.breaker_states()["p1"] == CircuitState.CLOSED.value + + +# -------------------------------------------------------------------------- +# structured output +# -------------------------------------------------------------------------- + + +class Outfit(BaseModel): + top: str + bottom: str + + +async def test_structured_output_is_parsed_and_validated(prompts_dir: Path) -> None: + primary = FakeProvider("p1", [response('{"top": "白衬衫", "bottom": "西裤"}')]) + gateway, _, _ = build_gateway(prompts_dir, {"p1": primary}, backup=None) + + result = await gateway.complete( + scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "hi"}, schema=Outfit + ) + + assert result.parsed_as(Outfit).top == "白衬衫" + assert result.attempts == 1 + assert primary.requests[0].json_mode is True + assert "JSON Schema" in primary.requests[0].messages[-1].content + + +async def test_invalid_structured_output_is_retried_with_the_validation_error( + prompts_dir: Path, +) -> None: + primary = FakeProvider( + "p1", + [ + response("这不是 JSON"), + response('{"top": "针织衫"}'), # missing `bottom` + response('{"top": "针织衫", "bottom": "阔腿裤"}'), + ], + ) + gateway, recorder, _ = build_gateway(prompts_dir, {"p1": primary}, backup=None) + + result = await gateway.complete( + scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "hi"}, schema=Outfit + ) + + assert result.attempts == 3 + assert result.parsed_as(Outfit).bottom == "阔腿裤" + # Every attempt costs money and every attempt is recorded. + assert len(recorder.records) == 3 + assert [r.success for r in recorder.records] == [False, False, True] + assert recorder.records[0].error_code == "llm_structured_output_invalid" + # The repair turn carries the validation error back to the model. + assert "校验错误" in primary.requests[1].messages[-1].content + + +async def test_structured_output_gives_up_and_fails_over(prompts_dir: Path) -> None: + primary = FakeProvider("p1", [response("never json")]) + backup = FakeProvider("p2", [response('{"top": "T恤", "bottom": "牛仔裤"}')]) + gateway, _, _ = build_gateway( + prompts_dir, {"p1": primary, "p2": backup}, structured_max_retries=1 + ) + + result = await gateway.complete( + scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "hi"}, schema=Outfit + ) + + assert result.provider == "p2" + assert len(primary.requests) == 2 # 1 + structured_max_retries + + +async def test_structured_failure_on_the_only_provider_surfaces(prompts_dir: Path) -> None: + primary = FakeProvider("p1", [response("never json")]) + gateway, _, _ = build_gateway( + prompts_dir, {"p1": primary}, backup=None, structured_max_retries=0 + ) + with pytest.raises(AllProvidersFailedError) as excinfo: + await gateway.complete( + scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "hi"}, schema=Outfit + ) + assert isinstance(excinfo.value.failures[0], StructuredOutputError) + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + ('{"a": 1}', '{"a": 1}'), + ('```json\n{"a": 1}\n```', '{"a": 1}'), + ('好的：\n{"a": 1}\n以上。', '{"a": 1}'), + ("no json here", "no json here"), + ], +) +def test_extract_json_handles_fenced_and_chatty_replies(raw: str, expected: str) -> None: + assert extract_json(raw) == expected + + +# -------------------------------------------------------------------------- +# cost attribution + daily budget +# -------------------------------------------------------------------------- + + +async def test_every_call_records_a_cost_row(prompts_dir: Path) -> None: + primary = FakeProvider("p1", [response("ok")]) + gateway, recorder, _ = build_gateway(prompts_dir, {"p1": primary}, backup=None) + + result = await gateway.complete( + scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "hi"}, user_id=42, ref_id="req-1" + ) + + record = recorder.records[0] + assert record.scene == "agent_chat" + assert record.provider == "p1" + assert record.model == "fake-model" + assert record.prompt_id == "agent_chat" + assert record.prompt_version == "v1" # 可复现: 记录 prompt 版本 + assert record.user_id == 42 + assert record.ref_id == "req-1" + assert record.usage.tokens_cached == 800 + assert record.cost_cents == result.cost_cents > Decimal("0") + assert record.created_at is not None + + +async def test_failed_calls_are_recorded_too(prompts_dir: Path) -> None: + primary = FakeProvider("p1", [ProviderTimeoutError("p1", "timed out")]) + gateway, recorder, _ = build_gateway(prompts_dir, {"p1": primary}, backup=None) + + with pytest.raises(AllProvidersFailedError): + await gateway.complete(scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "hi"}) + + record = recorder.records[0] + assert record.success is False + assert record.error_code == "llm_timeout" + assert record.cost_cents == Decimal("0") + + +async def test_a_broken_recorder_never_breaks_the_user_request(prompts_dir: Path) -> None: + class ExplodingRecorder: + async def record(self, record: object) -> None: + msg = "database is down" + raise RuntimeError(msg) + + primary = FakeProvider("p1", [response("ok")]) + gateway, _, _ = build_gateway(prompts_dir, {"p1": primary}, backup=None) + gateway._recorder = ExplodingRecorder() + + result = await gateway.complete(scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "hi"}) + assert result.text == "ok" + + +async def test_spend_accumulates_into_the_daily_budget(prompts_dir: Path) -> None: + primary = FakeProvider("p1", [response("ok")]) + gateway, _, guard = build_gateway(prompts_dir, {"p1": primary}, backup=None) + + first = await gateway.complete(scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "a"}) + await gateway.complete(scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "b"}) + + assert await guard.spent_today() == first.cost_cents * 2 + + +async def test_daily_budget_rejects_non_critical_scenes_before_spending( + prompts_dir: Path, +) -> None: + primary = FakeProvider("p1", [response("ok")]) + gateway, _, guard = build_gateway( + prompts_dir, {"p1": primary}, backup=None, budget_cents="0.00001" + ) + await guard.add_cost(Decimal("1")) + + with pytest.raises(DailyBudgetExceededError): + await gateway.complete(scene=Scene.AGENT_CHAT, prompt_vars={"user_input": "hi"}) + assert primary.requests == [] # rejected, not merely alerted + + +async def test_daily_budget_lets_critical_scenes_through(prompts_dir: Path, tmp_path: Path) -> None: + (prompts_dir / "intent_classify").mkdir() + (prompts_dir / "intent_classify" / "v1.txt").write_text( + "[system]\n你是意图分类器。\n\n[user]\n{user_input}\n", encoding="utf-8" + ) + primary = FakeProvider("p1", [response("scene_consult")]) + gateway, _, guard = build_gateway( + prompts_dir, {"p1": primary}, backup=None, budget_cents="0.00001" + ) + await guard.add_cost(Decimal("1")) + + result = await gateway.complete( + scene=Scene.INTENT_CLASSIFY, prompt_vars={"user_input": "面试穿什么"} + ) + assert result.text == "scene_consult" + + +async def test_aclose_closes_every_provider(prompts_dir: Path) -> None: + primary = FakeProvider("p1", [response("ok")]) + backup = FakeProvider("p2", [response("ok")]) + gateway, _, _ = build_gateway(prompts_dir, {"p1": primary, "p2": backup}) + await gateway.aclose() + assert primary.closed and backup.closed diff --git a/tests/unit/test_ids.py b/tests/unit/test_ids.py new file mode 100755 index 0000000..490224f --- /dev/null +++ b/tests/unit/test_ids.py @@ -0,0 +1,55 @@ +"""Snowflake ids (架构文档 §6.1 原则 2).""" + +from __future__ import annotations + +import pytest + +from packages.domain.ids import ( + DEFAULT_EPOCH_MS, + SnowflakeGenerator, + next_id, + timestamp_ms_of, +) + + +def test_ids_are_unique_and_monotonic() -> None: + generator = SnowflakeGenerator(node_id=1) + ids = [generator.next_id() for _ in range(5000)] + assert len(set(ids)) == len(ids) + assert ids == sorted(ids) + + +def test_ids_fit_in_a_signed_64_bit_integer() -> None: + assert 0 < next_id() < 2**63 + + +def test_different_nodes_do_not_collide() -> None: + a = SnowflakeGenerator(node_id=1) + b = SnowflakeGenerator(node_id=2) + assert not {a.next_id() for _ in range(500)} & {b.next_id() for _ in range(500)} + + +def test_node_id_is_bounded() -> None: + with pytest.raises(ValueError, match="node_id"): + SnowflakeGenerator(node_id=1024) + + +def test_timestamp_is_recoverable_for_partition_pruning() -> None: + generator = SnowflakeGenerator(node_id=3) + recovered = timestamp_ms_of(generator.next_id()) + assert recovered > DEFAULT_EPOCH_MS + + +def test_sequence_rollover_within_one_millisecond() -> None: + """More than 4096 ids in a single ms must wait, not duplicate.""" + frozen = DEFAULT_EPOCH_MS + 1_000 + calls = {"n": 0} + + def fake_now() -> int: + calls["n"] += 1 + # Hold the clock for the first 5000 calls, then let it advance. + return frozen if calls["n"] < 5000 else frozen + 1 + + generator = SnowflakeGenerator(node_id=1, clock=fake_now) + ids = [generator.next_id() for _ in range(4200)] + assert len(set(ids)) == len(ids) diff --git a/tests/unit/test_log_redaction.py b/tests/unit/test_log_redaction.py new file mode 100755 index 0000000..132b2c8 --- /dev/null +++ b/tests/unit/test_log_redaction.py @@ -0,0 +1,75 @@ +"""Log redaction (NFR-SEC-03, 架构文档 §11.2). + +日志中禁止出现手机号、UnionID、令牌、图片地址与诊断结论。 +""" + +from __future__ import annotations + +from packages.shared.logging import REDACTED, make_redaction_processor + + +def redact(event: dict[str, object], extra: tuple[str, ...] = ()) -> dict[str, object]: + return dict(make_redaction_processor(extra)(None, "info", dict(event))) + + +def test_sensitive_top_level_keys_are_masked() -> None: + out = redact( + { + "event": "login", + "phone": "13800138000", + "union_id": "u-123", + "access_token": "abc.def.ghi", + "avatar_key": "user/1/avatar.png", + "user_id": 42, + } + ) + assert out["phone"] == REDACTED + assert out["union_id"] == REDACTED + assert out["access_token"] == REDACTED + assert out["avatar_key"] == REDACTED + assert out["user_id"] == 42 # internal id is not sensitive on its own + + +def test_nested_structures_are_redacted() -> None: + out = redact({"event": "x", "payload": {"inner": {"phone": "13800138000", "ok": 1}}}) + payload = out["payload"] + assert isinstance(payload, dict) + assert payload["inner"]["phone"] == REDACTED + assert payload["inner"]["ok"] == 1 + + +def test_phone_numbers_inside_free_text_are_masked() -> None: + out = redact({"event": "sms", "detail": "验证码已发送至 13800138000"}) + assert "13800138000" not in str(out["detail"]) + assert "138****8000" in str(out["detail"]) + + +def test_api_keys_in_free_text_are_masked() -> None: + # Never put a real key in a fixture, not even to prove it gets redacted: + # the test file itself would then carry the secret into git history. + fake_key = "sk-" + "0f1e2d3c4b5a69788796a5b4c3d2e1f0" + out = redact({"event": "cfg", "detail": f"using {fake_key}"}) + assert fake_key not in str(out["detail"]) + assert REDACTED in str(out["detail"]) + + +def test_bearer_tokens_in_free_text_are_masked() -> None: + out = redact({"event": "req", "detail": "Authorization: Bearer eyJhbGciOiJIUzI1NiJ9.x.y"}) + assert "eyJhbGciOiJIUzI1NiJ9" not in str(out["detail"]) + + +def test_diagnosis_output_is_treated_as_sensitive() -> None: + out = redact({"event": "report", "dimensions": {"color_season": "浅暖柔和型"}}) + assert out["dimensions"] == REDACTED + + +def test_extra_keys_can_be_configured() -> None: + out = redact({"event": "x", "custom_field": "secret"}, extra=("custom_field",)) + assert out["custom_field"] == REDACTED + + +def test_lists_are_traversed() -> None: + out = redact({"event": "x", "items": [{"phone": "13800138000"}]}) + items = out["items"] + assert isinstance(items, list) + assert items[0]["phone"] == REDACTED diff --git a/tests/unit/test_prompt_cache_prefix.py b/tests/unit/test_prompt_cache_prefix.py new file mode 100755 index 0000000..b944dd6 --- /dev/null +++ b/tests/unit/test_prompt_cache_prefix.py @@ -0,0 +1,89 @@ +"""Prompt-cache prefix discipline (架构文档 §7.2 缓存前缀纪律, TD-04). + +A volatile token in the fixed prefix cannot be caught by functional tests: the +answer still looks right, the bill just quietly multiplies. So it gets its own +mechanical check. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from packages.llm.errors import CachePrefixViolationError, PromptNotFoundError +from packages.llm.prompts import ( + PromptRegistry, + find_volatile_tokens, + parse_prompt_file, + validate_cache_prefix, +) + +CLEAN_PREFIX = "你是「美搭」的穿搭顾问助手。只回答穿搭、色彩与风格相关的问题。" + + +def test_clean_prefix_has_no_violations() -> None: + assert find_volatile_tokens(CLEAN_PREFIX) == [] + assert validate_cache_prefix(CLEAN_PREFIX, mode="strict") == [] + + +@pytest.mark.parametrize( + ("prefix", "expected_kind"), + [ + ("今天是 2026-08-13，请为用户搭配。", "iso-date"), + ("当前时间 09:31:00，请回答。", "iso-time"), + ("现在是 2026 年 8 月。", "cn-date"), + ("时间戳 1755012345 供参考。", "epoch-seconds"), + ("会话 550e8400-e29b-41d4-a716-446655440000 开始。", "uuid"), + ("追踪号 9f8c1d2e3b4a5f6071829304 记录在案。", "long-hex-id"), + ("你好 {nickname}，我来帮你搭配。", "volatile-placeholder"), + ("用户编号 {user_id} 的偏好如下。", "volatile-placeholder"), + ("当前时刻 {now}。", "volatile-placeholder"), + ], +) +def test_volatile_tokens_are_detected(prefix: str, expected_kind: str) -> None: + kinds = {v.kind for v in find_volatile_tokens(prefix)} + assert expected_kind in kinds + + +def test_strict_mode_raises_with_the_offending_samples() -> None: + with pytest.raises(CachePrefixViolationError) as excinfo: + validate_cache_prefix("生成于 2026-08-13 的提示词", mode="strict", ref="p@v1") + assert "p@v1" in str(excinfo.value) + assert excinfo.value.violations + + +def test_warn_mode_reports_without_raising() -> None: + violations = validate_cache_prefix("生成于 2026-08-13", mode="warn") + assert [v.kind for v in violations] == ["iso-date"] + + +def test_off_mode_skips_the_check() -> None: + assert validate_cache_prefix("生成于 2026-08-13", mode="off") == [] + + +def test_registry_rejects_a_bad_prompt_at_load_time(tmp_path: Path) -> None: + bad = tmp_path / "bad" / "v1.txt" + bad.parent.mkdir(parents=True) + bad.write_text("[system]\n你好 {nickname}\n\n[user]\n{user_input}\n", encoding="utf-8") + + registry = PromptRegistry(tmp_path, cache_prefix_mode="strict") + with pytest.raises(CachePrefixViolationError): + registry.get("bad", "v1") + + +def test_shipped_prompts_all_satisfy_the_discipline(repo_root: Path) -> None: + """Regression guard: every prompt in the repo keeps a cacheable prefix.""" + prompts_root = repo_root / "prompts" + files = sorted(prompts_root.glob("*/*.txt")) + assert files, "no prompt templates found" + for path in files: + template = parse_prompt_file( + path.read_text(encoding="utf-8"), prompt_id=path.parent.name, version=path.stem + ) + assert find_volatile_tokens(template.system_prefix) == [], f"{path} has a volatile prefix" + + +def test_prompt_file_without_system_section_is_rejected() -> None: + with pytest.raises(PromptNotFoundError): + parse_prompt_file("[user]\nhello\n", prompt_id="x", version="v1") diff --git a/tests/unit/test_prompt_registry.py b/tests/unit/test_prompt_registry.py new file mode 100755 index 0000000..f7985d3 --- /dev/null +++ b/tests/unit/test_prompt_registry.py @@ -0,0 +1,58 @@ +"""Prompt versioning (架构文档 §8.1 Prompt 版本).""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from packages.llm.errors import PromptNotFoundError +from packages.llm.prompts import PromptRegistry, parse_prompt_file + + +def test_loads_by_id_and_version(prompts_dir: Path) -> None: + registry = PromptRegistry(prompts_dir) + v1 = registry.get("agent_chat", "v1") + v2 = registry.get("agent_chat", "v2") + + assert v1.prompt_id == "agent_chat" + assert v1.version == "v1" + assert v1.ref == "agent_chat@v1" + assert v1.system_prefix != v2.system_prefix + + +def test_templates_are_cached_after_first_load(prompts_dir: Path) -> None: + registry = PromptRegistry(prompts_dir) + assert registry.get("agent_chat", "v1") is registry.get("agent_chat", "v1") + + +def test_missing_prompt_raises(prompts_dir: Path) -> None: + registry = PromptRegistry(prompts_dir) + with pytest.raises(PromptNotFoundError): + registry.get("agent_chat", "v99") + + +def test_available_versions(prompts_dir: Path) -> None: + registry = PromptRegistry(prompts_dir) + assert registry.available_versions("agent_chat") == ["v1", "v2"] + assert registry.available_versions("nope") == [] + + +def test_user_template_rendering() -> None: + template = parse_prompt_file( + "[system]\n固定前缀\n\n[user]\n用户说：{user_input}\n", prompt_id="x", version="v1" + ) + assert template.render_user({"user_input": "面试穿什么"}) == "用户说：面试穿什么" + + +def test_missing_variable_is_reported_with_the_prompt_ref() -> None: + template = parse_prompt_file( + "[system]\n固定前缀\n\n[user]\n{user_input}\n", prompt_id="x", version="v1" + ) + with pytest.raises(PromptNotFoundError, match="user_input"): + template.render_user({}) + + +def test_prompt_without_user_section_renders_empty() -> None: + template = parse_prompt_file("[system]\n只有系统前缀\n", prompt_id="x", version="v1") + assert template.render_user({"anything": "1"}) == "" diff --git a/tests/unit/test_quota_domain.py b/tests/unit/test_quota_domain.py new file mode 100755 index 0000000..03a735b --- /dev/null +++ b/tests/unit/test_quota_domain.py @@ -0,0 +1,111 @@ +"""Quota three-phase arithmetic (架构文档 §7.4, FR-BILL-01/05/06).""" + +from __future__ import annotations + +from datetime import UTC, datetime + +import pytest + +from packages.domain.billing import quota +from packages.domain.errors import QuotaExceededError + + +def test_period_key_day_uses_utc_plus_8() -> None: + # 17:00Z is already 01:00 the next day in UTC+8. + assert quota.period_key(datetime(2026, 8, 13, 17, 0, tzinfo=UTC), "day") == "d:2026-08-14" + assert quota.period_key(datetime(2026, 8, 13, 15, 0, tzinfo=UTC), "day") == "d:2026-08-13" + + +def test_period_key_month_and_lifetime() -> None: + now = datetime(2026, 8, 13, 1, 0, tzinfo=UTC) + assert quota.period_key(now, "month") == "m:2026-08" + assert quota.period_key(now, "lifetime") == "life" + + +def test_unknown_period_type_is_rejected() -> None: + with pytest.raises(ValueError, match="unknown period_type"): + quota.period_key(datetime.now(UTC), "weekly") # type: ignore[arg-type] + + +def test_period_reset_at_rolls_to_the_next_local_midnight() -> None: + reset = quota.period_reset_at(datetime(2026, 8, 13, 1, 0, tzinfo=UTC), "day") + assert reset is not None + assert reset == datetime(2026, 8, 13, 16, 0, tzinfo=UTC) # 2026-08-14 00:00 +08 + + +def test_period_reset_at_handles_december_rollover() -> None: + reset = quota.period_reset_at(datetime(2026, 12, 20, 1, 0, tzinfo=UTC), "month") + assert reset is not None + assert reset.year == 2026 # 2027-01-01 00:00 +08 == 2026-12-31 16:00 UTC + assert reset.month == 12 + + +def test_lifetime_quota_never_resets() -> None: + assert quota.period_reset_at(datetime.now(UTC), "lifetime") is None + + +def test_task_bonus_raises_the_effective_limit() -> None: + state = quota.QuotaState(used=20, base_limit=20, bonus=3) + assert state.effective_limit == 23 + assert state.remaining == 3 + assert quota.evaluate(state, 1).allowed + + +def test_reserve_freezes_the_amount() -> None: + state = quota.reserve(quota.QuotaState(used=0, base_limit=20), 1) + assert state.used == 1 + assert state.remaining == 19 + + +def test_reserve_beyond_the_limit_is_rejected_server_side() -> None: + state = quota.QuotaState(used=20, base_limit=20) + with pytest.raises(QuotaExceededError) as excinfo: + quota.reserve(state, 1) + assert excinfo.value.remaining == 0 + assert excinfo.value.required == 1 + + +def test_reserve_rejects_a_batch_that_would_overshoot() -> None: + with pytest.raises(QuotaExceededError): + quota.reserve(quota.QuotaState(used=18, base_limit=20), 3) + + +def test_non_positive_amounts_are_programming_errors() -> None: + with pytest.raises(ValueError, match="positive"): + quota.evaluate(quota.QuotaState(used=0, base_limit=1), 0) + with pytest.raises(ValueError, match="positive"): + quota.refund(quota.QuotaState(used=1, base_limit=1), -1) + + +def test_refund_releases_the_reservation() -> None: + reserved = quota.reserve(quota.QuotaState(used=5, base_limit=20), 1) + assert quota.refund(reserved, 1).used == 5 + + +def test_refund_never_drives_usage_negative() -> None: + assert quota.refund(quota.QuotaState(used=0, base_limit=20), 5).used == 0 + + +def test_balance_after_records_remaining_quota() -> None: + assert quota.balance_after(quota.QuotaState(used=3, base_limit=20)) == 17 + + +def test_refund_key_is_derived_so_a_retry_cannot_double_refund() -> None: + reserve_key = quota.reserve_idempotency_key( + user_id=7, quota_key="agent_chat", request_key="req-1" + ) + assert reserve_key == "rsv:7:agent_chat:req-1" + assert quota.refund_idempotency_key(reserve_key) == "rfd:rsv:7:agent_chat:req-1" + assert quota.refund_idempotency_key(reserve_key) == quota.refund_idempotency_key(reserve_key) + + +def test_ledger_can_recompute_the_balance() -> None: + """流水是真相，余额是缓存 (架构文档 §6.1 原则 4).""" + state = quota.QuotaState(used=0, base_limit=20) + deltas = [] + for _ in range(3): + state = quota.reserve(state, 1) + deltas.append(-1) + state = quota.refund(state, 1) + deltas.append(1) + assert state.used == -sum(deltas) diff --git a/tests/unit/test_response_envelope.py b/tests/unit/test_response_envelope.py new file mode 100755 index 0000000..74c17fc --- /dev/null +++ b/tests/unit/test_response_envelope.py @@ -0,0 +1,96 @@ +"""Unified response and error codes (架构文档 §9.2, FR-SYS-02).""" + +from __future__ import annotations + +from packages.shared.errors import ( + ERROR_META, + AppError, + ErrorCode, + QuotaExhaustedError, +) +from packages.shared.response import fail, ok +from packages.shared.security import ( + AccessTokenClaims, + create_access_token, + decode_access_token, +) + + +def test_success_envelope_shape() -> None: + body = ok({"a": 1}) + assert body["code"] == 0 + assert body["message"] == "ok" + assert body["data"] == {"a": 1} + + +def test_error_envelope_carries_message_and_hint() -> None: + body = fail(ErrorCode.QUOTA_EXHAUSTED, data={"required": 1, "remaining": 0}) + assert body["code"] == 40301 + assert body["message"] == "今日额度已用完" + assert body["hint"] + + +def test_quota_hint_never_mentions_price_or_purchase() -> None: + """FR-BILL-07: 提示中不得出现任何价格、购买、会员、升级字样。""" + forbidden = ("价格", "购买", "会员", "升级", "元", "付费", "充值") + for code, meta in ERROR_META.items(): + text = f"{meta.message}{meta.hint or ''}" + assert not any(word in text for word in forbidden), f"{code.name}: {text}" + + +def test_every_error_code_has_metadata() -> None: + for code in ErrorCode: + assert code in ERROR_META, f"{code.name} has no message/hint" + assert ERROR_META[code].message + + +def test_error_code_segments_map_to_http_families() -> None: + for code, meta in ERROR_META.items(): + if code is ErrorCode.OK: + continue + segment = int(code) // 10000 + assert segment in (4, 5, 6), code.name + if segment == 4: + assert 400 <= meta.http_status < 500, code.name + else: + assert meta.http_status >= 500, code.name + + +def test_app_error_defaults_come_from_the_registry() -> None: + err = AppError(ErrorCode.NOT_FOUND) + assert err.http_status == 404 + assert err.message == "内容不存在或已删除" + + +def test_quota_exhausted_error_carries_actionable_data() -> None: + err = QuotaExhaustedError(required=1, remaining=0, reset_at="2026-08-14T00:00:00+08:00") + assert err.code is ErrorCode.QUOTA_EXHAUSTED + assert err.data["remaining"] == 0 + assert err.data["reset_at"] + + +def test_jwt_round_trip() -> None: + token = create_access_token(user_id=123, secret="s" * 32, ttl_seconds=900) + claims = decode_access_token(token, secret="s" * 32) + assert isinstance(claims, AccessTokenClaims) + assert claims.user_id == 123 + + +def test_expired_token_maps_to_a_user_readable_code() -> None: + token = create_access_token(user_id=1, secret="s" * 32, ttl_seconds=1, now=0) + try: + decode_access_token(token, secret="s" * 32) + except AppError as exc: + assert exc.code is ErrorCode.TOKEN_EXPIRED + else: # pragma: no cover + raise AssertionError("expected AppError") + + +def test_tampered_token_is_rejected() -> None: + token = create_access_token(user_id=1, secret="s" * 32) + try: + decode_access_token(token + "x", secret="s" * 32) + except AppError as exc: + assert exc.code is ErrorCode.TOKEN_INVALID + else: # pragma: no cover + raise AssertionError("expected AppError") diff --git a/tests/unit/test_routing.py b/tests/unit/test_routing.py new file mode 100755 index 0000000..11fae94 --- /dev/null +++ b/tests/unit/test_routing.py @@ -0,0 +1,77 @@ +"""Scene routing configuration (架构文档 §8.1: 改模型不改代码).""" + +from __future__ import annotations + +import pytest + +from packages.llm.errors import SceneNotRoutedError +from packages.llm.routing import SceneRouter, build_scene_configs +from packages.llm.types import Scene + + +def test_defaults_put_the_primary_first_and_the_backup_second() -> None: + configs = build_scene_configs(primary_provider="deepseek", backup_provider="zhipu") + assert configs[Scene.AGENT_CHAT].providers == ("deepseek", "zhipu") + + +def test_a_missing_backup_yields_a_single_provider_chain() -> None: + configs = build_scene_configs(primary_provider="deepseek", backup_provider=None) + assert configs[Scene.AGENT_CHAT].providers == ("deepseek",) + + +def test_every_scene_is_routed_and_bound_to_a_prompt_version() -> None: + configs = build_scene_configs(primary_provider="deepseek") + assert set(configs) == set(Scene) + for config in configs.values(): + assert config.prompt_id + assert config.prompt_version + + +def test_intent_classification_is_critical_so_budget_pressure_cannot_disable_it() -> None: + configs = build_scene_configs(primary_provider="deepseek") + assert configs[Scene.INTENT_CLASSIFY].critical is True + assert configs[Scene.AGENT_CHAT].critical is False + + +def test_agent_chat_output_budget_matches_the_token_plan() -> None: + # 架构文档 §7.2: 输出 ≤ 800 + assert build_scene_configs(primary_provider="x")[Scene.AGENT_CHAT].max_tokens <= 800 + + +def test_overrides_change_routing_without_a_code_change() -> None: + configs = build_scene_configs( + primary_provider="deepseek", + backup_provider="zhipu", + overrides={ + "agent_chat": { + "max_tokens": 1200, + "prompt_version": "v2", + "models": {"deepseek": "deepseek-v4-pro"}, + "providers": ["zhipu", "deepseek"], + } + }, + ) + config = configs[Scene.AGENT_CHAT] + assert config.max_tokens == 1200 + assert config.prompt_version == "v2" + assert config.providers == ("zhipu", "deepseek") + assert config.model_for("deepseek", "fallback") == "deepseek-v4-pro" + assert config.model_for("zhipu", "fallback") == "fallback" + # Untouched scenes keep their defaults. + assert configs[Scene.REPORT_COPY].max_tokens == 3000 + + +def test_structured_retry_budget_is_configurable() -> None: + configs = build_scene_configs(primary_provider="x", structured_max_retries=5) + assert configs[Scene.AGENT_CHAT].structured_max_retries == 5 + + +def test_router_raises_for_an_unconfigured_scene() -> None: + router = SceneRouter({}) + with pytest.raises(SceneNotRoutedError): + router.resolve(Scene.AGENT_CHAT) + + +def test_router_lists_configured_scenes() -> None: + router = SceneRouter(build_scene_configs(primary_provider="x")) + assert set(router.scenes()) == set(Scene) diff --git a/tests/unit/test_task_state.py b/tests/unit/test_task_state.py new file mode 100755 index 0000000..30f2471 --- /dev/null +++ b/tests/unit/test_task_state.py @@ -0,0 +1,66 @@ +"""tk_task state machine (架构文档 §6.2 task 域, FR-TASK-01).""" + +from __future__ import annotations + +from itertools import pairwise + +import pytest + +from packages.domain.errors import InvalidTaskTransitionError +from packages.domain.task.state import ( + ALLOWED_TRANSITIONS, + TaskStatus, + TaskType, + assert_transition, + can_transition, + is_terminal, + needs_refund, + progress_for, +) + + +def test_happy_path() -> None: + path = [TaskStatus.PENDING, TaskStatus.QUEUED, TaskStatus.RUNNING, TaskStatus.SUCCEEDED] + for current, target in pairwise(path): + assert can_transition(current, target) + + +def test_terminal_states_are_final_except_retry() -> None: + assert ALLOWED_TRANSITIONS[TaskStatus.SUCCEEDED] == frozenset() + assert ALLOWED_TRANSITIONS[TaskStatus.CANCELED] == frozenset() + # FR-TASK-03: 失败项有原因与重试入口 + assert can_transition(TaskStatus.FAILED, TaskStatus.QUEUED) + + +def test_illegal_transition_names_both_states() -> None: + with pytest.raises(InvalidTaskTransitionError) as excinfo: + assert_transition(TaskStatus.SUCCEEDED, TaskStatus.RUNNING) + assert excinfo.value.current == "succeeded" + assert excinfo.value.target == "running" + + +def test_cannot_skip_straight_from_pending_to_succeeded() -> None: + assert not can_transition(TaskStatus.PENDING, TaskStatus.SUCCEEDED) + + +def test_terminal_and_refund_classification() -> None: + assert is_terminal(TaskStatus.SUCCEEDED) + assert not is_terminal(TaskStatus.RUNNING) + # FR-TRY-05 / FR-BILL-05: 无「扣了额度但没出图」case + assert needs_refund(TaskStatus.FAILED) + assert needs_refund(TaskStatus.CANCELED) + assert not needs_refund(TaskStatus.SUCCEEDED) + + +def test_progress_is_defined_for_every_status() -> None: + for status in TaskStatus: + assert 0 <= progress_for(status) <= 100 + assert progress_for(TaskStatus.SUCCEEDED) == 100 + + +def test_every_status_has_a_transition_entry() -> None: + assert set(ALLOWED_TRANSITIONS) == set(TaskStatus) + + +def test_task_types_cover_the_long_running_work() -> None: + assert {"report", "tryon", "hairstyle", "matting", "render"} <= {t.value for t in TaskType}