from __future__ import annotations from copy import deepcopy from importlib import import_module from fastapi.testclient import TestClient from app import main from app.module_audit.service import ModuleAuditService class _Provider: mode = "test" def __init__(self, bundle): self.bundle = deepcopy(bundle) self.load_bundle_calls = 0 def load_bundle(self, _script_build_id: int): self.load_bundle_calls += 1 return deepcopy(self.bundle) def load_event_detail(self, _script_build_id: int, event_id: int): event = next(item for item in self.bundle["events"] if item["id"] == event_id) return { **deepcopy(event), "input": {"content": deepcopy(event.get("inputData"))}, "output": { "content": deepcopy( event.get("agentOutputData") or event.get("output_preview") ) }, } class _Repository: def __init__(self, bundle): self.events = {item["id"]: deepcopy(item) for item in bundle["events"]} self.load_event_calls = 0 self.load_log_calls = 0 def load_event(self, _script_build_id: int, event_id: int): self.load_event_calls += 1 event = deepcopy(self.events.get(event_id)) if event is None: return None event["eventBody"] = { "input_content_type": "application/json", "input_content": '{"task":"完整派发任务"}', "output_content_type": "text/plain", "output_content": "完整运行返回", } return event def load_log(self, _script_build_id: int): self.load_log_calls += 1 return ( "[AGENT_TASK:script_implementer]\n" "[MSG_ANCHOR:msg_id=task-50:seq=10]\n" "完整派发任务\n" "[TOOL_ANCHOR:msg_id=tool-51:seq=11]\n" "后续工具调用" ) def _service(current_bundle): return ModuleAuditService( provider=_Provider(current_bundle), repository=_Repository(current_bundle) ) def test_audit_index_is_lightweight_and_collects_full_flow_modules(current_bundle): payload = _service(current_bundle).index(88001) module_ids = {item["id"] for item in payload["modules"]} assert "run:objective" in module_ids assert "round:1:branch:1:task" in module_ids assert "event:53" in module_ids assert "round:1:result" in module_ids assert not any(item["displayVariant"] == "collapsed-summary" for item in payload["modules"]) assert "round:1:summary" not in module_ids assert "round:1:branch:1:summary" not in module_ids assert "eventBody" not in str(payload) assert "contexts" not in payload def test_audit_reuses_completed_run_snapshot_between_index_and_detail(current_bundle): service = _service(current_bundle) service.index(88001) service.detail(88001, "round:1:branch:1:task") assert service.provider.load_bundle_calls == 1 def test_audit_reuses_completed_module_detail(current_bundle): service = _service(current_bundle) first = service.detail(88001, "round:1:branch:1:task") event_calls = service.repository.load_event_calls log_calls = service.repository.load_log_calls second = service.detail(88001, "round:1:branch:1:task") assert second is first assert service.repository.load_event_calls == event_calls assert service.repository.load_log_calls == log_calls def test_branch_task_audit_separates_card_event_and_fallback_sources(current_bundle): current_bundle["events"][4]["msg_id"] = "task-50" current_bundle["events"][4]["inputData"] = { "task": "完整的实现 Agent 派发任务" } payload = _service(current_bundle).detail(88001, "round:1:branch:1:task") assert payload["actualUse"]["card"]["title"] == "实现任务" reads = payload["actualReads"] assert any( item["consumer"] == "card" and item["fieldPath"] == "impl_task" and item["disposition"] == "selected" for item in reads ) event_read = next( item for item in reads if item["consumer"] == "inspector" and item["fieldPath"] == "inputData.task" ) assert event_read["value"] == "完整的实现 Agent 派发任务" assert event_read["disposition"] == "selected" assert event_read["transformation"] == "直接摘取" assert event_read["source"]["sourceKind"] == "database" assert event_read["sourceKey"] == "database:event:50" assert event_read["usageAreas"] == [ "Inspector 业务详情", "Inspector 技术详情", ] assert event_read["usageRatio"] == { "calculable": True, "value": 1.0, "label": "100.0%", } assert event_read["readId"].startswith("read:") assert event_read["contextRefs"] == [ { "contextId": "event:50", "sourceKey": "database:event:50", "contextFieldPath": "inputData.task", "relation": "direct", "exact": True, } ] fallback_read = next( item for item in reads if item["consumer"] == "inspector" and item["fieldPath"] == "impl_task" ) assert fallback_read["disposition"] == "fallback" assert fallback_read["sourcePriority"] == "备用来源" assert fallback_read["usageRatio"]["calculable"] is False assert any( item["consumer"] == "inspector" and item["fieldPath"] == "technical.branch" and item["transformation"] == "完整记录读取" for item in reads ) assert any( context["kind"] == "event-record" and context["content"]["eventBody"]["output_content"] == "完整运行返回" and isinstance(context["usedFragments"], list) for context in payload["contexts"] ) assert not any(item["label"] == "table" for item in reads) assert any( context["kind"] == "function-result" and context["source"]["sourceKind"] == "function" for context in payload["contexts"] ) assert any( context["kind"] == "log-module" and context["source"]["sourceKind"] == "log" and context["relatedSourceKey"] == "database:event:50" for context in payload["contexts"] ) assert any(context["kind"] == "log-module" for context in payload["contexts"]) def test_audit_routes_serve_page_assets_and_full_stream(monkeypatch, current_bundle): module_audit_router = import_module("app.module_audit.router") audit_service = _service(current_bundle) monkeypatch.setattr(module_audit_router, "service", audit_service) client = TestClient(main.app) page = client.get("/audit/88001") assert page.status_code == 200 assert 'data-run-id="88001"' in page.text assert 'id="module-jump"' in page.text assert "只看使用字段" in page.text assert 'id="source-filter"' in page.text assert client.get("/audit-assets/styles.css").status_code == 200 assert client.get("/audit-assets/app.js").status_code == 200 index = client.get("/api/script-builds/88001/module-audit") assert index.status_code == 200 assert "eventBody" not in index.text detail = client.get( "/api/script-builds/88001/module-audit/round%3A1%3Abranch%3A1%3Atask" ) assert detail.status_code == 200 assert detail.json()["id"] == "round:1:branch:1:task" stream = client.get("/api/script-builds/88001/module-audit-stream") assert stream.status_code == 200 assert stream.headers["content-type"].startswith("application/x-ndjson") assert '"type":"index"' in stream.text assert '"type":"detail"' in stream.text assert '"type":"complete"' in stream.text def test_audit_page_collapses_large_blocks_without_truncating_source(): script = ( main.__file__.replace("main.py", "module_audit/static/app.js") ) with open(script, encoding="utf-8") as handle: source = handle.read() assert "count > 800" in source assert "slice(0, length)" in source assert "jsonText(value)" in source assert 'contentBox("技术详情"' in source assert 'contentBox("业务详情"' in source assert 'class="businessPart${' in source assert 'class="correspondenceRow"' in source assert "contextsForGroups" in source assert "correlationMark" in source assert "setCorrelationFocus" in source assert "businessSegments" in source assert "cardSegments" in source assert "populateModuleJump" in source assert "jumpToModule" in source assert "scrollIntoView" in source assert "renderCompactReads" in source assert "renderCompactContexts" in source assert "实际读取值" in source assert "原始字段值" in source assert "原始记录位置" in source assert "定位原始字段" in source assert "data-locate-read-id" in source assert "data-read-ids" in source assert "module-audit-stream" in source assert "requestAnimationFrame" in source assert 'get("module")' in source assert "data-load-module" not in source assert "IntersectionObserver" not in source assert "Inspector 业务详情 ·" not in source assert "Inspector 读取 · table" not in source assert "来源投影" not in source assert "数据库：" in source and "函数：" in source and "日志：" in source