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

import inspect
from types import SimpleNamespace

from app.repositories import ScriptBuildRepository


class Row:
    def __init__(self, **values):
        self.__dict__.update(values)
        self.__table__ = SimpleNamespace(
            columns=[SimpleNamespace(name=name) for name in values]
        )


def test_final_snapshot_filters_branch_zero_and_overlay_replaces_base_rows():
    repository = ScriptBuildRepository()
    paragraphs = [
        Row(id=1, branch_id=0, is_active=True, base_ref_id=None, content="base"),
        Row(id=2, branch_id=7, is_active=True, base_ref_id=1, content="overlay"),
        Row(id=3, branch_id=9, is_active=True, base_ref_id=None, content="other"),
    ]
    elements = [
        Row(id=10, branch_id=0, is_active=True, base_ref_id=None, content="base-element"),
        Row(id=11, branch_id=7, is_active=True, base_ref_id=10, content="overlay-element"),
    ]
    links = [
        Row(id=20, branch_id=0, paragraph_id=1, element_id=10),
        Row(id=21, branch_id=7, paragraph_id=2, element_id=11),
    ]

    final_snapshot = repository._snapshot_for_branch(0, paragraphs, elements, links)
    assert {row["branch_id"] for row in final_snapshot["paragraphs"]} == {0}
    assert {row["branch_id"] for row in final_snapshot["elements"]} == {0}
    assert {row["branch_id"] for row in final_snapshot["paragraphElements"]} == {0}

    overlay = repository._snapshot_for_branch(7, paragraphs, elements, links)
    assert [row["content"] for row in overlay["paragraphs"]] == ["overlay"]
    assert [row["content"] for row in overlay["elements"]] == ["overlay-element"]


def test_candidate_snapshot_uses_saved_content_for_terminal_branch_and_overlay_for_park():
    repository = ScriptBuildRepository()
    merged = Row(status="merged", branch_id=1, content_snapshot={"paragraphs": [{"id": "saved"}]})
    parked = Row(status="parked", branch_id=2, content_snapshot=None)

    saved = repository._candidate_snapshot(merged, paragraphs=[], elements=[], links=[])
    assert saved["snapshot"]["paragraphs"][0]["id"] == "saved"
    assert saved["historicalAccuracy"] == "exact"
    assert saved["snapshotKind"] == "saved-at-disposition"

    overlay = repository._candidate_snapshot(parked, paragraphs=[], elements=[], links=[])
    assert overlay["sourceOrigin"] == "database"
    assert overlay["historicalAccuracy"] == "unknown"
    assert overlay["currentProjectionAccuracy"] == "exact"
    assert "不是暂存时的历史快照" in overlay["note"]


def test_terminal_branch_without_snapshot_is_not_reconstructed_from_current_base():
    repository = ScriptBuildRepository()
    merged = Row(status="merged", branch_id=7, content_snapshot=None)
    result = repository._candidate_snapshot(merged, paragraphs=[], elements=[], links=[])
    assert result["snapshot"] is None
    assert result["historicalAccuracy"] == "missing"
    assert result["currentProjectionAccuracy"] == "not-applicable"
    assert result["snapshotKind"] == "missing-disposition-snapshot"


def test_base_links_require_both_active_endpoints_and_override_has_effective_id():
    repository = ScriptBuildRepository()
    paragraphs = [
        Row(id=1, branch_id=0, is_active=True, base_ref_id=None),
        Row(id=2, branch_id=0, is_active=False, base_ref_id=None),
        Row(id=3, branch_id=7, is_active=True, base_ref_id=1),
    ]
    elements = [Row(id=10, branch_id=0, is_active=True, base_ref_id=None)]
    links = [
        Row(id=20, branch_id=0, paragraph_id=1, element_id=10),
        Row(id=21, branch_id=0, paragraph_id=2, element_id=10),
    ]
    base = repository._snapshot_for_branch(0, paragraphs, elements, links)
    assert [row["id"] for row in base["paragraphElements"]] == [20]
    overlay = repository._snapshot_for_branch(7, paragraphs, elements, links)
    assert overlay["paragraphs"][0]["rowId"] == 3
    assert overlay["paragraphs"][0]["effectiveId"] == 1


def test_repository_implementation_contains_no_write_or_commit_calls():
    source = inspect.getsource(ScriptBuildRepository)
    for forbidden in (".commit(", ".flush(", ".add(", ".delete(", ".execute("):
        assert forbidden not in source


def test_latest_split_event_body_fields_are_parsed_without_old_schema():
    repository = ScriptBuildRepository()
    body = Row(
        id=1,
        event_id=9,
        input_content_type="json",
        input_content='{"task": "查找案例"}',
        output_content_type="json",
        output_content='{"summary": "初筛完成"}',
    )
    parsed = repository._event_body(body)
    assert parsed["input"]["content"] == {"task": "查找案例"}
    assert parsed["output"]["content"] == {"summary": "初筛完成"}
    source = inspect.getsource(ScriptBuildRepository._event_body)
    assert "input_body_id" not in source
    assert "output_body_id" not in source


def test_oversized_event_body_is_safely_bounded_and_reports_omitted_characters():
    raw = '{"content":"' + ("x" * 250_000) + '"}'
    parsed = ScriptBuildRepository._body_side("json", raw)

    assert parsed["truncated"] is True
    assert parsed["omittedCharacters"] == len(raw) - 240_000
    assert "[truncated" in parsed["content"]
    assert len(parsed["content"]) < len(raw)


def test_main_view_reads_bodies_only_for_retrieval_evaluation_and_creative_agents():
    repository = ScriptBuildRepository()
    assert repository._needs_light_agent_body({"event_type": "agent_invoke", "event_name": "retrieve_data_knowledge"})
    assert repository._needs_light_agent_body({"event_type": "agent_invoke", "event_name": "script_multipath_evaluator"})
    assert repository._needs_light_agent_body({"event_type": "agent_invoke", "event_name": "script_implementer"})
    assert not repository._needs_light_agent_body({"event_type": "tool_call", "event_name": "retrieve_data_knowledge"})