Files
stack/tests/zot/fetch/conftest.py
kert bee5f57893 feat(zot): stack zot fetch — open-access full-text connector
Attaches an open-access PDF to Zotero items that lack a full text.
Built on the selection, ledger and attach machinery from PR #826
(feat/zot-fetch-z1, left untouched on its branch); delivery is the
stack's legal open-access cascade only — every Unpaywall location →
Semantic Scholar openAccessPdf → PMC S3 → PMC OA service → Europe PMC
(prisma.fetch.fetch_one with no proxied client, so its mirror tier
never runs). No DOI? a Crossref title match first. Downloads must be
real PDFs before they are attached; the run ledgers every attempt
(attached / no_candidate / error; misses wait 30 days unless --retry).

From the #828 review list:
- Lazy selection: select_ids() is one cheap query (deleted and
  full-text items excluded in SQL for --all) and iter_refs() reads each
  item only when the run reaches it, so --limit bites before ~167k
  file-less items are materialised while the desktop is held down.
- A locked database (--no-hold with the desktop writing) aborts the run
  with a usage error that names the Zotero desktop, instead of a raw
  sqlite3.OperationalError — or one error per item.
- --dry-run reads only (checked against the running desktop), so it no
  longer stops/starts zotero; UNPAYWALL_EMAIL is documented in --help.

Tests: 66 (scratch Zotero DB, no network), 100% of src/zot/fetch.
2026-10-27 14:20:05 -04:00

51 lines
1.1 KiB
Python

"""Scratch Zotero DB for the fetch tests (no live DB, no network)."""
from __future__ import annotations
import sqlite3
from pathlib import Path
import pytest
from tests.zot.test_db import ZOTERO_SCHEMA, _seed_schema_maps
from zot.db import TYPE_MAP, Db
EXTRA_SCHEMA = """
CREATE TABLE IF NOT EXISTS deletedItems (
itemID INTEGER PRIMARY KEY,
dateDeleted TEXT DEFAULT CURRENT_TIMESTAMP NOT NULL
);
"""
@pytest.fixture()
def zdb(tmp_path: Path) -> Db:
path = str(tmp_path / "zotero.sqlite")
con = sqlite3.connect(path)
con.executescript(ZOTERO_SCHEMA)
con.executescript(EXTRA_SCHEMA)
_seed_schema_maps(con)
con.commit()
con.close()
d = Db(path)
yield d
d.close()
def mk_item(
db: Db,
*,
type_name: str = "journalArticle",
title: str = "",
doi: str = "",
isbn: str = "",
date: str = "",
extra: str = "",
) -> int:
item_id = db.create_item(TYPE_MAP[type_name])
db.set_fields(
item_id,
{"title": title, "DOI": doi, "ISBN": isbn, "date": date, "extra": extra},
)
return item_id