POSA_Copyrighter/tests/test_diagram_facts.py

"""다이어그램 spec 이 복제한 코드 유래 값이 실제 코드와 여전히 같은지 검사한다.

`test_diagram_sync.py` 가 "산출물이 spec 보다 낡았나"를 본다면, 이 파일은
"spec 이 코드보다 낡았나"를 본다. 그림에는 코드에서 뽑은 상수·버전·개수가
라벨과 카드 문장으로 하드카피돼 있는데, 그걸 지켜주는 장치가 없었다 —
SCORER_VERSION 을 v4 로 올리거나 믹스인을 하나 추가하면 코드는 통과하고
그림만 조용히 틀려진다.

같은 클래스의 드리프트를 테스트로 잡는 선례가 이미 이 저장소에 있다:
`tests/operator_gui/test_static_workbench.py` 의 APP_SCRIPT_NAMES.

실패하면 고칠 곳은 코드가 아니라 spec 이고, 고친 뒤에는
`scripts/rebuild-diagrams.ps1` 로 산출물과 수신증을 함께 갱신해야 한다.
"""

from __future__ import annotations

import json
from pathlib import Path

import pytest

from rights_filter.analysis.llm_assistance import DEFAULT_MODEL, MAX_PROMPT_EVIDENCE
from rights_filter.analysis.llm_prompts import PROMPT_VERSION
from rights_filter.analysis.risk_scoring import (
    SCORER_VERSION,
    WEB_ENTITY_MAX_POINTS,
    WEB_PRESENCE_MAX_POINTS,
    RiskScorer,
)
from rights_filter.domain.records import Evidence, EvidenceSource
from rights_filter.server.http_app import _MAX_BODY_BYTES
from rights_filter.server.auth_rate_limit import MAX_FAILURES_PER_WINDOW, WINDOW_SECONDS
from rights_filter.server.sqlite_store import CopyrighterStore
from rights_filter.server.store_schema import SCHEMA_VERSION

ROOT = Path(__file__).resolve().parents[1]
DIAGRAM_DIR = ROOT / "docs" / "diagrams"


def _spec(name: str) -> dict:
    return json.loads((DIAGRAM_DIR / name).read_text(encoding="utf-8"))


def _node(spec: dict, collection: str, node_id: str) -> dict:
    for node in spec[collection]:
        if node["id"] == node_id:
            return node
    raise AssertionError(
        f"spec 에 '{node_id}' 노드가 없다 — 노드를 rename 했다면 이 테스트도 함께 고친다."
    )


def _contains(haystack: str, needle: str, where: str) -> None:
    assert needle in haystack, (
        f"{where} 가 코드와 어긋났다: {needle!r} 를 담고 있어야 하는데 {haystack!r} 이다. "
        "spec 을 고치고 scripts/rebuild-diagrams.ps1 로 재생성한다."
    )


# --- architecture -----------------------------------------------------------


def test_architecture_store_mixin_count_matches_the_class():
    mixins = [base for base in CopyrighterStore.__bases__ if base.__name__.endswith("Mixin")]
    node = _node(_spec("system-architecture.architecture.json"), "components", "store")

    _contains(node["sublabel"], f"{len(mixins)}개", "CopyrighterStore 노드의 믹스인 개수")


def test_architecture_console_script_count_matches_the_folder():
    scripts = sorted((ROOT / "web" / "operator-gui").glob("app-*.js"))
    node = _node(_spec("system-architecture.architecture.json"), "components", "console")

    _contains(node["tag"], f"app-*.js {len(scripts)}", "운영자 콘솔 노드의 스크립트 개수")


def test_architecture_throttle_matches_the_rate_limit_constants():
    node = _node(_spec("system-architecture.architecture.json"), "components", "throttle")

    _contains(node["sublabel"], str(MAX_FAILURES_PER_WINDOW), "인증 스로틀 노드의 실패 허용 횟수")
    _contains(node["sublabel"], str(WINDOW_SECONDS), "인증 스로틀 노드의 윈도 길이")


def test_architecture_body_limit_matches_http_app():
    node = _node(_spec("system-architecture.architecture.json"), "components", "http")

    _contains(node["tag"], f"{_MAX_BODY_BYTES // (1024 * 1024)}MB", "http_app 노드의 바디 상한")


def test_architecture_schema_version_matches_store_schema():
    node = _node(_spec("system-architecture.architecture.json"), "components", "sqlite")

    _contains(node["tag"], f"SCHEMA v{SCHEMA_VERSION}", "SQLite 노드의 스키마 버전")


def test_architecture_scorer_version_matches_risk_scoring():
    node = _node(_spec("system-architecture.architecture.json"), "components", "analysis")

    _contains(node["tag"], f"RiskScorer {SCORER_VERSION}", "분석·채점 노드의 스코어러 버전")


