30db67b7e1
attach_file is idempotent by filename, but a re-run still called it for an attachment already on file — a wasted copy every time. Skip the call outright once the dup check finds it, instead of calling it and only using the dup flag to decide whether to count it as new.
276 lines
9.9 KiB
Python
276 lines
9.9 KiB
Python
"""bib.zotero_import — pull a Zotero collection's book items (with
|
|
files) into bib."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from bib.store import Store
|
|
from bib.zotero_import import ZoteroBook, import_books, list_collection
|
|
|
|
|
|
def _zotero_db(tmp_path):
|
|
"""A tiny sqlite db with the Zotero schema subset list_collection needs:
|
|
two book items in "AMA Coding Publications" (one with a storage:
|
|
attachment, one without), plus a distractor item in another
|
|
collection and a deleted item that must never surface."""
|
|
db = tmp_path / "zotero.sqlite"
|
|
con = sqlite3.connect(db)
|
|
con.executescript(
|
|
"""
|
|
CREATE TABLE items (itemID INTEGER PRIMARY KEY, itemTypeID INTEGER, key TEXT);
|
|
CREATE TABLE itemTypes (itemTypeID INTEGER PRIMARY KEY, typeName TEXT);
|
|
CREATE TABLE fields (fieldID INTEGER PRIMARY KEY, fieldName TEXT);
|
|
CREATE TABLE itemDataValues (valueID INTEGER PRIMARY KEY, value TEXT);
|
|
CREATE TABLE itemData (itemID INTEGER, fieldID INTEGER, valueID INTEGER);
|
|
CREATE TABLE collections (collectionID INTEGER PRIMARY KEY, collectionName TEXT);
|
|
CREATE TABLE collectionItems (collectionID INTEGER, itemID INTEGER);
|
|
CREATE TABLE itemAttachments (itemID INTEGER, parentItemID INTEGER, path TEXT, contentType TEXT);
|
|
CREATE TABLE deletedItems (itemID INTEGER);
|
|
|
|
INSERT INTO itemTypes VALUES (1, 'book'), (2, 'note');
|
|
|
|
-- item 1: CPT 2018, has a pdf attachment (item 10)
|
|
INSERT INTO items VALUES (1, 1, 'CPT2018K');
|
|
-- item 2: CPT Changes 2020, no attachment
|
|
INSERT INTO items VALUES (2, 1, 'CPTCHG20');
|
|
-- item 3: a note item in the same collection — never a "book"
|
|
INSERT INTO items VALUES (3, 2, 'NOTEKEY1');
|
|
-- item 4: deleted book — must never surface
|
|
INSERT INTO items VALUES (4, 1, 'DELETED1');
|
|
-- item 10: the attachment for item 1
|
|
INSERT INTO items VALUES (10, 3, 'ATTKEY01');
|
|
|
|
INSERT INTO fields VALUES (1, 'title'), (2, 'url'), (3, 'date'), (4, 'publisher');
|
|
|
|
INSERT INTO itemDataValues VALUES
|
|
(100, 'CPT 2018'), (101, 'https://ama.example/cpt2018'),
|
|
(102, '2017-11-01'), (103, 'American Medical Association'),
|
|
(200, 'CPT Changes 2020');
|
|
|
|
INSERT INTO itemData VALUES
|
|
(1, 1, 100), (1, 2, 101), (1, 3, 102), (1, 4, 103),
|
|
(2, 1, 200);
|
|
|
|
INSERT INTO collections VALUES (1, 'AMA Coding Publications'), (2, 'Other');
|
|
INSERT INTO collectionItems VALUES (1, 1), (1, 2), (1, 3), (1, 4), (2, 1);
|
|
|
|
INSERT INTO itemAttachments VALUES (10, 1, 'storage:CPT 2018.pdf', 'application/pdf');
|
|
|
|
INSERT INTO deletedItems VALUES (4);
|
|
"""
|
|
)
|
|
con.commit()
|
|
con.close()
|
|
return db
|
|
|
|
|
|
@pytest.fixture
|
|
def zcon(tmp_path):
|
|
con = sqlite3.connect(_zotero_db(tmp_path))
|
|
yield con
|
|
con.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def storage_dir(tmp_path):
|
|
d = tmp_path / "storage" / "ATTKEY01"
|
|
d.mkdir(parents=True)
|
|
(d / "CPT 2018.pdf").write_bytes(b"%PDF-1.4 fake cpt 2018")
|
|
return tmp_path / "storage"
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path):
|
|
s = Store(":memory:", storage_dir=tmp_path / "bib-storage")
|
|
yield s
|
|
s.close()
|
|
|
|
|
|
class TestListCollection:
|
|
def test_two_books_one_with_attachment(self, zcon, storage_dir):
|
|
books = list_collection(
|
|
zcon, "AMA Coding Publications", with_files=False, storage_dir=storage_dir
|
|
)
|
|
assert {b.key for b in books} == {"CPT2018K", "CPTCHG20"}
|
|
cpt = next(b for b in books if b.key == "CPT2018K")
|
|
assert cpt.title == "CPT 2018"
|
|
assert cpt.url == "https://ama.example/cpt2018"
|
|
assert cpt.year == "2017"
|
|
assert cpt.publisher == "American Medical Association"
|
|
assert cpt.attachments == (storage_dir / "ATTKEY01" / "CPT 2018.pdf",)
|
|
|
|
changes = next(b for b in books if b.key == "CPTCHG20")
|
|
assert changes.attachments == ()
|
|
assert changes.url == ""
|
|
|
|
def test_excludes_notes_and_deleted_items(self, zcon, storage_dir):
|
|
books = list_collection(
|
|
zcon, "AMA Coding Publications", with_files=False, storage_dir=storage_dir
|
|
)
|
|
keys = {b.key for b in books}
|
|
assert "NOTEKEY1" not in keys
|
|
assert "DELETED1" not in keys
|
|
|
|
def test_with_files_filters_to_items_with_attachments(self, zcon, storage_dir):
|
|
books = list_collection(
|
|
zcon, "AMA Coding Publications", with_files=True, storage_dir=storage_dir
|
|
)
|
|
assert [b.key for b in books] == ["CPT2018K"]
|
|
|
|
def test_unknown_collection_returns_empty(self, zcon, storage_dir):
|
|
assert (
|
|
list_collection(zcon, "Nope", with_files=False, storage_dir=storage_dir)
|
|
== []
|
|
)
|
|
|
|
|
|
class TestImportBooks:
|
|
def _cpt_book(self, storage_dir):
|
|
return ZoteroBook(
|
|
key="CPT2018K",
|
|
title="CPT 2018",
|
|
url="https://ama.example/cpt2018",
|
|
year="2017",
|
|
publisher="American Medical Association",
|
|
attachments=(storage_dir / "ATTKEY01" / "CPT 2018.pdf",),
|
|
)
|
|
|
|
def test_creates_source_item_with_tags_year_and_attachment(
|
|
self, store, storage_dir
|
|
):
|
|
book = self._cpt_book(storage_dir)
|
|
stats = import_books(
|
|
store, [book], tags=["source:ama", "module:coding", "llm:skip"]
|
|
)
|
|
assert stats["created"] == 1
|
|
assert stats["updated"] == 0
|
|
assert stats["attached"] == 1
|
|
|
|
(rec,) = stats["items"]
|
|
assert rec["zotero_key"] == "CPT2018K"
|
|
item = store.get(rec["key"])
|
|
assert item.item_type == "source"
|
|
assert item.title == "CPT 2018"
|
|
assert item.url == "https://ama.example/cpt2018"
|
|
assert set(item.tags) == {
|
|
"source:ama",
|
|
"module:coding",
|
|
"llm:skip",
|
|
"year:2017",
|
|
}
|
|
|
|
attachments = (
|
|
store._con()
|
|
.execute(
|
|
"SELECT filename FROM attachments a "
|
|
"JOIN items i ON i.id = a.item_id WHERE i.key = ?",
|
|
(rec["key"],),
|
|
)
|
|
.fetchall()
|
|
)
|
|
assert [a[0] for a in attachments] == ["CPT 2018.pdf"]
|
|
|
|
def test_fallback_url_used_when_book_has_none(self, store):
|
|
book = ZoteroBook(
|
|
key="NOURLKEY",
|
|
title="No URL Book",
|
|
url="",
|
|
year="2020",
|
|
publisher="",
|
|
attachments=(),
|
|
)
|
|
import_books(store, [book], tags=["source:ama"])
|
|
item = store.get(store.list_items(query="No URL Book")[0].key)
|
|
assert item.url == "zotero://select/library/items/NOURLKEY"
|
|
|
|
def test_rerun_is_unchanged_with_no_duplicate_attachment(self, store, storage_dir):
|
|
book = self._cpt_book(storage_dir)
|
|
tags = ["source:ama", "module:coding", "llm:skip"]
|
|
first = import_books(store, [book], tags=tags)
|
|
second = import_books(store, [book], tags=tags)
|
|
|
|
assert second["created"] == 0
|
|
assert second["unchanged"] == 1
|
|
assert second["attached"] == 0
|
|
|
|
bib_key = first["items"][0]["key"]
|
|
rows = (
|
|
store._con()
|
|
.execute(
|
|
"SELECT filename FROM attachments a "
|
|
"JOIN items i ON i.id = a.item_id WHERE i.key = ?",
|
|
(bib_key,),
|
|
)
|
|
.fetchall()
|
|
)
|
|
assert len(rows) == 1 # no duplicate attachment
|
|
|
|
def test_rerun_never_calls_attach_file_for_a_duplicate(self, store, storage_dir):
|
|
"""Minor: attach_file is idempotent-by-filename, but a re-run
|
|
used to call it anyway (a wasted copy) — it must now be skipped
|
|
entirely once the dup check finds the attachment already there."""
|
|
book = self._cpt_book(storage_dir)
|
|
tags = ["source:ama"]
|
|
import_books(store, [book], tags=tags)
|
|
|
|
calls = []
|
|
orig_attach_file = store.attach_file
|
|
store.attach_file = lambda *a, **k: (
|
|
calls.append((a, k)) or orig_attach_file(*a, **k)
|
|
)
|
|
try:
|
|
second = import_books(store, [book], tags=tags)
|
|
finally:
|
|
store.attach_file = orig_attach_file
|
|
|
|
assert calls == []
|
|
assert second["attached"] == 0
|
|
|
|
def test_dry_run_makes_no_writes(self, store, storage_dir):
|
|
book = self._cpt_book(storage_dir)
|
|
stats = import_books(store, [book], tags=["source:ama"], dry_run=True)
|
|
assert stats["created"] == 1
|
|
assert stats["attached"] == 1
|
|
assert store.list_items() == [] # nothing written
|
|
|
|
def test_dry_run_after_real_import_reports_unchanged(self, store, storage_dir):
|
|
book = self._cpt_book(storage_dir)
|
|
tags = ["source:ama"]
|
|
import_books(store, [book], tags=tags)
|
|
stats = import_books(store, [book], tags=tags, dry_run=True)
|
|
assert stats["unchanged"] == 1
|
|
assert stats["attached"] == 0
|
|
assert stats["items"][0]["key"] # bib key resolved from the existing url
|
|
|
|
def test_missing_attachment_file_counts_and_skips(self, store, storage_dir):
|
|
ghost_path = storage_dir / "GHOSTKEY" / "missing.pdf"
|
|
book = ZoteroBook(
|
|
key="GHOSTBK1",
|
|
title="Ghost Book",
|
|
url="https://ama.example/ghost",
|
|
year="2020",
|
|
publisher="American Medical Association",
|
|
attachments=(ghost_path,),
|
|
)
|
|
stats = import_books(store, [book], tags=["source:ama"])
|
|
|
|
assert stats["missing"] == 1
|
|
assert stats["attached"] == 0
|
|
(rec,) = stats["items"]
|
|
assert rec["missing"] == 1
|
|
assert rec["missing_paths"] == [str(ghost_path)]
|
|
|
|
rows = (
|
|
store._con()
|
|
.execute(
|
|
"SELECT filename FROM attachments a "
|
|
"JOIN items i ON i.id = a.item_id WHERE i.key = ?",
|
|
(rec["key"],),
|
|
)
|
|
.fetchall()
|
|
)
|
|
assert rows == [] # no attachment row for the missing file
|