stack/tests/zot/fetch/test_select.py

from __future__ import annotations

import pytest

from tests.zot.fetch.conftest import mk_item
from zot.db import Db
from zot.fetch.select import (
    ItemRef,
    first_isbn,
    item_ref,
    normalise_doi,
    select_items,
    year_of,
)


@pytest.mark.parametrize(
    "raw, want",
    [
        ("10.1056/NEJMoa1000678", "10.1056/NEJMoa1000678"),
        ("  https://doi.org/10.1056/NEJMoa1000678 ", "10.1056/NEJMoa1000678"),
        ("http://dx.doi.org/10.1/ab", "10.1/ab"),
        ("doi:10.1/ab", "10.1/ab"),
        ("", ""),
    ],
)
def test_doi_normalised_from_url_form(raw, want):
    assert normalise_doi(raw) == want


@pytest.mark.parametrize(
    "raw, want",
    [
        ("978-1-64016-228-0 1640162283", "9781640162280"),
        ("1640162283", "1640162283"),
        ("ISBN 0-19-852663-6", "0198526636"),
        ("no isbn here", ""),
        ("", ""),
    ],
)
def test_isbn_prefers_13_digit_and_strips_hyphens(raw, want):
    assert first_isbn(raw) == want


@pytest.mark.parametrize(
    "raw, want", [("2010-08-19", "2010"), ("August 2010", "2010"), ("", "")]
)
def test_year_of(raw, want):
    assert year_of(raw) == want


def test_item_ref_reads_fields_and_kind(zdb: Db):
    paper = mk_item(zdb, title="P", doi="https://doi.org/10.1/x", date="2020-01-01")
    book = mk_item(zdb, type_name="book", title="B", isbn="978-1-64016-228-0")
    p = item_ref(zdb, paper)
    b = item_ref(zdb, book)
    key = zdb.get_item(paper)["key"]
    assert p == ItemRef(paper, key, "journalArticle", "P", "10.1/x", "", "2020")
    assert p.kind == "paper"
    assert p.ids == {"doi": "10.1/x"}
    assert b.isbn == "9781640162280" and b.kind == "book"


def test_select_requires_exactly_one_primary_selector(zdb: Db):
    with pytest.raises(ValueError, match="exactly one"):
        select_items(zdb)
    with pytest.raises(ValueError, match="exactly one"):
        select_items(zdb, tag="t", all_=True)


def test_select_by_tag_type_item_collection_all(zdb: Db):
    a = mk_item(zdb, title="A", doi="10.1/a")
    b = mk_item(zdb, type_name="book", title="B", isbn="9781640162280")
    zdb.tag_item(a, "rfi:codeset")
    ckey = zdb.ensure_collection("AMA Coding Publications")
    zdb.add_to_collection(b, collection_key=ckey)
    assert [r.item_id for r in select_items(zdb, tag="rfi:codeset")] == [a]
    assert [r.item_id for r in select_items(zdb, type_="book")] == [b]
    akey = zdb.get_item(a)["key"]
    assert [r.item_id for r in select_items(zdb, item=akey)] == [a]
    got = select_items(zdb, collection="AMA Coding Publications")
    assert [r.item_id for r in got] == [b]
    assert [r.item_id for r in select_items(zdb, all_=True)] == [a, b]


def test_type_filter_combines_with_primary_selector(zdb: Db):
    a = mk_item(zdb, title="A")
    b = mk_item(zdb, type_name="book", title="B")
    ckey = zdb.ensure_collection("C")
    zdb.add_to_collection(a, collection_key=ckey)
    zdb.add_to_collection(b, collection_key=ckey)
    got = select_items(zdb, collection="C", type_="book")
    assert [r.item_id for r in got] == [b]


def test_unknown_collection_or_item_or_type_raises(zdb: Db):
    with pytest.raises(ValueError, match="no collection named"):
        select_items(zdb, collection="Nope")
    with pytest.raises(ValueError, match="no item with key"):
        select_items(zdb, item="ZZZZZZZZ")
    with pytest.raises(ValueError, match="unknown item type"):
        select_items(zdb, type_="novel")


def test_select_excludes_full_text_deleted_document_and_children(zdb: Db):
    has_pdf = mk_item(zdb, title="has pdf")
    zdb.add_attachment(has_pdf, content_type="application/pdf", path="storage:x.pdf")
    deleted = mk_item(zdb, title="deleted")
    zdb.con.execute("INSERT INTO deletedItems (itemID) VALUES (?)", (deleted,))
    mk_item(zdb, type_name="document", title="reg pdf")
    keep = mk_item(zdb, title="keep")
    got = [r.item_id for r in select_items(zdb, all_=True)]
    assert got == [keep]


def test_select_all_excludes_document_even_when_typed(zdb: Db):
    mk_item(zdb, type_name="document", title="reg pdf")
    assert list(select_items(zdb, type_="document")) == []


def test_selection_is_lazy_so_limit_bites_before_reading_everything(
    zdb: Db, monkeypatch
):
    for n in range(50):
        mk_item(zdb, title=f"paper {n}", doi=f"10.1/{n}")
    import zot.fetch.select as sel

    reads = []
    real = sel.item_ref
    monkeypatch.setattr(sel, "item_ref", lambda db, i: reads.append(i) or real(db, i))
    refs = select_items(zdb, all_=True)
    assert reads == []  # nothing materialised yet
    first = next(refs)
    assert first.title == "paper 0" and len(reads) == 1


def test_all_excludes_deleted_and_full_text_in_sql(zdb: Db):
    keep = mk_item(zdb, title="needs text", doi="10.1/a")
    gone = mk_item(zdb, title="trashed", doi="10.1/b")
    has_pdf = mk_item(zdb, title="already has pdf", doi="10.1/c")
    zdb.con.execute("INSERT INTO deletedItems (itemID) VALUES (?)", (gone,))
    zdb.add_attachment(
        has_pdf, key="ATTPDF01", content_type="application/pdf", path="storage:x.pdf"
    )
    zdb.commit()
    from zot.fetch.select import select_ids

    assert select_ids(zdb, all_=True) == [keep]


def test_iter_refs_skips_full_text_for_targeted_populations(zdb: Db):
    a = mk_item(zdb, title="a", doi="10.1/a")
    b = mk_item(zdb, title="b", doi="10.1/b")
    zdb.add_attachment(
        b, key="ATTPDF02", content_type="application/pdf", path="storage:b.pdf"
    )
    zdb.commit()
    from zot.fetch.select import iter_refs

    assert [r.item_id for r in iter_refs(zdb, [a, b])] == [a]