def test_architecture_llm_model_matches_the_default():
    node = _node(_spec("system-architecture.architecture.json"), "components", "ollama")

    _contains(node["sublabel"], DEFAULT_MODEL, "로컬 LLM 노드의 모델명")


def test_architecture_cards_name_the_actual_runtime_dependencies():
    """'런타임 서드파티 3종' 이라고 써 두었으니 그 3종이 실제 목록과 같아야 한다."""
    requirements = [
        line.split("==")[0].split(">=")[0].strip().lower()
        for line in (ROOT / "requirements.txt").read_text(encoding="utf-8").splitlines()
        if line.strip() and not line.startswith("#")
    ]
    cards = " ".join(
        item for card in _spec("system-architecture.architecture.json")["cards"] for item in card["items"]
    ).lower()

    for package in requirements:
        _contains(cards, package, "아키텍처 카드의 런타임 의존성 목록")
    _contains(cards, f"{len(requirements)}종", "아키텍처 카드의 런타임 의존성 개수")


# --- dataflow ---------------------------------------------------------------


def test_dataflow_scorer_tag_matches_scorer_version():
    node = _node(_spec("evidence-dataflow.dataflow.json"), "nodes", "scorer")

    assert node["tag"] == SCORER_VERSION, (
        f"RiskScorer 노드 태그({node['tag']!r})가 SCORER_VERSION({SCORER_VERSION!r})과 다르다"
    )


def test_dataflow_prompt_version_tag_is_a_prefix_of_the_real_one():
    node = _node(_spec("evidence-dataflow.dataflow.json"), "nodes", "llm")

    assert PROMPT_VERSION.startswith(node["tag"]), (
        f"LLM 노드 태그({node['tag']!r})가 PROMPT_VERSION({PROMPT_VERSION!r})의 접두어가 아니다"
    )


def test_dataflow_prompt_evidence_cap_matches_the_constant():
    node = _node(_spec("evidence-dataflow.dataflow.json"), "nodes", "llm")

    _contains(node["sublabel"], f"{MAX_PROMPT_EVIDENCE}행", "LLM 노드의 프롬프트 입력 상한")


def test_dataflow_card_pool_ceilings_match_risk_scoring():
    cards = " ".join(item for card in _spec("evidence-dataflow.dataflow.json")["cards"] for item in card["items"])

    _contains(cards, f"presence {WEB_PRESENCE_MAX_POINTS}", "데이터흐름 카드의 presence 풀 상한")
    _contains(cards, f"entity {WEB_ENTITY_MAX_POINTS}", "데이터흐름 카드의 entity 풀 상한")


def test_dataflow_knowledge_edge_matches_actual_fingerprint_scoring():
    """라벨이 '지문 유사도 0.9+ → 80점' 이라고 주장하니 실제로 그런지 채점해 본다.

    이 둘은 이름 있는 상수가 아니라 risk_scoring 안의 인라인 리터럴이라,
    import 대신 행위로 확인하는 것이 정확하다.
    """
    spec = _spec("evidence-dataflow.dataflow.json")
    label = next(flow["label"] for flow in spec["flows"] if flow["id"] == "fingerprint-match")

    result = RiskScorer().score(
        [
            Evidence(
                source=EvidenceSource.FINGERPRINT,
                reason="지식DB 등록 이미지와 일치",
                confidence=1.0,
                data={"similarity": 0.9},
            )
        ]
    )

    _contains(label, "0.9", "지식DB 엣지의 유사도 임계값")
    _contains(label, str(result.score), "지식DB 엣지의 점수")


# --- workflow ---------------------------------------------------------------


def test_workflow_scorer_tag_matches_scorer_version():
    node = _node(_spec("operator-review-workflow.workflow.json"), "nodes", "score")

    assert node["tag"] == SCORER_VERSION, (
        f"위험도 산출 노드 태그({node['tag']!r})가 SCORER_VERSION({SCORER_VERSION!r})과 다르다"
    )


def test_workflow_llm_evidence_cap_matches_the_constant():
    node = _node(_spec("operator-review-workflow.workflow.json"), "nodes", "llm")

    _contains(node["sublabel"], f"{MAX_PROMPT_EVIDENCE}행", "LLM 노드의 프롬프트 입력 상한")


@pytest.mark.parametrize(
    "spec_name, collection",
    [
        ("system-architecture.architecture.json", "components"),
        ("evidence-dataflow.dataflow.json", "nodes"),
        ("operator-review-workflow.workflow.json", "nodes"),
        ("submission-lifecycle.lifecycle.json", "states"),
    ],
)
def test_node_ids_are_unique_within_a_spec(spec_name: str, collection: str):
    """위 조회들이 첫 일치만 보므로, id 중복은 조용한 오탐이 된다."""
    ids = [node["id"] for node in _spec(spec_name)[collection]]

    assert len(ids) == len(set(ids)), f"{spec_name} 에 중복 id 가 있다: {ids}"