stack/tests/zot/fetch/test_run.py

"""zot.fetch.run — the per-item loop (scratch Zotero DB, no network)."""

from __future__ import annotations

import shutil
import sqlite3
from pathlib import Path

import pytest

from tests.zot.fetch.conftest import mk_item
from zot.db import Db
from zot.fetch.ledger import Ledger
from zot.fetch.run import run, safe_name
from zot.fetch.select import item_ref

PDF = b"%PDF-1.7\n" + b"x" * 2048


def _copier(src: Path, dest_dir: Path) -> None:
    dest_dir.mkdir(parents=True, exist_ok=True)
    shutil.copy2(src, dest_dir / src.name)


def _finder_writing(tmp_path: Path, source: str = "unpaywall"):
    def finder(client, ref, *, email, scratch):
        p = scratch / f"{ref.key}.pdf"
        p.write_bytes(PDF)
        return source, p

    return finder


def _go(zdb: Db, tmp_path: Path, refs, **kw):
    lines: list[str] = []
    kw.setdefault("copier", _copier)
    stats = run(
        zdb,
        refs,
        ledger=kw.pop("ledger", Ledger(tmp_path / "ledger.jsonl")),
        client=None,
        email="e@x",
        storage_dir=tmp_path / "storage",
        scratch_dir=tmp_path / "scratch",
        out=lines.append,
        **kw,
    )
    return stats, lines


def test_attaches_an_open_access_pdf_and_ledgers_it(zdb: Db, tmp_path: Path):
    item_id = mk_item(zdb, title="Risk adjustment: a review", doi="10.1/a")
    ref = item_ref(zdb, item_id)
    stats, lines = _go(zdb, tmp_path, [ref], finder=_finder_writing(tmp_path, "pmc"))
    assert stats["attached"] == 1 and stats["attempted"] == 1
    row = zdb.con.execute(
        "SELECT contentType, path FROM itemAttachments WHERE parentItemID = ?",
        (item_id,),
    ).fetchone()
    assert tuple(row) == ("application/pdf", "storage:Risk adjustment a review.pdf")
    assert "source:pmc" in zdb.get_tags(item_id)
    assert list((tmp_path / "storage").rglob("*.pdf"))  # copied into storage
    assert not list((tmp_path / "scratch").iterdir())  # scratch cleaned up
    assert "attached  pmc" in lines[-1]
    led = Ledger(tmp_path / "ledger.jsonl")
    assert not led.should_try(ref.key)  # attached items are never retried


def test_no_open_access_copy_is_a_terminal_miss(zdb: Db, tmp_path: Path):
    ref = item_ref(zdb, mk_item(zdb, title="Paywalled", doi="10.1/p"))
    stats, lines = _go(zdb, tmp_path, [ref], finder=lambda c, r, **k: None)
    assert stats["no_candidate"] == 1
    assert "no open-access copy" in lines[-1]
    assert not Ledger(tmp_path / "ledger.jsonl").should_try(ref.key)


def test_a_failing_item_is_rolled_back_and_the_run_goes_on(zdb: Db, tmp_path: Path):
    bad = item_ref(zdb, mk_item(zdb, title="Bad", doi="10.1/bad"))
    good = item_ref(zdb, mk_item(zdb, title="Good", doi="10.1/good"))
    finder_ok = _finder_writing(tmp_path)

    def finder(client, ref, **kw):
        if ref.key == bad.key:
            raise RuntimeError("cdn exploded")
        return finder_ok(client, ref, **kw)

    stats, lines = _go(zdb, tmp_path, [bad, good], finder=finder)
    assert stats["error"] == 1 and stats["attached"] == 1
    assert any("error  RuntimeError: cdn exploded" in ln for ln in lines)


def test_a_locked_database_aborts_the_run(zdb: Db, tmp_path: Path):
    ref = item_ref(zdb, mk_item(zdb, title="X", doi="10.1/x"))

    def finder(client, ref, **kw):
        raise sqlite3.OperationalError("database is locked")

    with pytest.raises(sqlite3.OperationalError):
        _go(zdb, tmp_path, [ref], finder=finder)


def test_ledger_skips_and_limit_stops_a_lazy_population(zdb: Db, tmp_path: Path):
    ids = [mk_item(zdb, title=f"p{n}", doi=f"10.1/{n}") for n in range(5)]
    led = Ledger(tmp_path / "ledger.jsonl")
    led.record(item_ref(zdb, ids[0]).key, outcome="attached")
    pulled = []

    def lazy():
        for i in ids:
            pulled.append(i)
            yield item_ref(zdb, i)

    stats, _ = _go(
        zdb, tmp_path, lazy(), ledger=led, limit=2, finder=lambda c, r, **k: None
    )
    assert stats["skipped"] == 1 and stats["attempted"] == 2
    assert len(pulled) == 4  # one skipped, two tried, one pulled to see the limit


def test_dry_run_previews_and_writes_nothing(zdb: Db, tmp_path: Path):
    a = item_ref(zdb, mk_item(zdb, title="Open", doi="10.1/o"))
    b = item_ref(zdb, mk_item(zdb, title="Closed"))

    def previewer(client, ref, *, email):
        return (ref.doi, ["https://oa.example/o.pdf"]) if ref.doi else ("", [])

    stats, lines = _go(
        zdb,
        tmp_path,
        [a, b],
        dry_run=True,
        previewer=previewer,
        finder=lambda *a, **k: pytest.fail("dry run must not download"),
    )
    assert stats["attempted"] == 2 and stats["attached"] == 0
    assert "    unpaywall  https://oa.example/o.pdf" in lines
    assert any("doi=-" in ln for ln in lines)
    assert any("a real run also tries S2 and PMC" in ln for ln in lines)
    assert not (tmp_path / "ledger.jsonl").exists()


def test_safe_name_falls_back_to_the_key():
    assert safe_name("A/B: c?", "KEY") == "A B c"
    assert safe_name("///", "KEY") == "KEY"