stack/tests/lead/test_review.py

"""Gold labels are a separate file. They do not tag the library."""

from __future__ import annotations

import sqlite3
import threading
from pathlib import Path

import pytest

from lead.review.codes import GATE, PROJECTS, criteria, reason_codes, reasons
from lead.review.library import (
    connect,
    load_item,
    next_unlabeled,
    pass_two_keys,
    project_keys,
)
from lead.review.store import Store
from prisma.audit import disagreements
from prisma.calibrate import sized
from zot.db import TYPE_MAP, Db
from zot.schema import create_db


def test_gate_sizes_match_the_calibration_module():
    for stage, n in GATE.items():
        assert sized(n, stage)
        assert not sized(n - 1, stage)


def test_every_strand_has_an_exclude_codebook():
    for project in PROJECTS:
        codes = reason_codes(project)
        assert "off-topic" in codes
        assert "unavailable-fulltext" in codes


def test_exclude_needs_a_codebook_reason_and_a_rationale(tmp_path: Path):
    store = Store(tmp_path / "labels.sqlite")
    store.record(
        zotero_key="AAAAAAAA",
        project="lead-history",
        stage="title_abstract",
        pass_n=1,
        decision="include",
        rationale="Names a predecessor of a registry target.",
        actor="kert",
    )
    with pytest.raises(ValueError, match="codebook"):
        store.record(
            zotero_key="BBBBBBBB",
            project="lead-history",
            stage="title_abstract",
            pass_n=1,
            decision="exclude",
            rationale="No.",
            actor="kert",
        )
    with pytest.raises(ValueError, match="only on exclude"):
        store.record(
            zotero_key="CCCCCCCC",
            project="lead-history",
            stage="title_abstract",
            pass_n=1,
            decision="uncertain",
            reason="off-topic",
            rationale="The abstract does not say.",
            actor="kert",
        )
    with pytest.raises(ValueError, match="rationale"):
        store.record(
            zotero_key="DDDDDDDD",
            project="lead-history",
            stage="title_abstract",
            pass_n=1,
            decision="include",
            rationale="  ",
            actor="kert",
        )


def test_a_later_label_is_the_current_one(tmp_path: Path):
    store = Store(tmp_path / "labels.sqlite")
    kwargs = {
        "zotero_key": "AAAAAAAA",
        "project": "lead-limits",
        "stage": "title_abstract",
        "pass_n": 1,
        "actor": "kert",
    }
    store.record(**kwargs, decision="exclude", reason="no-limit", rationale="No bound.")
    store.record(
        **kwargs, decision="uncertain", rationale="The formula might be there."
    )
    assert store.latest("lead-limits", "title_abstract", 1) == {"AAAAAAAA": "uncertain"}
    assert store.count("lead-limits", "title_abstract", 1) == 1
    assert len(store.history("lead-limits", "title_abstract", 1)) == 2


def test_pass_two_is_the_pass_one_keys_still_open():
    keys = ("AAAAAAAA", "BBBBBBBB", "CCCCCCCC")
    assert next_unlabeled(keys, {"AAAAAAAA"}) == "BBBBBBBB"
    assert next_unlabeled(keys, set(keys)) is None
    assert pass_two_keys(keys, {"BBBBBBBB", "CCCCCCCC"}, {"CCCCCCCC"}) == ("BBBBBBBB",)


def test_two_passes_can_disagree_without_a_sensitivity(tmp_path: Path):
    store = Store(tmp_path / "labels.sqlite")
    for key, first, second in (
        ("AAAAAAAA", "include", "exclude"),
        ("BBBBBBBB", "exclude", "exclude"),
    ):
        store.record(
            zotero_key=key,
            project="lead-scan",
            stage="title_abstract",
            pass_n=1,
            decision=first,
            reason="off-topic" if first == "exclude" else "",
            rationale="Pass 1.",
            actor="kert",
        )
        store.record(
            zotero_key=key,
            project="lead-scan",
            stage="title_abstract",
            pass_n=2,
            decision=second,
            reason="off-topic" if second == "exclude" else "",
            rationale="Pass 2.",
            actor="kert",
        )
    one = store.latest("lead-scan", "title_abstract", 1)
    two = store.latest("lead-scan", "title_abstract", 2)
    assert disagreements(one, two) == ("AAAAAAAA",)


