661 lines
22 KiB
Python
661 lines
22 KiB
Python
"""Tests for bib.sync — push bib items into Zotero SQLite.
|
|
|
|
ORM primitives (key gen, ensure_value, ensure_tag, set_field, sync_tags,
|
|
ensure_collection) are tested in ``tests/zot/test_db.py``. This file
|
|
tests the bib→Zotero adapter: field mapping and push_to_zotero.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from bib.item import Download, Manual, Regulation, Rule, Source
|
|
from bib.sync import _item_to_zotero_fields, push_to_zotero
|
|
|
|
# ── Zotero schema for tests ─────────────────────────────────────────
|
|
|
|
ZOTERO_SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS items (
|
|
itemID INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
itemTypeID INTEGER NOT NULL,
|
|
dateAdded TEXT,
|
|
dateModified TEXT,
|
|
clientDateModified TEXT,
|
|
libraryID INTEGER DEFAULT 1,
|
|
key TEXT NOT NULL UNIQUE,
|
|
version INTEGER DEFAULT 0,
|
|
synced INTEGER DEFAULT 0
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS itemDataValues (
|
|
valueID INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
value TEXT NOT NULL UNIQUE
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS itemData (
|
|
itemID INTEGER NOT NULL,
|
|
fieldID INTEGER NOT NULL,
|
|
valueID INTEGER NOT NULL,
|
|
PRIMARY KEY (itemID, fieldID)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS tags (
|
|
tagID INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL UNIQUE
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS itemTags (
|
|
itemID INTEGER NOT NULL,
|
|
tagID INTEGER NOT NULL,
|
|
type INTEGER DEFAULT 0,
|
|
PRIMARY KEY (itemID, tagID)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS collections (
|
|
collectionID INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
collectionName TEXT NOT NULL,
|
|
parentCollectionID INTEGER,
|
|
clientDateModified TEXT,
|
|
libraryID INTEGER DEFAULT 1,
|
|
key TEXT NOT NULL UNIQUE,
|
|
version INTEGER DEFAULT 0,
|
|
synced INTEGER DEFAULT 0
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS collectionItems (
|
|
collectionID INTEGER NOT NULL,
|
|
itemID INTEGER NOT NULL,
|
|
orderIndex INTEGER NOT NULL DEFAULT 0,
|
|
PRIMARY KEY (collectionID, itemID)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS creators (
|
|
creatorID INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
firstName TEXT,
|
|
lastName TEXT,
|
|
fieldMode INT,
|
|
UNIQUE (lastName, firstName, fieldMode)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS itemCreators (
|
|
itemID INT NOT NULL,
|
|
creatorID INT NOT NULL,
|
|
creatorTypeID INT NOT NULL DEFAULT 1,
|
|
orderIndex INT NOT NULL DEFAULT 0,
|
|
PRIMARY KEY (itemID, creatorID, creatorTypeID, orderIndex),
|
|
UNIQUE (itemID, orderIndex)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS itemTypes (
|
|
itemTypeID INTEGER PRIMARY KEY,
|
|
typeName TEXT
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS fields (
|
|
fieldID INTEGER PRIMARY KEY,
|
|
fieldName TEXT
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS creatorTypes (
|
|
creatorTypeID INTEGER PRIMARY KEY,
|
|
creatorType TEXT
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS itemNotes (
|
|
itemID INTEGER PRIMARY KEY,
|
|
parentItemID INT,
|
|
note TEXT,
|
|
title TEXT
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS itemAttachments (
|
|
itemID INTEGER PRIMARY KEY,
|
|
parentItemID INT,
|
|
linkMode INT,
|
|
contentType TEXT,
|
|
charsetID INT,
|
|
path TEXT,
|
|
syncState INT DEFAULT 0,
|
|
storageModTime INT,
|
|
storageHash TEXT
|
|
);
|
|
"""
|
|
|
|
|
|
def _make_zotero_db(path: str = ":memory:") -> sqlite3.Connection:
|
|
from zot.db import CREATOR_TYPES, FIELD_MAP, TYPE_MAP
|
|
|
|
con = sqlite3.connect(path)
|
|
con.row_factory = sqlite3.Row
|
|
con.executescript(ZOTERO_SCHEMA)
|
|
for name, tid in TYPE_MAP.items():
|
|
con.execute("INSERT OR IGNORE INTO itemTypes VALUES (?, ?)", (tid, name))
|
|
for name, fid in FIELD_MAP.items():
|
|
con.execute("INSERT OR IGNORE INTO fields VALUES (?, ?)", (fid, name))
|
|
for name, cid in CREATOR_TYPES.items():
|
|
con.execute("INSERT OR IGNORE INTO creatorTypes VALUES (?, ?)", (cid, name))
|
|
con.commit()
|
|
return con
|
|
|
|
|
|
# ── _item_to_zotero_fields ─────────────────────────────────────────
|
|
|
|
|
|
class TestItemToZoteroFields:
|
|
def test_rule(self) -> None:
|
|
item = Rule(
|
|
title="PFS Final Rule",
|
|
fr_volume="90",
|
|
fr_page="98452",
|
|
document_number="2025-19787",
|
|
cms_id="CMS-1832-F",
|
|
rule_type="final",
|
|
date_published="2025-11-01",
|
|
effective_date="2026-01-01",
|
|
url="https://example.com/rule",
|
|
abstract="Rule abstract",
|
|
)
|
|
fields = _item_to_zotero_fields(item)
|
|
assert fields["nameOfAct"] == "PFS Final Rule"
|
|
assert fields["code"] == "FR"
|
|
assert fields["codeNumber"] == "90"
|
|
assert fields["pages"] == "98452"
|
|
assert fields["session"] == "CMS-1832-F"
|
|
assert "Document: 2025-19787" in fields["history"]
|
|
assert "Type: final" in fields["history"]
|
|
assert "Effective: 2026-01-01" in fields["history"]
|
|
|
|
def test_regulation(self) -> None:
|
|
item = Regulation(
|
|
title="42 CFR Part 414",
|
|
cfr_title="42",
|
|
cfr_part="414",
|
|
cfr_section="414.22",
|
|
authority="42 USC 1395w-4",
|
|
effective_date="2025-01-01",
|
|
url="https://ecfr.gov/414",
|
|
)
|
|
fields = _item_to_zotero_fields(item)
|
|
assert fields["code"] == "C.F.R."
|
|
assert fields["codeNumber"] == "42"
|
|
assert fields["section"] == "414.22"
|
|
assert "Part 414" in fields["history"]
|
|
assert "Authority: 42 USC 1395w-4" in fields["history"]
|
|
|
|
def test_regulation_no_authority(self) -> None:
|
|
item = Regulation(cfr_title="42", cfr_part="414")
|
|
fields = _item_to_zotero_fields(item)
|
|
assert fields["history"] == "Part 414"
|
|
assert "Authority" not in fields["history"]
|
|
|
|
def test_manual(self) -> None:
|
|
item = Manual(
|
|
title="Chapter 12",
|
|
manual_name="Claims Processing Manual",
|
|
pub_number="100-04",
|
|
chapter="12",
|
|
transmittal="R100",
|
|
institution="CMS",
|
|
date_published="2025-01-01",
|
|
url="https://cms.gov/manual",
|
|
)
|
|
fields = _item_to_zotero_fields(item)
|
|
assert fields["reportType"] == "Internet-Only Manual"
|
|
assert fields["reportNumber"] == "100-04"
|
|
assert fields["seriesTitle"] == "Claims Processing Manual"
|
|
assert "Chapter 12" in fields["extra"]
|
|
assert "Transmittal: R100" in fields["extra"]
|
|
assert fields["place"] == "Baltimore, MD"
|
|
|
|
def test_manual_no_transmittal(self) -> None:
|
|
item = Manual(title="Test", chapter="5")
|
|
fields = _item_to_zotero_fields(item)
|
|
assert "Transmittal" not in fields["extra"]
|
|
|
|
def test_manual_no_chapter(self) -> None:
|
|
item = Manual(title="Test")
|
|
fields = _item_to_zotero_fields(item)
|
|
assert "seriesNumber" not in fields
|
|
|
|
def test_download(self) -> None:
|
|
item = Download(
|
|
title="RVU26A",
|
|
file_urls=["https://cms.gov/rvu.zip"],
|
|
date_published="2026-01-01",
|
|
url="https://cms.gov/rvu26a",
|
|
)
|
|
fields = _item_to_zotero_fields(item)
|
|
assert fields["title"] == "RVU26A"
|
|
assert fields["websiteType"] == "Government Data Portal"
|
|
assert "Files: https://cms.gov/rvu.zip" in fields["extra"]
|
|
|
|
def test_download_no_files(self) -> None:
|
|
item = Download(title="Test", url="https://cms.gov/test")
|
|
fields = _item_to_zotero_fields(item)
|
|
assert "Files:" not in fields.get("extra", "")
|
|
|
|
def test_source(self) -> None:
|
|
item = Source(
|
|
title="Test Doc",
|
|
doc_type="guidance",
|
|
institution="CMS",
|
|
date_published="2025-01-01",
|
|
url="https://cms.gov/doc",
|
|
)
|
|
fields = _item_to_zotero_fields(item)
|
|
assert fields["title"] == "Test Doc"
|
|
assert "Type: guidance" in fields["extra"]
|
|
assert fields["publisher"] == "CMS"
|
|
|
|
|
|
# ── push_to_zotero ──────────────────────────────────────────────────
|
|
|
|
|
|
class TestPushToZotero:
|
|
def test_push_rule(self, tmp_path) -> None:
|
|
db_path = str(tmp_path / "zotero.sqlite")
|
|
con = _make_zotero_db(db_path)
|
|
con.close()
|
|
|
|
items = [
|
|
Rule(
|
|
key="RULEKEY2",
|
|
title="PFS Final Rule",
|
|
fr_volume="90",
|
|
fr_page="98452",
|
|
url="https://example.com/rule",
|
|
tags=["module:pfs"],
|
|
)
|
|
]
|
|
stats = push_to_zotero(items, zotero_db=db_path)
|
|
assert stats["created"] == 1
|
|
assert stats["skipped"] == 0
|
|
assert stats["tags"] == 1
|
|
|
|
def test_push_all_types(self, tmp_path) -> None:
|
|
db_path = str(tmp_path / "zotero.sqlite")
|
|
con = _make_zotero_db(db_path)
|
|
con.close()
|
|
|
|
items = [
|
|
Rule(title="Rule", url="https://ex.com/rule"),
|
|
Manual(title="Manual", url="https://ex.com/manual"),
|
|
Regulation(title="Reg", url="https://ex.com/reg"),
|
|
Download(title="DL", url="https://ex.com/dl"),
|
|
Source(title="Src", url="https://ex.com/src"),
|
|
]
|
|
stats = push_to_zotero(items, zotero_db=db_path)
|
|
assert stats["created"] == 5
|
|
|
|
def test_skip_existing_url(self, tmp_path) -> None:
|
|
db_path = str(tmp_path / "zotero.sqlite")
|
|
con = _make_zotero_db(db_path)
|
|
con.close()
|
|
|
|
items = [Rule(title="First", url="https://ex.com/rule")]
|
|
push_to_zotero(items, zotero_db=db_path)
|
|
|
|
items2 = [Rule(title="Second", url="https://ex.com/rule", tags=["new-tag"])]
|
|
stats = push_to_zotero(items2, zotero_db=db_path)
|
|
assert stats["skipped"] == 1
|
|
assert stats["created"] == 0
|
|
|
|
def test_skip_unknown_type(self, tmp_path) -> None:
|
|
db_path = str(tmp_path / "zotero.sqlite")
|
|
con = _make_zotero_db(db_path)
|
|
con.close()
|
|
|
|
from bib.item import Item
|
|
|
|
items = [Item(item_type="unknown", title="Unknown")]
|
|
stats = push_to_zotero(items, zotero_db=db_path)
|
|
assert stats["skipped"] == 1
|
|
|
|
def test_with_collection(self, tmp_path) -> None:
|
|
db_path = str(tmp_path / "zotero.sqlite")
|
|
con = _make_zotero_db(db_path)
|
|
con.execute(
|
|
"INSERT INTO collections (collectionName, libraryID, key, version, synced) "
|
|
"VALUES ('Test', 1, 'CKEY2345', 0, 0)"
|
|
)
|
|
con.commit()
|
|
con.close()
|
|
|
|
items = [Rule(title="R1", url="https://ex.com/r1")]
|
|
stats = push_to_zotero(items, zotero_db=db_path, collection_key="CKEY2345")
|
|
assert stats["collections"] == 1
|
|
|
|
def test_item_collections(self, tmp_path) -> None:
|
|
db_path = str(tmp_path / "zotero.sqlite")
|
|
con = _make_zotero_db(db_path)
|
|
con.execute(
|
|
"INSERT INTO collections (collectionName, libraryID, key, version, synced) "
|
|
"VALUES ('ACO', 1, 'ACKEY234', 0, 0)"
|
|
)
|
|
con.commit()
|
|
con.close()
|
|
|
|
items = [
|
|
Rule(
|
|
title="R1",
|
|
url="https://ex.com/r1",
|
|
collections=["ACKEY234"],
|
|
)
|
|
]
|
|
stats = push_to_zotero(items, zotero_db=db_path)
|
|
assert stats["created"] == 1
|
|
|
|
def test_key_collision(self, tmp_path) -> None:
|
|
db_path = str(tmp_path / "zotero.sqlite")
|
|
con = _make_zotero_db(db_path)
|
|
con.execute("INSERT INTO items (itemTypeID, key) VALUES (20, 'RKEY2345')")
|
|
con.commit()
|
|
con.close()
|
|
|
|
items = [Rule(key="RKEY2345", title="Collision")]
|
|
stats = push_to_zotero(items, zotero_db=db_path)
|
|
assert stats["created"] == 1
|
|
|
|
def test_no_url_item(self, tmp_path) -> None:
|
|
db_path = str(tmp_path / "zotero.sqlite")
|
|
con = _make_zotero_db(db_path)
|
|
con.close()
|
|
|
|
items = [Rule(title="No URL")]
|
|
stats = push_to_zotero(items, zotero_db=db_path)
|
|
assert stats["created"] == 1
|
|
|
|
def test_short_key_generates_new(self, tmp_path) -> None:
|
|
db_path = str(tmp_path / "zotero.sqlite")
|
|
con = _make_zotero_db(db_path)
|
|
con.close()
|
|
|
|
items = [Rule(key="SHORT", title="Short Key")]
|
|
stats = push_to_zotero(items, zotero_db=db_path)
|
|
assert stats["created"] == 1
|
|
|
|
def test_collection_key_not_found(self, tmp_path) -> None:
|
|
db_path = str(tmp_path / "zotero.sqlite")
|
|
con = _make_zotero_db(db_path)
|
|
con.close()
|
|
|
|
items = [Rule(title="R1")]
|
|
stats = push_to_zotero(items, zotero_db=db_path, collection_key="ZZZZ2345")
|
|
assert stats["collections"] == 0
|
|
assert stats["created"] == 1
|
|
|
|
def test_item_collection_not_found(self, tmp_path) -> None:
|
|
db_path = str(tmp_path / "zotero.sqlite")
|
|
con = _make_zotero_db(db_path)
|
|
con.close()
|
|
|
|
items = [Rule(title="R1", collections=["ZZZZ2345"])]
|
|
stats = push_to_zotero(items, zotero_db=db_path)
|
|
assert stats["created"] == 1
|
|
|
|
def test_item_collection_same_as_main(self, tmp_path) -> None:
|
|
"""If item collection matches collection_key, skip duplicate."""
|
|
db_path = str(tmp_path / "zotero.sqlite")
|
|
con = _make_zotero_db(db_path)
|
|
con.execute(
|
|
"INSERT INTO collections (collectionName, libraryID, key, version, synced) "
|
|
"VALUES ('Main', 1, 'MKEY2345', 0, 0)"
|
|
)
|
|
con.commit()
|
|
con.close()
|
|
|
|
items = [
|
|
Rule(
|
|
title="R1",
|
|
url="https://ex.com/r1",
|
|
collections=["MKEY2345"],
|
|
)
|
|
]
|
|
stats = push_to_zotero(items, zotero_db=db_path, collection_key="MKEY2345")
|
|
assert stats["collections"] == 1
|
|
|
|
|
|
class TestSyncNotes:
|
|
"""bib notes → Zotero itemNotes (non-attachment children)."""
|
|
|
|
def _setup_bib_with_note(self, tmp_path, *, html: str, title: str):
|
|
from bib import connect
|
|
from bib.item import Source
|
|
|
|
bib_db = tmp_path / "bib.sqlite"
|
|
store = connect(str(bib_db))
|
|
item = Source(
|
|
title="Comment CMS-2024-0001-0001",
|
|
url="https://www.regulations.gov/comment/CMS-2024-0001-0001",
|
|
)
|
|
item_key = store.upsert(item)
|
|
store.attach_note(item_key, html, title=title)
|
|
store.close()
|
|
return bib_db, item_key
|
|
|
|
def test_note_creates_zotero_itemnote(self, tmp_path):
|
|
bib_db, item_key = self._setup_bib_with_note(
|
|
tmp_path, html="<h1>Body</h1><p>text</p>", title="Comment text"
|
|
)
|
|
|
|
zot_db = str(tmp_path / "zotero.sqlite")
|
|
con = _make_zotero_db(zot_db)
|
|
con.close()
|
|
|
|
from bib import connect
|
|
|
|
store = connect(str(bib_db))
|
|
items = store.list_items(tag=None)
|
|
stats = push_to_zotero(items, store=store, zotero_db=zot_db)
|
|
store.close()
|
|
|
|
assert stats["notes"] == 1
|
|
|
|
con = sqlite3.connect(zot_db)
|
|
rows = con.execute("SELECT title, note FROM itemNotes").fetchall()
|
|
con.close()
|
|
assert rows == [
|
|
(
|
|
"Comment text",
|
|
'<div class="zotero-note znv1"><h1>Body</h1><p>text</p></div>',
|
|
)
|
|
]
|
|
|
|
def test_resync_is_idempotent(self, tmp_path):
|
|
"""Second sync must not duplicate the note (dedup by parent + title)."""
|
|
bib_db, item_key = self._setup_bib_with_note(
|
|
tmp_path, html="<p>x</p>", title="Comment text"
|
|
)
|
|
|
|
zot_db = str(tmp_path / "zotero.sqlite")
|
|
con = _make_zotero_db(zot_db)
|
|
con.close()
|
|
|
|
from bib import connect
|
|
|
|
store = connect(str(bib_db))
|
|
items = store.list_items(tag=None)
|
|
|
|
push_to_zotero(items, store=store, zotero_db=zot_db)
|
|
stats = push_to_zotero(items, store=store, zotero_db=zot_db)
|
|
store.close()
|
|
|
|
assert stats["notes"] == 0 # second pass, nothing new
|
|
|
|
con = sqlite3.connect(zot_db)
|
|
n = con.execute("SELECT COUNT(*) FROM itemNotes").fetchone()[0]
|
|
con.close()
|
|
assert n == 1
|
|
|
|
|
|
# ── dedupe_collection ───────────────────────────────────────────────
|
|
|
|
|
|
class TestDedupeCollection:
|
|
# Two distinct sends of the same message (CMS re-sends with a fresh
|
|
# Message-ID → distinct bib keys, same title+date) plus one genuinely
|
|
# different email. Store.create generates keys, so tests capture the
|
|
# returned key per item.
|
|
_TAGS = ["source:email", "mailbox:cmsupdates"]
|
|
|
|
def test_removes_title_date_twins_keeps_earliest(self, tmp_path) -> None:
|
|
from bib.store import Store
|
|
from bib.sync import dedupe_collection
|
|
|
|
zot_db = str(tmp_path / "zotero.sqlite")
|
|
_make_zotero_db(zot_db).close()
|
|
|
|
store = Store(str(tmp_path / "bib.sqlite"))
|
|
first = store.create(
|
|
Source(title="Weekly Update", date_published="2026-07-01"),
|
|
tags=self._TAGS,
|
|
)
|
|
twin = store.create(
|
|
Source(title="Weekly Update", date_published="2026-07-01"),
|
|
tags=self._TAGS,
|
|
)
|
|
distinct = store.create(
|
|
Source(title="Weekly Update", date_published="2026-07-08"),
|
|
tags=self._TAGS,
|
|
)
|
|
push_to_zotero(store.list_items(), store=store, zotero_db=zot_db)
|
|
|
|
stats = dedupe_collection("Cmsupdates", store=store, zotero_db=zot_db)
|
|
assert stats == {"groups": 1, "removed": 1}
|
|
|
|
con = sqlite3.connect(zot_db)
|
|
keys = {r[0] for r in con.execute("SELECT key FROM items").fetchall()}
|
|
con.close()
|
|
assert first in keys # earliest kept
|
|
assert twin not in keys
|
|
assert distinct in keys
|
|
# bib twin removed too, so the next sync can't resurrect it
|
|
with pytest.raises(KeyError):
|
|
store.get(twin)
|
|
assert store.get(first) is not None
|
|
store.close()
|
|
|
|
def test_noop_when_no_dupes(self, tmp_path) -> None:
|
|
from bib.store import Store
|
|
from bib.sync import dedupe_collection
|
|
|
|
zot_db = str(tmp_path / "zotero.sqlite")
|
|
_make_zotero_db(zot_db).close()
|
|
store = Store(str(tmp_path / "bib.sqlite"))
|
|
store.create(
|
|
Source(title="Weekly Update", date_published="2026-07-08"),
|
|
tags=self._TAGS,
|
|
)
|
|
push_to_zotero(store.list_items(), store=store, zotero_db=zot_db)
|
|
|
|
assert dedupe_collection("Cmsupdates", store=store, zotero_db=zot_db) == {
|
|
"groups": 0,
|
|
"removed": 0,
|
|
}
|
|
store.close()
|
|
|
|
|
|
# ── FR jump links → child link attachments (#637) ───────────────────
|
|
|
|
|
|
class TestSyncFrLinks:
|
|
def _bib_with_placed_link(self, tmp_path):
|
|
from bib import frlink
|
|
from bib.store import Store
|
|
from tests.bib.test_frlink import DOC_META, FIXTURE_HTML
|
|
|
|
store = Store(str(tmp_path / "bib.sqlite"))
|
|
key = store.create(
|
|
Rule(
|
|
key="RUEFR423",
|
|
title="FR Rule",
|
|
url="https://www.federalregister.gov/documents/2026/07/16/2026-14327/x",
|
|
document_number="2026-14327",
|
|
fr_volume="91",
|
|
)
|
|
)
|
|
frlink.grab(store, key, fetch=lambda _doc: (DOC_META, FIXTURE_HTML))
|
|
frlink.place(store, "91 FR 100 ¶2", label="¶ p-2 — 91 FR 100")
|
|
return store, key
|
|
|
|
def test_push_creates_linkmode3_child(self, tmp_path) -> None:
|
|
db_path = str(tmp_path / "zotero.sqlite")
|
|
_make_zotero_db(db_path).close()
|
|
store, key = self._bib_with_placed_link(tmp_path)
|
|
|
|
stats = push_to_zotero([store.get(key)], store=store, zotero_db=db_path)
|
|
assert stats["fr_links"] == 1
|
|
|
|
con = sqlite3.connect(db_path)
|
|
con.row_factory = sqlite3.Row
|
|
child = con.execute(
|
|
"""SELECT ia.linkMode, ia.contentType FROM itemAttachments ia
|
|
JOIN items p ON p.itemID = ia.parentItemID
|
|
WHERE p.key = 'RUEFR423'"""
|
|
).fetchone()
|
|
assert (child["linkMode"], child["contentType"]) == (3, "text/html")
|
|
url = con.execute(
|
|
"""SELECT idv.value FROM itemAttachments ia
|
|
JOIN itemData d ON d.itemID = ia.itemID
|
|
JOIN itemDataValues idv ON idv.valueID = d.valueID
|
|
JOIN fields f ON f.fieldID = d.fieldID
|
|
WHERE f.fieldName = 'url'"""
|
|
).fetchone()[0]
|
|
assert url == "https://example.test/doc#p-2"
|
|
con.close()
|
|
store.close()
|
|
|
|
def test_second_push_is_idempotent(self, tmp_path) -> None:
|
|
db_path = str(tmp_path / "zotero.sqlite")
|
|
_make_zotero_db(db_path).close()
|
|
store, key = self._bib_with_placed_link(tmp_path)
|
|
|
|
push_to_zotero([store.get(key)], store=store, zotero_db=db_path)
|
|
stats = push_to_zotero([store.get(key)], store=store, zotero_db=db_path)
|
|
assert stats["fr_links"] == 0
|
|
|
|
con = sqlite3.connect(db_path)
|
|
n = con.execute("SELECT count(*) FROM itemAttachments").fetchone()[0]
|
|
assert n == 1
|
|
con.close()
|
|
store.close()
|
|
|
|
|
|
class TestSyncCfrLinks:
|
|
def test_cfr_placed_row_syncs_as_link_child(self, tmp_path) -> None:
|
|
"""#640: a CFR-placed fr_links row (page=0, p_id NULL) rides the
|
|
same linkMode=3 path as FR links."""
|
|
from bib import cfrlink
|
|
from bib.store import Store
|
|
|
|
db_path = str(tmp_path / "zotero.sqlite")
|
|
_make_zotero_db(db_path).close()
|
|
|
|
store = Store(str(tmp_path / "bib.sqlite"))
|
|
key = store.create(
|
|
Regulation(
|
|
key="REGCFR42",
|
|
title="42 CFR Part 425",
|
|
url="https://www.ecfr.gov/current/title-42/part-425",
|
|
)
|
|
)
|
|
cfrlink.place(store, "42 CFR 425.402", label="assignment methodology")
|
|
|
|
stats = push_to_zotero([store.get(key)], store=store, zotero_db=db_path)
|
|
assert stats["fr_links"] == 1
|
|
|
|
con = sqlite3.connect(db_path)
|
|
con.row_factory = sqlite3.Row
|
|
url = con.execute(
|
|
"""SELECT idv.value FROM itemAttachments ia
|
|
JOIN itemData d ON d.itemID = ia.itemID
|
|
JOIN itemDataValues idv ON idv.valueID = d.valueID
|
|
JOIN fields f ON f.fieldID = d.fieldID
|
|
WHERE ia.linkMode = 3 AND f.fieldName = 'url'"""
|
|
).fetchone()[0]
|
|
assert url == "https://www.ecfr.gov/current/title-42/section-425.402"
|
|
con.close()
|
|
store.close()
|