`tests/test_provenance.py`

129 lines
"""acquisition_method provenance field (Cody must-fix, TK-12199): migration, backfill, API, exports, viewer badge."""
import csv
import json
import sqlite3

import pytest
from fastapi.testclient import TestClient

from conrad import config, db, exports, provenance
from conrad.models import CartoonRecord

SEEDS = [  # (source_id, canonical prefix, provenance, granularity)
    ("seed_wichita", "wsu", "seed:research/raw/wsu/112075.md", "item"),
    ("seed_syracuse", "syr", "seed:research/corpus.json#syracuse", "folder"),
    ("seed_loc", "loc", "seed:research/raw/loc-page-1.json", "item"),
    ("seed_catalog", "cat", "seed:data/cartoons.json", "item"),
    ("loc", "loc", "live:item-json -> https://www.loc.gov/pictures/item/9/?fo=json", "item"),
    ("huntington", "hunt", "live:https://cinco-prd.s3.amazonaws.com/media/ead/conrad.xml", "box_range"),
]


def _populate(conn):
    for i, (sid, pre, prov, gran) in enumerate(SEEDS):
        db.upsert_source(conn, sid, sid)
        rec = CartoonRecord(canonical_id=f"{pre}:{i}", identifier=str(i), granularity=gran, title=f"Cartoon {sid}",
                            year=1970, date_start="1970-01-01", date_end="1970-12-31", repository=f"Repo {sid}",
                            record_url=f"https://example.org/{sid}/{i}", provenance=prov)
        db.save_record(conn, rec, sid)
    conn.commit()


def test_derive_rules():
    assert provenance.derive("seed_wichita") == "seed_via_reader_bypass"
    assert provenance.derive("seed_syracuse") == "seed_via_reader_bypass"
    assert provenance.derive("seed_loc") == "seed_direct"
    assert provenance.derive("seed_huntington") == "seed_direct"
    assert provenance.derive("seed_catalog") == "secondary_citation"
    assert provenance.derive("loc") == "direct_api"
    assert provenance.derive("huntington", "live:https://cinco-prd.s3.amazonaws.com/media/ead/conrad.xml") == "direct_html"
    # a reader-proxied fetch is never labelled direct, whatever source claims it
    assert provenance.derive("wichita", "live:https://r.jina.ai/https://archivesspace.wichita.edu/x") == "seed_via_reader_bypass"


def test_save_record_sets_method_and_constraint(tmpdb):
    _populate(tmpdb)
    got = dict(tmpdb.execute("SELECT source_id, acquisition_method FROM cartoon_sources").fetchall())
    assert got["seed_wichita"] == got["seed_syracuse"] == "seed_via_reader_bypass"
    assert got["loc"] == "direct_api" and got["seed_catalog"] == "secondary_citation"
    with pytest.raises(sqlite3.IntegrityError):  # closed vocabulary
        tmpdb.execute("UPDATE cartoon_sources SET acquisition_method='scraped_somehow' WHERE source_id='loc'")


def test_migration_is_idempotent_on_old_db(tmp_path):
    """A DB created by the pre-TK-12199 schema (no column) gains + backfills it; re-running changes nothing."""
    p = tmp_path / "old.db"
    old_schema = db.SCHEMA.replace(
        """  acquisition_method TEXT CHECK (acquisition_method IS NULL OR acquisition_method IN
    ('direct_api','direct_html','seed_direct','seed_via_reader_bypass','secondary_citation')),
""", "")
    assert "acquisition_method" not in old_schema
    conn = db.connect(p)
    conn.executescript(old_schema)
    for i, (sid, pre, prov, gran) in enumerate(SEEDS):
        conn.execute("INSERT INTO sources(id,name) VALUES (?,?)", (sid, sid))
        cid = conn.execute("INSERT INTO cartoons(canonical_id,granularity,created_at,updated_at) VALUES (?,?,'x','x')",
                           (f"{pre}:{i}", gran)).lastrowid
        conn.execute("INSERT INTO cartoon_sources(cartoon_id,source_id,identifier,provenance) VALUES (?,?,?,?)",
                     (cid, sid, str(i), prov))
    conn.commit()
    db.init_db(conn)
    first = conn.execute("SELECT source_id, acquisition_method FROM cartoon_sources ORDER BY id").fetchall()
    db.init_db(conn)  # second run: no error, no change
    second = conn.execute("SELECT source_id, acquisition_method FROM cartoon_sources ORDER BY id").fetchall()
    assert [tuple(r) for r in first] == [tuple(r) for r in second]
    assert conn.execute("SELECT COUNT(*) FROM cartoon_sources WHERE acquisition_method IS NULL").fetchone()[0] == 0
    assert dict(first)["seed_wichita"] == "seed_via_reader_bypass"


