from __future__ import annotations """R2 저장소/통합 RED. 근거: docs/design/07-tdd-red-system.md §2 R2, docs/design/02-data-model.md §3. 목표: DB 이력·스냅샷·이벤트·하루 1회 SUCCESS 가드가 실제 사용자 운영 흐름을 망가뜨리지 않는지 검증한다. """ import sqlite3 from pathlib import Path from typing import Callable import pytest from dmf_crawler import paths from dmf_crawler.models import DiffEvent, DiffResult, DmfRecord, FieldChange from dmf_crawler.storage import db, repo def _bootstrap_temp_db(tmp_path: Path) -> sqlite3.Connection: conn = db.connect(tmp_path / "dmf.sqlite3") db.apply_migrations(conn, paths.MIGRATIONS_DIR, tmp_path / "backup") return conn def test_migrations_apply_to_empty_db_and_are_checksum_idempotent(tmp_path: Path) -> None: """새 PC 최초 실행에서 0001~0003 이 모두 적용되고, 재실행은 무변화여야 한다.""" conn = db.connect(tmp_path / "dmf.sqlite3") try: applied = db.apply_migrations(conn, paths.MIGRATIONS_DIR, tmp_path / "backup") assert applied == [1, 2, 3, 4] versions = conn.execute( "SELECT version, name, checksum FROM schema_version ORDER BY version" ).fetchall() assert [(row["version"], row["name"]) for row in versions] == [ (1, "init"), (2, "enrichment"), (3, "ops"), (4, "report_views"), ] assert all(len(row["checksum"]) == 64 for row in versions) tables = set(db.table_names(conn)) assert { "runs", "stage_status", "fetch_stats", "quality_checks", "record_versions", "records", "snapshots", "events", "event_changes", "agy_calls", "enrichment_run", "enrichment", "component_health", "alerts", "alert_events", "watchlist", }.issubset(tables) assert conn.execute("SELECT COUNT(*) FROM sources WHERE source_id='mfds_open_api'").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM component_health").fetchone()[0] >= 2 second = db.apply_migrations(conn, paths.MIGRATIONS_DIR, tmp_path / "backup") assert second == [] assert conn.execute("SELECT COUNT(*) FROM schema_version").fetchone()[0] == 4 finally: conn.close() def test_repo_baseline_roundtrip_creates_snapshot_without_events( tmp_path: Path, record_factory: Callable[..., DmfRecord] ) -> None: """첫 성공일은 기준선이다: 스냅샷은 저장되지만 NEW 이벤트를 만들지 않는다.""" conn = _bootstrap_temp_db(tmp_path) try: run_id = "run_20260903_baseline" repo.start_run(conn, run_id, "2026-09-03", "manual") records = [ record_factory("20200101-1-A-1-1", ingredient_name="아토르바스타틴칼슘"), record_factory("20200102-2-A-2-1", ingredient_name="메트포르민염산염"), ] assert repo.insert_snapshot(conn, run_id, records) == 2 assert repo.insert_events(conn, run_id, DiffResult(baseline=True)) == 0 assert repo.upsert_current_records(conn, run_id, records) == 2 repo.finish_run( conn, run_id, "SUCCESS", 0, "기준선 수립", report_path="reports/DMF_2026-09-03.xlsx", is_baseline=True, integrity_ok=True, diff_performed=False, ) loaded = repo.load_snapshot(conn, run_id) assert [key for key, _ in loaded] == [r.dmf_key for r in records] assert conn.execute("SELECT COUNT(*) FROM events").fetchone()[0] == 0 assert conn.execute("SELECT COUNT(*) FROM v_current_records WHERE status='ACTIVE'").fetchone()[0] == 2 last = repo.last_success_run_on(conn, "2026-09-03") assert last is not None assert last.run_id == run_id assert last.is_baseline is True assert last.report_path == "reports/DMF_2026-09-03.xlsx" finally: conn.close() def test_repo_changed_new_withdrawn_events_roundtrip( tmp_path: Path, record_factory: Callable[..., DmfRecord] ) -> None: """둘째 날 신규/변경/취하 3종이 append-only 이벤트와 현재 상태에 함께 반영된다.""" conn = _bootstrap_temp_db(tmp_path) try: # Day 1: baseline with two active records. day1 = "run_20260903_baseline" original = record_factory("20200101-1-A-1-1", manufacturer="Old Factory") withdrawn = record_factory("20200102-2-A-2-1", manufacturer="Withdrawn Factory") repo.start_run(conn, day1, "2026-09-03", "manual") repo.insert_snapshot(conn, day1, [original, withdrawn]) repo.upsert_current_records(conn, day1, [original, withdrawn]) repo.finish_run(conn, day1, "SUCCESS", 0, "baseline", is_baseline=True, integrity_ok=True) # Day 2: original changed, withdrawn disappeared, and one new record arrived. day2 = "run_20260904_delta" changed = record_factory("20200101-1-A-1-1", manufacturer="New Factory") new = record_factory("20200103-3-A-3-1", manufacturer="Brand New Factory") manufacturer_change = FieldChange( field="manufacturer", label="제조소명", before="Old Factory", after="New Factory", change_class="HIGH", ) diff = DiffResult( new=( DiffEvent( dmf_key=new.dmf_key, permit_no=new.permit_no, event_type="NEW", severity="INFO", after=new.compare_view(), after_hash=new.content_hash, ), ), changed=( DiffEvent( dmf_key=changed.dmf_key, permit_no=changed.permit_no, event_type="CHANGED", severity="HIGH", changes=(manufacturer_change,), before=original.compare_view(), after=changed.compare_view(), before_hash=original.content_hash, after_hash=changed.content_hash, ), ), withdrawn=( DiffEvent( dmf_key=withdrawn.dmf_key, permit_no=withdrawn.permit_no, event_type="WITHDRAWN", severity="CRITICAL", before=withdrawn.compare_view(), before_hash=withdrawn.content_hash, ), ), unchanged_count=0, ) repo.start_run(conn, day2, "2026-09-04", "scheduled") assert repo.insert_snapshot(conn, day2, [changed, new]) == 2 assert repo.insert_events(conn, day2, diff) == 3 assert repo.upsert_current_records(conn, day2, [changed, new]) == 2 repo.finish_run(conn, day2, "SUCCESS", 0, "delta", integrity_ok=True, diff_performed=True) event_counts = dict( conn.execute( "SELECT event_type, COUNT(*) AS n FROM events GROUP BY event_type" ).fetchall() ) assert event_counts == {"CHANGED": 1, "NEW": 1, "WITHDRAWN": 1} assert conn.execute("SELECT COUNT(*) FROM event_changes WHERE field='manufacturer'").fetchone()[0] == 1 withdrawn_row = conn.execute( "SELECT status, withdrawn_date FROM records WHERE dmf_key = ?", (withdrawn.dmf_key,) ).fetchone() assert dict(withdrawn_row) == {"status": "WITHDRAWN", "withdrawn_date": "2026-09-04"} changed_row = conn.execute( "SELECT status, content_hash FROM v_current_records WHERE dmf_key = ?", (changed.dmf_key,) ).fetchone() assert dict(changed_row) == {"status": "ACTIVE", "content_hash": changed.content_hash} day2_snapshot_keys = [key for key, _ in repo.load_snapshot(conn, day2)] assert day2_snapshot_keys == [changed.dmf_key, new.dmf_key] finally: conn.close() def test_same_day_success_is_enforced_by_database_guard(tmp_path: Path) -> None: """앱 코드가 idempotency 검사를 빼먹어도 하루 SUCCESS 2건은 DB 가 막아야 한다.""" conn = _bootstrap_temp_db(tmp_path) try: repo.start_run(conn, "run_20260903_first", "2026-09-03", "scheduled") repo.finish_run(conn, "run_20260903_first", "SUCCESS", 0, "first") assert repo.last_success_run_on(conn, "2026-09-03") is not None repo.start_run(conn, "run_20260903_second", "2026-09-03", "startup") with pytest.raises(sqlite3.IntegrityError): repo.finish_run(conn, "run_20260903_second", "SUCCESS", 0, "should be blocked") rows = conn.execute( "SELECT run_id, status FROM runs WHERE run_date='2026-09-03' ORDER BY run_seq" ).fetchall() assert [(row["run_id"], row["status"]) for row in rows] == [ ("run_20260903_first", "SUCCESS"), ("run_20260903_second", "RUNNING"), ] finally: conn.close()