"""다이어그램 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}"