script-build-and-framework-tmp/visualization/backend/tests/test_module_audit.py

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