def test_the_library_is_read_and_labels_do_not_tag_it(tmp_path: Path):
    library = tmp_path / "zotero.sqlite"
    create_db(str(library)).close()
    with Db(str(library)) as db:
        kept = db.create_item(TYPE_MAP["journalArticle"], key="AAAAAAAA")
        db.set_field(kept, "title", "A predecessor of the frailty index")
        db.set_field(kept, "abstractNote", "Medicare cohort, factor before outcome.")
        db.set_field(kept, "date", "2020")
        db.set_field(kept, "DOI", "10.0000/example")
        db.add_creators(kept, [("Ada", "Lovelace")])
        db.sync_tags(kept, ["project:lead-precursors"])
        note = db.create_item(TYPE_MAP["note"], key="BBBBBBBB")
        db.sync_tags(note, ["project:lead-precursors"])
        db.commit()

    con = connect(library)
    try:
        with pytest.raises(sqlite3.OperationalError):
            con.execute("CREATE TABLE leaked (id INTEGER)")
        assert project_keys(con, "lead-precursors") == ("AAAAAAAA",)
        item = load_item(con, "AAAAAAAA", tmp_path / "storage")
    finally:
        con.close()
    assert item.title.startswith("A predecessor")
    assert item.authors == ("Ada Lovelace",)
    assert item.year == "2020"
    assert item.doi == "10.0000/example"
    assert "Medicare" in item.abstract
    assert item.attachment_paths == ()

    Store(tmp_path / "labels.sqlite").record(
        zotero_key="AAAAAAAA",
        project="lead-precursors",
        stage="full_text",
        pass_n=1,
        decision="exclude",
        reason="cross-sectional-no-time-order",
        rationale="No temporal order.",
        actor="kert",
    )
    with Db(str(library)) as db:
        tags = [
            row[0]
            for row in db.con.execute(
                """SELECT t.name FROM itemTags it
                   JOIN tags t ON it.tagID = t.tagID
                   WHERE it.itemID = ?""",
                (kept,),
            ).fetchall()
        ]
    assert tags == ["project:lead-precursors"]


def test_two_threads_each_open_the_library(tmp_path: Path):
    library = tmp_path / "zotero.sqlite"
    create_db(str(library)).close()
    with Db(str(library)) as db:
        kept = db.create_item(TYPE_MAP["journalArticle"], key="AAAAAAAA")
        db.sync_tags(kept, ["project:lead-history"])
        db.commit()
    errors: list[BaseException] = []

    def read() -> None:
        con = connect(library)
        try:
            project_keys(con, "lead-history")
        except BaseException as exc:
            errors.append(exc)
        finally:
            con.close()

    threads = [threading.Thread(target=read) for _ in range(4)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    assert errors == []


def test_unknown_projects_and_blank_labels_are_refused(tmp_path: Path):
    with pytest.raises(ValueError):
        reason_codes("not-a-strand")
    with pytest.raises(ValueError):
        criteria("not-a-strand")
    with pytest.raises(ValueError):
        reasons("not-a-strand")
    store = Store(tmp_path / "labels.sqlite")
    kwargs = {
        "zotero_key": "AAAAAAAA",
        "project": "lead-history",
        "stage": "title_abstract",
        "pass_n": 1,
        "decision": "include",
        "rationale": "Names the predecessor.",
        "actor": "kert",
    }

    def _bad(**overrides):
        fields = dict(kwargs)
        fields.update(overrides)
        store.record(**fields)

    with pytest.raises(ValueError, match="zotero key"):
        _bad(zotero_key="  ")
    with pytest.raises(ValueError, match="not-a-strand"):
        _bad(project="not-a-strand")
    with pytest.raises(ValueError, match="screen"):
        _bad(stage="screen")
    with pytest.raises(ValueError, match="3"):
        _bad(pass_n=3)
    with pytest.raises(ValueError, match="maybe"):
        _bad(decision="maybe")


def test_a_missing_item_and_a_dateless_attachment_stay_out_of_the_label(
    tmp_path: Path,
):
    library = tmp_path / "zotero.sqlite"
    create_db(str(library)).close()
    paper = tmp_path / "paper.html"
    paper.write_text("<p>Exposure first.</p>")
    with Db(str(library)) as db:
        kept = db.create_item(TYPE_MAP["journalArticle"], key="AAAAAAAA")
        db.set_field(kept, "date", "n.d.")
        db.add_attachment(kept, path=str(paper), key="BBBBBBBB")
        db.commit()
    con = connect(library)
    try:
        with pytest.raises(KeyError):
            load_item(con, "ZZZZZZZZ", tmp_path)
        item = load_item(con, "AAAAAAAA", tmp_path)
    finally:
        con.close()
    assert item.year == ""
    assert item.attachment_paths == (paper,)