def test_api_exposes_method_and_flags(tmpdb, monkeypatch):
    monkeypatch.delenv("CONRAD_BASIC_USER", raising=False)
    monkeypatch.delenv("CONRAD_REQUIRE_AUTH", raising=False)
    _populate(tmpdb)
    from conrad.web.app import create_app
    c = TestClient(create_app())
    res = c.get("/api/search", params={"granularity": "all", "limit": 50}).json()["results"]
    by = {r["canonical_id"].split(":")[0]: r for r in res}
    assert by["wsu"]["acquisition_methods"] == ["seed_via_reader_bypass"]
    assert "pending_direct_verification" in by["wsu"]["provenance_flags"]
    assert "secondary_citation_only" in by["cat"]["provenance_flags"]
    wsu_id = by["wsu"]["id"]
    d = c.get(f"/api/cartoon/{wsu_id}").json()
    assert d["sources"][0]["acquisition_method"] == "seed_via_reader_bypass"
    assert "acquired_via_reader_bypass" in d["provenance_flags"]
    page = c.get(f"/cartoon/{wsu_id}").text
    assert provenance.BADGE_TEXT in page and 'data-flag="acquired_via_reader_bypass"' in page
    # a directly fetched record carries no bypass badge
    loc_ids = [r["id"] for r in res if r["canonical_id"].startswith("loc:")]
    for i in loc_ids:
        assert provenance.BADGE_TEXT not in c.get(f"/cartoon/{i}").text


def test_exports_include_method(tmpdb, tmp_path, monkeypatch):
    monkeypatch.setattr(config, "EXPORT_DIR", tmp_path)
    monkeypatch.setattr(config, "SOURCES_JSON", tmp_path / "sources.json")
    _populate(tmpdb)
    exports.export_all(tmpdb)
    src = list(csv.DictReader((tmp_path / "sources.csv").open()))
    assert src and all(r["acquisition_method"] for r in src)
    assert {r["acquisition_method"] for r in src if r["source_id"] in ("seed_wichita", "seed_syracuse")} == \
        {"seed_via_reader_bypass"}
    cart = list(csv.DictReader((tmp_path / "cartoons.csv").open()))
    assert "acquisition_methods" in cart[0] and "provenance_flags" in cart[0]
    wsu = next(r for r in cart if r["canonical_id"].startswith("wsu:"))
    assert wsu["acquisition_methods"] == "seed_via_reader_bypass" and "pending_direct_verification" in wsu["provenance_flags"]
    js = json.loads((tmp_path / "cartoons.json").read_text())
    assert all("provenance_flags" in r and all("acquisition_method" in s for s in r["sources"]) for r in js)


@pytest.mark.skipif(not (config.ROOT / "data/conrad.db").exists(), reason="project DB not built")
def test_live_db_every_row_has_method_and_bypass_rows_flagged():
    conn = db.connect(config.ROOT / "data/conrad.db")
    db.init_db(conn)
    assert conn.execute("SELECT COUNT(*) FROM cartoon_sources WHERE acquisition_method IS NULL").fetchone()[0] == 0
    bad = conn.execute("""SELECT COUNT(*) FROM cartoon_sources WHERE source_id IN ('seed_wichita','seed_syracuse')
                          AND acquisition_method != 'seed_via_reader_bypass'""").fetchone()[0]
    assert bad == 0
    assert conn.execute("SELECT COUNT(*) FROM cartoon_sources WHERE source_id IN ('seed_wichita','seed_syracuse')"
                        ).fetchone()[0] > 0