"""核心单元测试：跟踪器、计量器、来源、提示词、工具、并发、计费、存储。 这些用例全部离线可跑（不起服务、不连模型）。端到端的 HTTP 验证见 ``scripts/http_e2e.py``（需要一个真实可用的模型密钥）。 """ from __future__ import annotations import json import pytest # ============================================================ # 子智能体跟踪 # ============================================================ class TestSubagentTracker: def test_namespace_prefix_parsing(self): from runtime.subagent_tracker import parse_namespace assert parse_namespace(("tools:abc123",)) == "abc123" assert parse_namespace(["tools:x"]) == "x" assert parse_namespace("tools:y") == "y" def test_empty_namespace_is_main_agent(self): from runtime.subagent_tracker import parse_namespace assert parse_namespace(()) is None assert parse_namespace(None) is None assert parse_namespace([]) is None def test_control_segments_not_mistaken_for_ids(self): """带冒号/斜杠的控制段不能被当成 tool_call id。""" from runtime.subagent_tracker import parse_namespace assert parse_namespace(("model_request:1",)) is None assert parse_namespace(("a/b",)) is None def test_lifecycle_started_to_done(self): from runtime.subagent_tracker import SubagentTracker tracker = SubagentTracker() calls = [{"name": "task", "id": "c1", "args": {"subagent_type": "demo-searcher", "description": "查资料"}}] started = tracker.note_task_calls(calls) assert len(started) == 1 assert started[0].subagent_type == "demo-searcher" assert started[0].status == "started" run = tracker.note_namespace(("tools:c1",)) assert run is not None and run.status == "running" and run.tool_calls == 1 done = tracker.note_result("c1", ok=True) assert done is not None and done.status == "done" assert done.duration_ms is not None def test_non_task_calls_ignored(self): from runtime.subagent_tracker import SubagentTracker tracker = SubagentTracker() assert tracker.note_task_calls([{"name": "calculate", "id": "c1", "args": {}}]) == [] assert len(tracker) == 0 def test_call_without_id_skipped(self): """没有 id 就无法关联后续子图事件，跳过而不是编一个假 id。""" from runtime.subagent_tracker import SubagentTracker tracker = SubagentTracker() assert tracker.note_task_calls([{"name": "task", "args": {"subagent_type": "x"}}]) == [] assert len(tracker) == 0 def test_unknown_namespace_does_not_invent_run(self): """表里没有的 id 不能凭空造一条 subagent_type=unknown 的记录。""" from runtime.subagent_tracker import SubagentTracker tracker = SubagentTracker() assert tracker.note_namespace(("tools:never-seen",)) is None assert len(tracker) == 0 def test_finish_all_flushes_active(self): """流异常退出时不能让子智能体永远挂在 started，前端卡片会一直转。""" from runtime.subagent_tracker import SubagentTracker tracker = SubagentTracker() tracker.note_task_calls([ {"name": "task", "id": "c1", "args": {"subagent_type": "a"}}, {"name": "task", "id": "c2", "args": {"subagent_type": "b"}}, ]) tracker.note_result("c1", ok=True) flushed = tracker.finish_all() assert len(flushed) == 1 and flushed[0].call_id == "c2" assert not tracker.active() def test_duplicate_task_call_not_counted_twice(self): from runtime.subagent_tracker import SubagentTracker tracker = SubagentTracker() call = {"name": "task", "id": "c1", "args": {"subagent_type": "a"}} tracker.note_task_calls([call]) tracker.note_task_calls([call]) assert len(tracker) == 1 # ============================================================ # 上下文计量 # ============================================================ class _Msg: """最小的消息替身。""" def __init__(self, content="", usage_metadata=None, response_metadata=None, additional_kwargs=None, name=None) -> None: self.content = content self.usage_metadata = usage_metadata self.response_metadata = response_metadata or {} self.additional_kwargs = additional_kwargs or {} self.name = name class TestContextMeter: def test_extract_usage_from_langchain_metadata(self): from runtime.context_meter import extract_usage usage = extract_usage(_Msg(usage_metadata={ "input_tokens": 100, "output_tokens": 20, "total_tokens": 120, "input_token_details": {"cache_read": 30}, })) assert usage == {"input_tokens": 100, "output_tokens": 20, "cache_read_tokens": 30} def test_extract_usage_from_provider_raw(self): from runtime.context_meter import extract_usage usage = extract_usage(_Msg(response_metadata={ "token_usage": {"prompt_tokens": 50, "completion_tokens": 5, "prompt_tokens_details": {"cached_tokens": 8}} })) assert usage == {"input_tokens": 50, "output_tokens": 5, "cache_read_tokens": 8} def test_missing_usage_returns_none_not_zero(self): """「用量为 0」与「没有用量信息」必须能区分，否则估算兜底会失效。""" from runtime.context_meter import extract_usage assert extract_usage(_Msg()) is None assert extract_usage(_Msg(usage_metadata={})) is None def test_accumulates_across_model_calls(self): from runtime.context_meter import ContextMeter meter = ContextMeter(window=1000) meter.note_message(_Msg(usage_metadata={"input_tokens": 100, "output_tokens": 10})) meter.note_message(_Msg(usage_metadata={"input_tokens": 150, "output_tokens": 20})) assert meter.input_tokens == 250 assert meter.output_tokens == 30 assert meter.model_calls == 2 def test_context_tokens_is_last_input(self): """上下文占用取最近一次调用的 input，而不是累加值。""" from runtime.context_meter import ContextMeter meter = ContextMeter(window=1000) meter.note_message(_Msg(usage_metadata={"input_tokens": 100, "output_tokens": 10})) meter.note_message(_Msg(usage_metadata={"input_tokens": 180, "output_tokens": 10})) assert meter.context_tokens == 180 assert meter.input_tokens == 280 def test_source_is_provider_when_usage_present(self): from runtime.context_meter import ContextMeter meter = ContextMeter(window=1000) meter.note_message(_Msg(usage_metadata={"input_tokens": 5, "output_tokens": 5})) assert meter.usage_source == "provider" def test_estimate_marks_source_estimate(self): from runtime.context_meter import ContextMeter meter = ContextMeter(window=1000) meter.note_estimated(300) assert meter.usage_source == "estimate" assert meter.input_tokens == 300 def test_estimate_ignored_when_real_usage_seen(self): """已有真实用量时估算值只会污染口径，丢弃。""" from runtime.context_meter import ContextMeter meter = ContextMeter(window=1000) meter.note_message(_Msg(usage_metadata={"input_tokens": 100, "output_tokens": 10})) meter.note_estimated(9999) assert meter.input_tokens == 100 assert meter.usage_source == "provider" def test_compaction_detected_by_lc_source(self): from runtime.context_meter import ContextMeter meter = ContextMeter(window=1000) meter.note_message(_Msg(content="摘要", response_metadata={"lc_source": "summarization"})) assert meter.compacted and meter.compact_count == 1 def test_compaction_detected_by_context_drop(self): """占用断崖式下跌也视为压缩。""" from runtime.context_meter import ContextMeter meter = ContextMeter(window=10000) meter.note_message(_Msg(usage_metadata={"input_tokens": 9000, "output_tokens": 10})) assert not meter.compacted meter.note_message(_Msg(usage_metadata={"input_tokens": 2000, "output_tokens": 10})) assert meter.compacted and meter.compact_count == 1 def test_compaction_counted_once(self): """两个信号同时命中不能重复计数。""" from runtime.context_meter import ContextMeter meter = ContextMeter(window=10000) meter.note_message(_Msg(usage_metadata={"input_tokens": 9000, "output_tokens": 1})) meter.note_message(_Msg(usage_metadata={"input_tokens": 100, "output_tokens": 1})) meter.note_message(_Msg(content="x", response_metadata={"lc_source": "summarization"})) assert meter.compact_count == 1 def test_archive_text_captured_on_compaction(self): from runtime.context_meter import ContextMeter meter = ContextMeter(window=1000) meter.note_text("被压缩掉的早期对话") meter.note_message(_Msg(content="摘要", response_metadata={"lc_source": "summarization"})) snapshot = meter.snapshot() assert snapshot.compacted and snapshot.archive_text == "被压缩掉的早期对话" def test_fill_ratio(self): from runtime.context_meter import ContextMeter meter = ContextMeter(window=1000) meter.note_message(_Msg(usage_metadata={"input_tokens": 850, "output_tokens": 0})) assert meter.fill_ratio() == pytest.approx(0.85) def test_estimate_tokens_positive_for_cjk(self): from runtime.context_meter import estimate_tokens assert estimate_tokens("矿井通风") > 0 assert estimate_tokens("") == 0 # ============================================================ # 来源标注 # ============================================================ class TestSourceTracker: def test_extract_from_sourced_payload(self): from runtime.source_tracker import SourceTracker from tools.registry import source, sourced tracker = SourceTracker() payload = json.dumps(sourced(data=[1], sources=[source("T", "https://a.invalid/x", "摘要", 0.5)])) added = tracker.note_tool_result("search_demo_knowledge", payload) assert len(added) == 1 and added[0]["title"] == "T" def test_extract_from_dict_payload(self): from runtime.source_tracker import SourceTracker from tools.registry import source, sourced tracker = SourceTracker() tracker.note_tool_result("t", sourced(data=[], sources=[source("T", "https://a.invalid")])) assert tracker.count == 1 def test_dedup_by_url(self): from runtime.source_tracker import SourceTracker from tools.registry import source, sourced tracker = SourceTracker() payload = sourced(data=[], sources=[source("T", "https://same.invalid"), source("T2", "https://same.invalid")]) tracker.note_tool_result("t", payload) assert tracker.count == 1 def test_records_without_title_or_url_dropped(self): """既无标题也无链接的来源无法核实，不展示空壳。""" from runtime.source_tracker import SourceTracker from tools.registry import sourced tracker = SourceTracker() tracker.note_tool_result("t", sourced(data=[], sources=[{"snippet": "只有摘要"}])) assert tracker.count == 0 def test_plain_text_yields_nothing(self): from runtime.source_tracker import SourceTracker tracker = SourceTracker() tracker.note_tool_result("t", "这是一段普通文本，没有 sources") assert tracker.count == 0 def test_classification_web_vs_kb(self): from runtime.source_tracker import SourceTracker from tools.registry import source, sourced tracker = SourceTracker() tracker.note_tool_result("t", sourced(data=[], sources=[ source("网页", "https://news.example.com/a"), source("知识库", "https://kb.example.com/b"), ])) types = {i["title"]: i["type"] for i in tracker.build_citations()} assert types["网页"] == "web" assert types["知识库"] == "knowledge_base" def test_nested_payload_lookup(self): from runtime.source_tracker import SourceTracker from tools.registry import source tracker = SourceTracker() tracker.note_tool_result("t", {"data": {"sources": [source("深层", "https://d.invalid")]}}) assert tracker.count == 1 and tracker.build_citations()[0]["title"] == "深层" def test_citations_sorted_by_score(self): from runtime.source_tracker import SourceTracker from tools.registry import source, sourced tracker = SourceTracker() tracker.note_tool_result("t", sourced(data=[], sources=[ source("低", "https://a.invalid", score=0.1), source("高", "https://b.invalid", score=0.9), ])) assert [i["title"] for i in tracker.build_citations()] == ["高", "低"] def test_label_attached(self): from runtime.source_tracker import SourceTracker from tools.registry import source, sourced tracker = SourceTracker() tracker.note_tool_result("search_demo_knowledge", sourced(data=[], sources=[source("T", "https://a.invalid")])) item = tracker.build_citations()[0] assert item["tool"] == "search_demo_knowledge" assert item["tool_cn"] == "检索演示知识库" # ============================================================ # 提示词与消息前缀 # ============================================================ class TestPrompts: def test_skill_prefix_extracted(self): from agents.prompts import extract_skill_prefix text, skill = extract_skill_prefix("#skill:demo-skill 帮我写一份简报") assert skill == "demo-skill" and text == "帮我写一份简报" def test_skill_prefix_only_at_start(self): """正文中间的 #skill: 是普通文本，不能当指令。""" from agents.prompts import extract_skill_prefix text, skill = extract_skill_prefix("请看这个 #skill:demo-skill 的用法") assert skill is None and text.startswith("请看这个") def test_skill_prefix_without_message(self): from agents.prompts import extract_skill_prefix text, skill = extract_skill_prefix("#skill:demo-skill") assert skill == "demo-skill" and text == "" def test_path_traversal_rejected(self): """技能名带路径分隔符不能当技能解析——否则能读任意文件。""" from agents.prompts import extract_skill_prefix for bad in ("#skill:../../etc/passwd x", "#skill:.hidden x", "#skill:a/b x"): _, skill = extract_skill_prefix(bad) assert skill is None, bad def test_session_ref_extracted(self): from agents.prompts import extract_session_ref text, ref = extract_session_ref("#session:abc123 接着上次说") assert ref == "abc123" and text == "接着上次说" def test_both_prefixes_stacked(self): from agents.prompts import extract_session_ref, extract_skill_prefix text, skill = extract_skill_prefix("#skill:demo-skill #session:abc 继续") text, ref = extract_session_ref(text) assert skill == "demo-skill" and ref == "abc" and text == "继续" def test_mode_instruction_differs(self): from agents.prompts import build_mode_instruction plan = build_mode_instruction("plan") full = build_mode_instruction("full") assert "plan" in plan and "计划" in plan assert "full" in full and plan != full def test_preference_instruction_empty_when_no_prefs(self): from agents.prompts import build_preference_instruction assert build_preference_instruction("") == "" def test_time_block_format(self): from agents.prompts import build_time_block block = build_time_block() assert "## 系统时间" in block and "当前时间" in block def test_system_prompt_includes_subagents(self): from agents.prompts import build_system_prompt prompt = build_system_prompt([ {"name": "demo-searcher", "description": "检索专员"} ]) assert "demo-searcher" in prompt assert "检索专员" in prompt def test_system_prompt_without_subagents(self): from agents.prompts import build_system_prompt prompt = build_system_prompt([]) assert "子智能体" in prompt # 通用说明里提到，但不该有具体名单 assert "可委派的子智能体" not in prompt # ============================================================ # 工具 # ============================================================ class TestTools: def test_builtin_tools_registered(self): from tools import registry names = set(registry.get_tool_names()) assert {"get_current_time", "search_demo_knowledge", "calculate", "request_plan_approval"} <= names def test_tools_are_basetool_instances(self): """子智能体的 tools 必须是 BaseTool（ToolNode 要读 .name）。""" from tools import registry for tool in registry.get_all_tools(): assert hasattr(tool, "name") and hasattr(tool, "description") def test_get_tools_by_name_skips_unknown(self): from tools import registry tools = registry.get_tools_by_name(["calculate", "does-not-exist"]) assert [t.name for t in tools] == ["calculate"] def test_duplicate_registration_raises(self): from tools.registry import register, reset_for_tests reset_for_tests() @register(group="X") def dup_tool() -> str: """d""" return "" with pytest.raises(ValueError, match="工具名冲突"): @register(group="Y") def dup_tool() -> str: # noqa: F811 """d""" return "" reset_for_tests() from tools import registry as r r.load_builtin_tools() def test_calculate_basic(self): from tools.demo_tools import calculate assert calculate("(128000 * 0.85)")["result"] == 108800 def test_calculate_integer_result_is_int(self): from tools.demo_tools import calculate assert isinstance(calculate("1+1")["result"], int) def test_calculate_rejects_code_execution(self): """表达式求值绝不能变成代码执行入口。""" from tools.demo_tools import calculate for evil in [ "__import__('os').system('echo x')", "open('/etc/passwd').read()", "(lambda: 1)()", "1 if True else 2", "[x for x in range(3)]", ]: result = calculate(evil) assert "error" in result, evil def test_calculate_rejects_huge_power(self): from tools.demo_tools import calculate assert "error" in calculate("9 ** 9 ** 9") def test_calculate_zero_division(self): from tools.demo_tools import calculate assert "零" in calculate("1/0")["error"] def test_calculate_empty(self): from tools.demo_tools import calculate assert "error" in calculate("") def test_calculate_returns_error_not_raises(self): """工具失败要返回可读原因，抛异常会打断整轮工具调用。""" from tools.demo_tools import calculate result = calculate("1 +") assert isinstance(result, dict) and "error" in result def test_search_returns_sources_contract(self): from tools.demo_tools import search_demo_knowledge result = search_demo_knowledge("需风量") assert result["data"] and result["sources"] assert result["sources"][0]["url"] def test_search_empty_query_returns_error(self): from tools.demo_tools import search_demo_knowledge result = search_demo_knowledge(" ") assert result["error"] and result["sources"] == [] def test_search_no_match(self): from tools.demo_tools import search_demo_knowledge result = search_demo_knowledge("不存在的关键词zzz") assert result["data"] == [] and result["sources"] == [] def test_registry_describe_shape(self): from tools import registry described = registry.describe() assert described["count"] >= 4 assert all(isinstance(v, list) for v in described["groups"].values()) # ============================================================ # 并发闸 # ============================================================ class TestConcurrency: def setup_method(self): from runtime import concurrency concurrency.reset_for_tests() def teardown_method(self): from runtime import concurrency concurrency.reset_for_tests() def test_acquire_and_release(self): from runtime import concurrency slot = concurrency.acquire_task_slot("u1") assert concurrency.stats()["global_active"] == 1 assert slot.release() is True assert concurrency.stats()["global_active"] == 0 def test_release_is_idempotent(self): """SSE 有多条退出路径，重复 release 不能把计数减成负数。""" from runtime import concurrency slot = concurrency.acquire_task_slot("u1") slot.release() slot.release() assert concurrency.stats()["global_active"] == 0 def test_per_user_limit(self, monkeypatch): from runtime import concurrency monkeypatch.setattr(concurrency.config, "MAX_USER_TASKS", 1) concurrency.acquire_task_slot("u1") with pytest.raises(concurrency.TooManyTasksError) as exc: concurrency.acquire_task_slot("u1") assert exc.value.scope == "user" def test_global_limit(self, monkeypatch): from runtime import concurrency monkeypatch.setattr(concurrency.config, "MAX_CONCURRENT_TASKS", 1) monkeypatch.setattr(concurrency.config, "MAX_USER_TASKS", 10) concurrency.acquire_task_slot("u1") with pytest.raises(concurrency.TooManyTasksError) as exc: concurrency.acquire_task_slot("u2") assert exc.value.scope == "global" def test_different_users_independent(self, monkeypatch): from runtime import concurrency monkeypatch.setattr(concurrency.config, "MAX_USER_TASKS", 1) concurrency.acquire_task_slot("u1") concurrency.acquire_task_slot("u2") assert concurrency.stats()["global_active"] == 2 # ============================================================ # 计费 # ============================================================ class TestBilling: def test_cost_computation(self, pricing_ready): from billing import pricing from db import billing_store billing_store.set_pricing("m1", 0.001, 0.002) cost, currency = pricing.compute_cost("m1", 1000, 500) assert cost == pytest.approx(0.001 + 0.001) assert currency == "CNY" def test_unpriced_model_fails_closed(self, pricing_ready, monkeypatch): """静默按 0 计费等于白送额度，所以默认必须报错。""" from billing import pricing monkeypatch.setattr(pricing.config, "get_bool", lambda *a, **k: False) with pytest.raises(pricing.UnpricedModelError): pricing.compute_cost("unknown-model", 100, 100) def test_unpriced_model_allowed_when_configured(self, pricing_ready, monkeypatch): from billing import pricing monkeypatch.setattr(pricing.config, "get_bool", lambda *a, **k: True) cost, _ = pricing.compute_cost("unknown-model", 100, 100) assert cost == 0.0 def test_quota_blocks_when_exhausted(self, pricing_ready, unique_user): from billing import guard from db import billing_store, db_conn day = db_conn.day_key() billing_store.set_quota_limit(unique_user, day, 100) billing_store.record_usage(unique_user, "s", "chat", "m", 100, 0, 0.0, day) with pytest.raises(guard.QuotaExceededError): guard.precheck(unique_user, day) def test_quota_zero_means_unlimited(self, pricing_ready, unique_user): from billing import guard from db import billing_store, db_conn day = db_conn.day_key() billing_store.set_quota_limit(unique_user, day, 0) billing_store.record_usage(unique_user, "s", "chat", "m", 10_000_000, 0, 0.0, day) guard.precheck(unique_user, day) # 不抛即通过 def test_meter_settle_idempotent(self, pricing_ready, unique_user): from billing import guard from billing.meter import UsageMeter meter = UsageMeter(unique_user, "s1", "chat", "deepseek-flash") meter.add(1000, 100) first = meter.settle() second = meter.settle() assert first["settled"] is True assert second.get("duplicated") is True assert guard.snapshot(unique_user)["requests"] == 1 def test_meter_no_usage_skips(self, pricing_ready, unique_user): from billing import guard from billing.meter import UsageMeter meter = UsageMeter(unique_user, "s1", "chat", "deepseek-flash") result = meter.settle() assert result["settled"] is False and result["reason"] == "no_usage" assert guard.snapshot(unique_user)["requests"] == 0 def test_meter_source_downgrades_to_estimate(self, pricing_ready, unique_user): from billing.meter import UsageMeter meter = UsageMeter(unique_user, "s1", "chat", "deepseek-flash") meter.add(100, 10, "provider") meter.add(50, 5, "estimate") assert meter.usage_source == "estimate" def test_snapshot_has_total_tokens(self, pricing_ready, unique_user): """字段名与 billing_store.usage_for_day 对齐，消费方不用记两套名字。""" from billing import guard snapshot = guard.snapshot(unique_user) assert "total_tokens" in snapshot and "used_tokens" in snapshot def test_usage_day_bound_to_request_start(self, pricing_ready, unique_user): """跨零点的长请求不能被算到第二天。""" from billing import guard from billing.meter import UsageMeter from db import db_conn yesterday = "20200101" meter = UsageMeter(unique_user, "s1", "chat", "deepseek-flash", usage_day=yesterday) meter.add(100, 10) meter.settle() assert guard.snapshot(unique_user, yesterday)["requests"] == 1 assert guard.snapshot(unique_user, db_conn.day_key())["requests"] == 0 # ============================================================ # 存储 # ============================================================ class TestStores: def test_message_ordering_stable_within_same_second(self): """同秒写入的多条消息顺序必须稳定——排序错乱会让补种的上下文角色错位。""" from db import chat_store sid = chat_store.create_session("u", "full")["id"] for i in range(6): chat_store.save_message(sid, "user", f"Q{i}") chat_store.save_message(sid, "assistant", f"A{i}") rows = chat_store.get_messages(sid) assert [r["msg_role"] for r in rows] == ["user", "assistant"] * 6 assert [r["content"] for r in rows] == [ c for i in range(6) for c in (f"Q{i}", f"A{i}") ] def test_get_messages_limit_returns_most_recent_in_order(self): from db import chat_store sid = chat_store.create_session("u", "full")["id"] for i in range(5): chat_store.save_message(sid, "user", f"M{i}") recent = chat_store.get_messages(sid, limit=3) assert [r["content"] for r in recent] == ["M2", "M3", "M4"] def test_message_json_fields_roundtrip(self): from db import chat_store sid = chat_store.create_session("u", "full")["id"] chat_store.save_message( sid, "assistant", "答", tool_calls=[{"name": "calculate", "args": {"expression": "1+1"}}], citations=[{"title": "T", "url": "u"}], ) row = chat_store.get_messages(sid)[0] assert row["tool_calls"][0]["name"] == "calculate" assert row["citations"][0]["title"] == "T" def test_session_ownership_check(self): from db import chat_store sid = chat_store.create_session("alice", "full")["id"] assert chat_store.get_session_for_user(sid, "alice") is not None assert chat_store.get_session_for_user(sid, "bob") is None def test_delete_session_cascades(self): from db import chat_store sid = chat_store.create_session("u", "full")["id"] chat_store.save_message(sid, "user", "x") chat_store.save_context_usage(sid, 1, 10, 5) chat_store.delete_session(sid) assert chat_store.get_session(sid) is None assert chat_store.get_messages(sid) == [] assert chat_store.get_context_usage(sid)["history"] == [] def test_preference_upsert(self): from db import chat_store chat_store.save_preference("u", "单位", "米") chat_store.save_preference("u", "单位", "千米") prefs = {p["pref_key"]: p["pref_value"] for p in chat_store.list_preferences("u")} assert prefs == {"单位": "千米"} def test_preferences_as_text_empty(self): from db import chat_store assert chat_store.preferences_as_text("nobody") == "" def test_search_sessions_by_title(self): from db import chat_store chat_store.create_session("u", "full", "通风计算讨论") chat_store.create_session("u", "full", "其他话题") found = chat_store.search_sessions("u", "通风") assert len(found) == 1 and "通风" in found[0]["title"] def test_pinned_sessions_first(self): from db import chat_store a = chat_store.create_session("u", "full", "A")["id"] chat_store.create_session("u", "full", "B") chat_store.update_session_pin(a, True) assert chat_store.list_sessions("u")[0]["id"] == a def test_context_usage_aggregates(self): from db import chat_store sid = chat_store.create_session("u", "full")["id"] chat_store.save_context_usage(sid, 1, 100, 10, usage_source="provider") chat_store.save_context_usage(sid, 2, 200, 20, compacted=True, compact_count=1, usage_source="estimate") usage = chat_store.get_context_usage(sid) assert usage["compact_count"] == 1 assert usage["latest"]["total_tokens"] == 220 assert usage["latest"]["usage_source"] == "estimate" def test_schedule_crud_and_runs(self): from db import schedule_store schedule = schedule_store.create_schedule("u", "每日", "0 9 * * *", "汇报") sid = schedule["id"] assert schedule["enabled"] is True run_id = schedule_store.start_run(sid, "thread-1") schedule_store.finish_run(run_id, "success", output="ok") runs = schedule_store.list_runs(sid) assert len(runs) == 1 and runs[0]["status"] == "success" updated = schedule_store.update_schedule(sid, enabled=False, title="改名") assert updated["enabled"] is False and updated["title"] == "改名" assert schedule_store.delete_schedule(sid) is True assert schedule_store.get_schedule(sid) is None assert schedule_store.list_runs(sid) == [] def test_schedule_update_ignores_unknown_fields(self): from db import schedule_store sid = schedule_store.create_schedule("u", "T", "0 9 * * *", "P")["id"] schedule_store.update_schedule(sid, title="新", 恶意字段="x") assert schedule_store.get_schedule(sid)["title"] == "新" def test_init_schema_idempotent(self): from db import db_conn db_conn.init_schema() db_conn.init_schema(force=True) assert db_conn.health()["status"] == "ok" # ============================================================ # 路由面（把「删掉 /chat/only」这个决定变成可执行的守卫） # ============================================================ class TestRouting: """断言**实际对外服务**的路由面。 这类断言的价值不在于「测试代码」，而在于把架构决定固化下来： 没有它，将来有人「为了兼容 1.5」把 /chat/only 加回来时，只有 code review 能拦住；有了它，CI 会拦。 刻意断言 ``main.app.openapi()["paths"]`` 而不是 ``api.router.routes``： 后者在本版 FastAPI 里是 ``_IncludedRouter`` 包装（不含 ``.path``）， 而且真正要守住的是**服务出去的**那套路径。顺带还能验证路由定义合法—— schema 生成失败会在这里报出来。 """ def _paths(self) -> set[str]: from main import app return set(app.openapi()["paths"]) def test_chat_entry_exists(self): assert "/api/chat" in self._paths() def test_chat_only_is_gone(self): """2.0 刻意删除 /chat/only：它与 /chat 行为完全相同，唯一的 kind 标签 无人消费，且 1.5 为此付过计费 bug 与重复验证的代价。见 AGENTS.md §六。""" assert "/api/chat/only" not in self._paths() def test_interrupt_endpoints_kept(self): """/chat/resume 与 /chat/interrupt 承载的是不同能力，不是同一能力的 别名，所以保留——别把它们和 /chat/only 一起删。""" paths = self._paths() assert "/api/chat/resume" in paths assert "/api/chat/interrupt" in paths def test_detail_routes_not_lost_when_removing(self): """删入口时最容易连带弄丢同族路由，这里把对话族钉死。""" chat_paths = {p for p in self._paths() if p.startswith("/api/chat")} assert chat_paths == { "/api/chat", "/api/chat/resume", "/api/chat/interrupt", "/api/chat/history/{session_id}", }, chat_paths def test_chat_only_route_not_defined_in_source(self): """源码里不得再出现该路由的装饰器。 只在装饰器形态上断言，不搜 ``chat_only`` 这个词——它在注释与文档里 正当地出现（用来解释为什么没有它），搜词会把说明文字也判成违规。 """ import inspect from api import chat_routes src = inspect.getsource(chat_routes) assert '@router.post("/chat/only")' not in src assert '"chat_only"' not in src def test_kind_values_are_meaningful(self): """kind 只剩三个有意义取值：chat / resume / scheduled。""" import inspect from api import chat_routes from scheduler import runner chat_src = inspect.getsource(chat_routes) assert 'kind="chat"' in chat_src assert 'kind="resume"' in chat_src assert 'kind="scheduled"' in inspect.getsource(runner) # ============================================================ # 模型工厂 # ============================================================ class TestModelFactory: def test_context_window_known(self): from agents import model_factory assert model_factory.context_window("deepseek-flash") == 128_000 def test_context_limit_override_wins(self, monkeypatch): from agents import model_factory monkeypatch.setattr(model_factory.config, "CONTEXT_LIMIT", 5000) assert model_factory.context_window("deepseek-flash") == 5000 def test_unknown_model_uses_fallback(self): from agents import model_factory assert model_factory.context_window("some-unknown-model-xyz") == \ model_factory.FALLBACK_CONTEXT_WINDOW def test_thinking_levels_normalized(self): from agents import model_factory assert model_factory.normalize_thinking("HIGH") == "high" assert model_factory.normalize_thinking("bogus") == "off" assert model_factory.normalize_thinking(None) in model_factory.VALID_THINKING_LEVELS def test_off_sends_no_extra_body(self): """off 时什么都不发：不同 provider 关闭思考的字段名不一致，发错会被拒。""" from agents import model_factory assert model_factory._thinking_payload("off") is None assert model_factory._thinking_payload("high") is not None def test_build_model_declares_profile(self): """必须显式声明上下文窗口，否则摘要中间件首次触发时会报错。""" from agents import model_factory model = model_factory.build_model() assert model.profile and model.profile.get("max_input_tokens") # ============================================================ # 中断恢复翻译 # ============================================================ class TestResumeTranslation: def _payload(self, **kw): from api.models import ResumeRequest base = {"session_id": "s", "status": "answered"} base.update(kw) return ResumeRequest(**base) def test_approve(self): from api.chat_routes import _to_official_resume assert _to_official_resume(self._payload(approved=True), 1) == \ {"decisions": [{"type": "approve"}]} def test_reject(self): from api.chat_routes import _to_official_resume decisions = _to_official_resume(self._payload(approved=False), 1)["decisions"] assert decisions[0]["type"] == "reject" and decisions[0]["message"] def test_cancelled_becomes_reject(self): from api.chat_routes import _to_official_resume decisions = _to_official_resume(self._payload(status="cancelled"), 1)["decisions"] assert decisions[0]["type"] == "reject" def test_answers_become_respond(self): from api.chat_routes import _to_official_resume decisions = _to_official_resume(self._payload(answers=["甲", "乙"]), 1)["decisions"] assert decisions[0]["type"] == "respond" and "甲" in decisions[0]["message"] def test_multiple_pending_actions_get_matching_decisions(self): """官方要求 decisions 与 action_requests 数量严格一致。""" from api.chat_routes import _to_official_resume decisions = _to_official_resume(self._payload(approved=True, interrupt_token="reviewed-batch"), 3)["decisions"] assert len(decisions) == 3 assert all(d["type"] == "approve" for d in decisions) def test_unknown_payload_defaults_to_safe_reject(self): """没有明确答复时宁可拒绝执行。""" from api.chat_routes import _to_official_resume decisions = _to_official_resume(self._payload(), 1)["decisions"] assert decisions[0]["type"] == "reject"