"""★★★사람이 정한 것이 **정본**이고 CP 는 그것의 투영이다 (Codex 계약 2026-09-02). `artifact/` 의 HTML·JSON 은 지워도 되는 화면 산출이지 결정이 아니다. 결정은 **입력 신원**(사진 SHA · acquisition identity · 요구 좌표)에 결속되고, 하나라도 바뀌면 옛 결정은 **안 따라간다**. """ from __future__ import annotations import pytest from app.modules.pipeline import grounding_fidelity_review as fr from app.modules.pipeline import reference_acquisition as ra BASE = {"project_id": "p", "episode_id": "e", "research_subject_id": "rs_x", "final_id": "LP01", "purpose": "detail", "acquisition_identity": "acq-1", "image_sha256": "sha-aaa", "era": "가나다 무렵", "region": "라마바 지방", "source_url": "https://x.invalid/1.jpg", "query_provenance_digest": "dig-1"} def _row(h, verdict=fr.VERDICT_VERIFIED, *, rid="r1", at="2026-09-02T10:00", supersedes=None): return {"id": rid, "review_input_hash": h, "verdict": verdict, "reviewed_at": at, "reviewer_actor": "사람", "supersedes_id": supersedes} class TestTheDecisionIsBoundToWhatWasSeen: def test_every_identity_field_changes_the_hash(self): base = fr.input_hash(BASE) for axis, other in (("image_sha256", "sha-bbb"), ("acquisition_identity", "acq-2"), ("era", "다른 시기"), ("region", "다른 곳"), ("purpose", "context"), ("source_url", "https://y.invalid/2.jpg"), ("query_provenance_digest", "dig-2")): assert fr.input_hash({**BASE, axis: other}) != base, axis def test_an_absent_coordinate_is_not_invented(self): """★시대 선언이 **없는** 원고와 있는 원고는 다른 신원이다.""" got = fr.canonical_input({**BASE, "era": ""}) assert '"era":""' in got assert fr.input_hash({**BASE, "era": ""}) != fr.input_hash(BASE) def test_a_missing_identity_field_stops(self): for axis in ("project_id", "research_subject_id", "acquisition_identity", "image_sha256"): with pytest.raises(fr.ReviewInputBroken, match=axis): fr.input_hash({**BASE, axis: ""}) def test_the_canonical_bytes_are_stable(self): a = fr.canonical_input(BASE) b = fr.canonical_input(dict(reversed(list(BASE.items())))) assert a == b, "★칸 순서가 신원을 바꾼다" class TestWhatTheCheckpointGetsIsAProjection: def test_no_decision_is_unverified(self): got = fr.fidelity_state([], want_hash=fr.input_hash(BASE)) assert got["state"] == ra.FIDELITY_UNVERIFIED assert got["fault"] def test_a_decision_for_another_input_does_not_follow(self): """★★사진이 바뀌면 옛 승인이 **안 따라간다**.""" old = fr.input_hash(BASE) rows = [_row(old, fr.VERDICT_VERIFIED)] now = fr.input_hash({**BASE, "image_sha256": "sha-bbb"}) assert fr.fidelity_state(rows, want_hash=now)["state"] == \ ra.FIDELITY_UNVERIFIED assert fr.fidelity_state(rows, want_hash=old)["state"] == \ ra.FIDELITY_VERIFIED def test_rejected_is_not_usable_either(self): h = fr.input_hash(BASE) got = fr.fidelity_state([_row(h, fr.VERDICT_REJECTED)], want_hash=h) assert got["state"] == ra.FIDELITY_REJECTED # ★HITL 0: production 의 붙는 조건은 판정 칸을 **안 읽는다** — 거절은 도구 # (`apply_reviews`)가 outcome 을 `reference_unavailable` 로 옮겨서만 반영된다 assert ra.usable_as_reference({"outcome": ra.STATUS_SELECTED, "grounding_fidelity": got}) assert not ra.usable_as_reference({"outcome": ra.STATUS_UNAVAILABLE, "grounding_fidelity": got}) def test_only_verified_is_usable(self): h = fr.input_hash(BASE) got = fr.fidelity_state([_row(h)], want_hash=h) assert ra.usable_as_reference( {"outcome": ra.STATUS_SELECTED, "grounding_fidelity": got}) def test_an_unknown_verdict_folds_to_unverified(self): h = fr.input_hash(BASE) got = fr.fidelity_state([_row(h, "무언가")], want_hash=h) assert got["state"] == ra.FIDELITY_UNVERIFIED and got["fault"] class TestCorrectionsAreNewRowsNotEdits: def test_a_superseded_row_dies(self): h = fr.input_hash(BASE) rows = [_row(h, fr.VERDICT_VERIFIED, rid="r1"), _row(h, fr.VERDICT_REJECTED, rid="r2", at="2026-09-02T11:00", supersedes="r1")] got = fr.fidelity_state(rows, want_hash=h) assert got["state"] == ra.FIDELITY_REJECTED assert got["review_id"] == "r2" def test_two_conflicting_live_rows_fail_safe(self): """★★★「둘 중 아무거나」로 고르면 사람이 안 정한 것을 정한 것으로 만든다 — 미확인 + 감사 결함이다.""" h = fr.input_hash(BASE) rows = [_row(h, fr.VERDICT_VERIFIED, rid="r1"), _row(h, fr.VERDICT_REJECTED, rid="r2", at="2026-09-02T11:00")] got = fr.fidelity_state(rows, want_hash=h) assert got["state"] == ra.FIDELITY_UNVERIFIED assert "모순" in got["fault"] def test_two_agreeing_live_rows_are_fine(self): h = fr.input_hash(BASE) rows = [_row(h, rid="r1"), _row(h, rid="r2", at="2026-09-02T11:00")] assert fr.fidelity_state(rows, want_hash=h)["state"] == \ ra.FIDELITY_VERIFIED class TestTheServerRefusesWhatWasNotSeen: """★화면이 옛 사진을 보여 주고 서버가 새 사진에 결정을 붙이면, 사람이 **안 본 것**을 승인한 것이 된다.""" def test_a_different_sha_stops(self): with pytest.raises(fr.ReviewInputBroken, match="image_sha256"): fr.assert_observed_matches( BASE, {"image_sha256": "sha-bbb", "era": BASE["era"], "region": BASE["region"]}) def test_a_different_coordinate_stops(self): with pytest.raises(fr.ReviewInputBroken, match="region"): fr.assert_observed_matches( BASE, {"image_sha256": BASE["image_sha256"], "era": BASE["era"], "region": "다른 곳"}) def test_the_same_thing_passes(self): fr.assert_observed_matches( BASE, {"image_sha256": BASE["image_sha256"], "era": BASE["era"], "region": BASE["region"]}) def test_nothing_observed_stops(self): with pytest.raises(fr.ReviewInputBroken): fr.assert_observed_matches(BASE, None) class TestTheTableIsTheRecord: def test_it_is_its_own_table(self): from app.models.project import GroundingReferenceFidelityReview as M assert M.__tablename__ == "grounding_reference_fidelity_review" cols = {c.name for c in M.__table__.columns} for want in ("review_input_json", "review_input_hash", "verdict", "reviewer_actor", "observed_json", "supersedes_id", "idempotency_key"): assert want in cols, want def test_the_same_send_twice_is_one_row(self): from app.models.project import GroundingReferenceFidelityReview as M uq = [c for c in M.__table__.constraints if c.__class__.__name__ == "UniqueConstraint"] assert uq assert {c.name for c in list(uq)[0].columns} == { "project_id", "episode_id", "idempotency_key"} def test_the_migration_is_additive_only(self): from pathlib import Path src = (Path(__file__).resolve().parents[2] / "alembic" / "versions" / "012_grounding_fidelity_review.py" ).read_text(encoding="utf-8") for banned in ("drop_column", "alter_column", 'op.execute("UPDATE', 'op.execute("DELETE'): assert banned not in src, banned for want in ("review_input_hash", "verdict", "reviewer_actor", "supersedes_id", "idempotency_key"): assert want in src, want def test_production_schema_path_is_closed_too(self): """★canary `init_db` 만이 아니라 **alembic head** 도 이 표를 안다.""" from pathlib import Path vers = (Path(__file__).resolve().parents[2] / "alembic" / "versions") heads = {p.stem for p in vers.glob("*.py")} assert "012_grounding_fidelity_review" in heads chain = {} for p in vers.glob("*.py"): src = p.read_text(encoding="utf-8") for line in src.splitlines(): if line.startswith("down_revision"): chain[p.stem] = line.split("=")[-1].strip().strip('"\' ') assert chain.get("012_grounding_fidelity_review") == \ "011_grounding_research_revision", "★사슬이 안 이어졌다" class TestTheOnlyDoorIsTheAuthenticatedEndpoint: """★★★결정은 **인증된 backend** 를 지나 DB 로만 들어온다. HTML·CP 파일 직접 수정, verdict 가 든 산출 JSON 업로드는 **금지**다 (Codex 계약 2026-09-02). """ def test_the_client_cannot_send_the_reviewer(self): """★★판정자는 **서버의 인증 주체**에서 온다.""" from app.api.v1.grounding_fidelity import VerdictIn assert "reviewer_actor" not in VerdictIn.model_fields, ( "★클라이언트가 판정자를 보낼 수 있다") assert "reviewed_at" not in VerdictIn.model_fields assert "review_contract_version" not in VerdictIn.model_fields def test_the_client_cannot_send_the_identity_either(self): """★★★앞 판은 canonical payload 를 **body 에서 통째로** 만들고 `observed` 도 같은 body 와만 견줬다 — **항진식**이다 (Codex BLOCK). 지어낸 입력과 지어낸 `observed` 를 함께 보내면 정본 표에 그대로 들어갔다. 이제 서버가 얼어붙은 것에서 다시 만든다. """ from app.api.v1.grounding_fidelity import VerdictIn sent = set(VerdictIn.model_fields) for banned in ("image_sha256", "era", "region", "source_url", "acquisition_identity", "research_subject_id", "final_id", "purpose", "query_provenance_digest"): assert banned not in sent, f"★{banned} 를 클라이언트가 보낸다" assert "review_input_hash" in sent, "★신원 열쇠는 있어야 찾는다" def test_the_server_rebuilds_the_payload_from_frozen_things(self): import inspect from app.api.v1 import grounding_fidelity as api src = inspect.getsource(api.record_verdict) assert "_one_candidate(" in src, "★서버가 다시 안 만든다" assert 'payload = dict(one["payload"])' in src assert "body.image_sha256" not in src and "body.era" not in src def test_the_photo_sha_comes_from_the_file_now(self): import inspect from app.api.v1 import grounding_fidelity as api src = inspect.getsource(api._candidates) # ★2026-09-02: 파일 bytes 의 SHA 를 내되, 이제는 사이드카·CP 판정과 **같은 helper** # (`grounding_sidecar_writer.row_content_sha256`)로 낸다 — 화면에서 본 bytes 와 # 붙는 bytes 가 한 함수에서 나온다 assert "row_content_sha256(" in src, "★지금 파일에서 SHA 를 안 낸다" from app.modules.pipeline import grounding_sidecar_writer as sw assert "hashlib.sha256()" in inspect.getsource(sw.row_content_sha256) def test_writing_needs_more_than_read_access(self): """★★읽기 권한으로 파이프라인 입력을 못 바꾼다 (Codex BLOCK).""" import inspect from app.api.v1 import grounding_fidelity as api assert "require_project_write" in inspect.getsource(api.record_verdict) gate = inspect.getsource(api.require_project_write) assert "owner" in gate and "admin" in gate for fn in (api.review_page, api.photo, api.list_verdicts): assert "require_project_write" not in inspect.getsource(fn), ( f"★{fn.__name__} 은 읽기인데 쓰기 권한을 요구한다") def test_the_server_takes_the_actor_from_the_session(self): import inspect from app.api.v1 import grounding_fidelity as api src = inspect.getsource(api.record_verdict) assert "current_user" in src and "reviewer_actor=" in src assert "body.reviewer_actor" not in src def test_it_checks_what_the_person_actually_saw(self): import inspect from app.api.v1 import grounding_fidelity as api src = inspect.getsource(api.record_verdict) assert "assert_observed_matches" in src i_obs = src.index("assert_observed_matches") i_add = src.index("db.add(") assert i_obs < i_add, "★적고 나서 확인하면 늦다" def test_it_never_updates_or_deletes(self): import inspect from app.api.v1 import grounding_fidelity as api # ★★글자로 막으면 **엉뚱한 것**이 걸린다 — `hashlib` 의 # `h.update(chunk)` 가 잡혔다(오늘 세 번째다). ORM 쓰기 자리를 # **AST 로** 본다: `<무엇>.delete(...)` · `<무엇>.update(...)` 중 # 해시 객체가 아닌 것. import ast src = inspect.getsource(api) tree = ast.parse(src) bad = [] for n in ast.walk(tree): if not isinstance(n, ast.Call): continue attr = getattr(n.func, "attr", "") if attr not in ("delete", "update"): continue owner = getattr(getattr(n.func, "value", None), "id", "") if owner == "h": # hashlib 누적 continue bad.append(f"{owner}.{attr}") assert bad == [], f"★{bad} — append-only 가 아니다" for banned in ('op.execute("UPDATE', 'text("UPDATE', 'text("DELETE'): assert banned not in src, banned def test_the_verdict_list_is_closed(self): import inspect from app.api.v1 import grounding_fidelity as api src = inspect.getsource(api.record_verdict) assert "KNOWN_VERDICTS" in src, "★모르는 판정이 들어간다" def test_both_routes_are_registered(self): from app.main import create_app paths = {r.path for r in create_app().routes if hasattr(r, "path")} assert "/api/v1/grounding-fidelity/verdict" in paths assert "/api/v1/grounding-fidelity/verdicts" in paths def test_every_route_requires_a_user(self): import inspect from app.api.v1 import grounding_fidelity as api for fn in (api.record_verdict, api.list_verdicts, api.review_page, api.photo): sig = inspect.signature(fn) assert "current_user" in sig.parameters, fn.__name__ src = inspect.getsource(fn) # ★쓰기 문은 **읽기 문을 안에서 부른다** — 둘 중 하나면 된다 assert ("verify_project_access" in src or "require_project_write" in src), fn.__name__ class TestTheProjectionReadsTheDatabase: """★CP 는 **투영**이다 — 사람이 CP 를 직접 고치는 길은 없다.""" @staticmethod def _cp(): return {"data": {"rows": [{ "research_subject_id": "rs_x", "identity": "acq-1", "outcome": "selected", "ledger_row": {"final_id": "LP01", "purpose": "detail"}, "acquisition": {"chosen": {"url": "https://x.invalid/1.jpg", "path": "refs/a.png"}, "rounds": []}}]}} def _one(self): got = fr.review_inputs_for(self._cp(), project_id="p", episode_id="e", coordinates={"era": "가", "region": "나"}, sha_of=lambda _r: "sha-aaa") assert len(got) == 1 return got[0] def test_a_verified_row_becomes_verified_in_the_checkpoint(self): one = self._one() got = fr.apply_reviews( self._cp(), [_row(one["hash"], fr.VERDICT_VERIFIED)], project_id="p", episode_id="e", coordinates={"era": "가", "region": "나"}, sha_of=lambda _r: "sha-aaa") row = got["data"]["rows"][0] assert row["grounding_fidelity"]["state"] == ra.FIDELITY_VERIFIED assert ra.usable_as_reference(row) def test_a_different_photo_does_not_inherit_the_verdict(self): one = self._one() got = fr.apply_reviews( self._cp(), [_row(one["hash"], fr.VERDICT_VERIFIED)], project_id="p", episode_id="e", coordinates={"era": "가", "region": "나"}, sha_of=lambda _r: "sha-bbb") # ★사진이 바뀌었다 row = got["data"]["rows"][0] assert row["grounding_fidelity"]["state"] == ra.FIDELITY_UNVERIFIED assert ra.usable_as_reference(row) # ★HITL 0 — 미확인이어도 자동 선택은 붙는다 def test_a_different_coordinate_does_not_inherit_it_either(self): one = self._one() got = fr.apply_reviews( self._cp(), [_row(one["hash"], fr.VERDICT_VERIFIED)], project_id="p", episode_id="e", coordinates={"era": "다른 시기", "region": "나"}, sha_of=lambda _r: "sha-aaa") assert got["data"]["rows"][0]["grounding_fidelity"]["state"] == \ ra.FIDELITY_UNVERIFIED def test_the_declared_axes_must_agree_with_what_went_out(self): """★그때 요구된 축과 지금 좌표가 다르면 **안 적는다**.""" cp = self._cp() cp["data"]["rows"][0]["acquisition"]["rounds"] = [ {"round_no": 1, "coordinates_carried": {"declared": ["era", "region"]}}] with pytest.raises(fr.ReviewInputBroken, match="그때 요구된 축"): fr.review_inputs_for(cp, project_id="p", episode_id="e", coordinates={"era": "가", "region": ""}, sha_of=lambda _r: "sha-aaa") class TestThePageComesFromTheSameDoor: """★★★화면을 정적 서버에 두고 다른 포트의 API 로 보내면 브라우저가 **CORS 로 막는다** — 이 판에는 CORS 미들웨어가 아예 없다 (실측 09-02). 그렇다고 CORS 를 여는 것은 이 화면 하나 때문에 바꿀 일이 아니다. 같은 문에서 내주면 세션 쿠키가 그대로 간다. """ def test_the_backend_serves_the_page_itself(self): from app.main import create_app paths = {r.path for r in create_app().routes if hasattr(r, "path")} assert "/api/v1/grounding-fidelity/review-page" in paths assert "/api/v1/grounding-fidelity/photo" in paths def test_the_page_posts_to_its_own_prefix(self): import inspect from app.api.v1 import grounding_fidelity as api src = inspect.getsource(api.review_page) assert "api_base=str(router.prefix)" in src, ( "★다른 origin 으로 보내면 브라우저가 막는다") def test_the_photo_route_opens_only_a_candidate(self): """★★★앞 판은 임의 `path` 를 받아 뿌리 **부모** 아래를 열어 줬다 — 프로젝트 하나에 권한만 있으면 같은 뿌리의 다른 프로젝트·장부·산출까지 볼 수 있었다 (Codex BLOCK). 이제 열쇠는 **판정 신원**이다. """ import inspect from app.api.v1 import grounding_fidelity as api sig = inspect.signature(api.photo) assert "path" not in sig.parameters, "★임의 경로를 아직 받는다" assert "key" in sig.parameters src = inspect.getsource(api.photo) assert "_one_candidate(" in src assert "_sha_of_file(want)" in src, "★내용이 신원과 같은지 안 본다" assert "photo_changed" in src def test_the_page_keys_photos_by_identity(self): import inspect from app.api.v1 import grounding_fidelity as api src = inspect.getsource(api.review_page) assert "&key=" in src assert 'key_of=lambda x: x["hash"]' in src, "★raw 경로를 아직 붙인다" def test_a_photo_that_will_not_load_disables_the_buttons(self): """★★서버 렌더 때 `is_file()` 만 보면 HTTP 로 403/404 가 나도 단추가 살아 있다 — 브라우저에서 **실제로 안 뜨면** 그때 끈다.""" import inspect from tools.grounding_audit import fidelity_review_page as pg src = inspect.getsource(pg.render) assert "onerror=" in src and "brokenPhoto" in src got = pg.render([{"payload": {"final_id": "X", "purpose": "p", "source_url": "u", "era": "", "region": "", "research_subject_id": "r"}, "hash": "h", "key": "h", "exists": True, "sha": "s", "kind": "K", "said": "", "anchor": "", "brief": "", "owner": "prop", "same_as": [], "terms": [], "provenance": []}], api_base="/x", image_base="/x/photo?key=", coords={"era": "", "region": ""}) assert "b.disabled=true" in got and 'onerror="brokenPhoto(this)"' in got def test_both_new_routes_need_a_user(self): import inspect from app.api.v1 import grounding_fidelity as api for fn in (api.review_page, api.photo): assert "current_user" in inspect.signature(fn).parameters assert "verify_project_access" in inspect.getsource(fn) def test_it_only_reads_frozen_files(self): import inspect from app.api.v1 import grounding_fidelity as api src = inspect.getsource(api._frozen) assert "write" not in src and "mkdir" not in src class TestTheRevisionIdFitsTheColumn: """★★★`alembic_version.version_num` 이 `varchar(32)` 다 (실측 09-02). 39자로 지었더니 **표는 만들고 판 기록에서 죽었다** — 그 판은 롤백돼서 표도 안 남았다. 이름 길이가 배포를 막는 자리다. """ def test_every_revision_id_is_short_enough(self): from pathlib import Path vers = Path(__file__).resolve().parents[2] / "alembic" / "versions" long = [] for p in vers.glob("*.py"): for line in p.read_text(encoding="utf-8").splitlines(): if line.startswith("revision: str"): rid = line.split("=")[-1].strip().strip('"\' ') if len(rid) > 32: long.append((p.name, rid, len(rid))) assert long == [], f"★32자를 넘는 revision id: {long}" class TestTheLoginFormGoesOnlyWhereItShould: """★★HTML escape 를 **JS 문자열**에 쓰면 안 된다 (실측 09-02): `&` 가 `&` 가 되어 redirect 주소가 깨졌다. escape 로 씨름하지 말고 **허용 목록**으로 본다. """ def test_it_keeps_the_query_string_intact(self): from app.api.v1 import grounding_fidelity as api want = (f"{api.router.prefix}/review-page?project_id=a&episode_id=b") got = api.login_page(next=want).body.decode("utf-8") assert "&" not in got, "★redirect 주소가 깨진다" assert want in got def test_it_refuses_somewhere_else(self): import pytest from app.api.v1 import grounding_fidelity as api from app.core.errors import AppError for bad in ("https://evil.invalid/x", "/api/v1/users", f"{api.router.prefix}/x\"