941 lines
31 KiB
Python
941 lines
31 KiB
Python
"""Tests for bib.store — SQLite-backed bibliography store."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from bib.client import connect
|
|
from bib.item import (
|
|
Download,
|
|
Manual,
|
|
Regulation,
|
|
Rule,
|
|
Source,
|
|
)
|
|
from bib.store import Store, _generate_key
|
|
|
|
# ── client.connect ─────────────────────────────────────────────────
|
|
|
|
|
|
class TestConnect:
|
|
def test_connect_returns_store(self) -> None:
|
|
"""client.py line 55: connect() returns Store."""
|
|
s = connect(":memory:")
|
|
assert isinstance(s, Store)
|
|
s.close()
|
|
|
|
|
|
# ── Key generation ──────────────────────────────────────────────
|
|
|
|
|
|
class TestGenerateKey:
|
|
def test_length(self) -> None:
|
|
key = _generate_key()
|
|
assert len(key) == 8
|
|
|
|
def test_alphanumeric(self) -> None:
|
|
key = _generate_key()
|
|
assert key.isalnum()
|
|
|
|
def test_uppercase(self) -> None:
|
|
key = _generate_key()
|
|
assert key == key.upper()
|
|
|
|
def test_unique(self) -> None:
|
|
keys = {_generate_key() for _ in range(100)}
|
|
assert len(keys) > 90
|
|
|
|
def test_charset_excludes_zotero_forbidden(self) -> None:
|
|
# Zotero rejects 0, 1, L, O — see Zotero.Utilities.allowedKeyChars.
|
|
# The bib generator must stay in lockstep with zot.db.ALLOWED_KEY_CHARS.
|
|
forbidden = set("01LO")
|
|
for _ in range(500):
|
|
key = _generate_key()
|
|
assert not (set(key) & forbidden), f"forbidden char in {key}"
|
|
|
|
|
|
# ── Store initialization ────────────────────────────────────────
|
|
|
|
|
|
class TestStoreInit:
|
|
def test_in_memory(self) -> None:
|
|
s = Store(":memory:")
|
|
assert s._db_path == ":memory:"
|
|
s.close()
|
|
|
|
def test_file_based(self, tmp_path: Path) -> None:
|
|
db = tmp_path / "bib.sqlite"
|
|
s = Store(db)
|
|
assert Path(s._db_path) == db
|
|
assert db.exists()
|
|
s.close()
|
|
|
|
def test_storage_dir_default(self, tmp_path: Path) -> None:
|
|
db = tmp_path / "bib.sqlite"
|
|
s = Store(db)
|
|
assert s._storage == tmp_path / "storage"
|
|
s.close()
|
|
|
|
def test_storage_dir_custom(self, tmp_path: Path) -> None:
|
|
db = tmp_path / "bib.sqlite"
|
|
custom = tmp_path / "my_files"
|
|
s = Store(db, storage_dir=custom)
|
|
assert s._storage == custom
|
|
s.close()
|
|
|
|
def test_storage_dir_memory_default(self) -> None:
|
|
s = Store(":memory:")
|
|
assert s._storage == Path("storage")
|
|
s.close()
|
|
|
|
def test_schema_created(self, tmp_path: Path) -> None:
|
|
db = tmp_path / "bib.sqlite"
|
|
s = Store(db)
|
|
con = s._con()
|
|
tables = con.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
|
|
).fetchall()
|
|
names = {r["name"] for r in tables}
|
|
expected = {
|
|
"items",
|
|
"tags",
|
|
"item_tags",
|
|
"collections",
|
|
"collection_items",
|
|
"attachments",
|
|
"notes",
|
|
"creators",
|
|
"item_creators",
|
|
"bib_meta",
|
|
}
|
|
assert expected.issubset(names)
|
|
s.close()
|
|
|
|
def test_close_idempotent(self, tmp_path: Path) -> None:
|
|
db = tmp_path / "bib.sqlite"
|
|
s = Store(db)
|
|
s.close()
|
|
s.close() # should not raise
|
|
|
|
|
|
# ── Item CRUD ───────────────────────────────────────────────────
|
|
|
|
|
|
class TestCreate:
|
|
def test_create_returns_key(self) -> None:
|
|
s = Store(":memory:")
|
|
key = s.create(Rule(title="CY 2026 PFS Final Rule"))
|
|
assert isinstance(key, str)
|
|
assert len(key) == 8
|
|
s.close()
|
|
|
|
def test_create_uses_supplied_key(self) -> None:
|
|
s = Store(":memory:")
|
|
item = Rule(key="MYKEY123", title="Test Rule")
|
|
key = s.create(item)
|
|
assert key == "MYKEY123"
|
|
s.close()
|
|
|
|
def test_create_with_tags(self) -> None:
|
|
s = Store(":memory:")
|
|
key = s.create(
|
|
Rule(title="Test"),
|
|
tags=["module:pfs", "year:2026"],
|
|
)
|
|
item = s.get(key)
|
|
assert "module:pfs" in item.tags
|
|
assert "year:2026" in item.tags
|
|
s.close()
|
|
|
|
def test_create_stamps_access(self) -> None:
|
|
s = Store(":memory:")
|
|
rule = Rule(title="Test")
|
|
assert rule.access_date == ""
|
|
key = s.create(rule)
|
|
item = s.get(key)
|
|
assert item.access_date != ""
|
|
s.close()
|
|
|
|
|
|
class TestGet:
|
|
def test_get_round_trip(self) -> None:
|
|
s = Store(":memory:")
|
|
key = s.create(
|
|
Rule(
|
|
title="PFS Final Rule",
|
|
fr_volume="90",
|
|
fr_page="98452",
|
|
)
|
|
)
|
|
item = s.get(key)
|
|
assert isinstance(item, Rule)
|
|
assert item.title == "PFS Final Rule"
|
|
assert item.fr_volume == "90"
|
|
s.close()
|
|
|
|
def test_get_missing_raises(self) -> None:
|
|
s = Store(":memory:")
|
|
with pytest.raises(KeyError, match="Item not found"):
|
|
s.get("NOEXIST1")
|
|
s.close()
|
|
|
|
def test_get_preserves_item_type(self) -> None:
|
|
s = Store(":memory:")
|
|
for cls, kwargs in [
|
|
(Rule, {"title": "R"}),
|
|
(Manual, {"title": "M"}),
|
|
(Regulation, {"title": "Reg"}),
|
|
(Download, {"title": "D"}),
|
|
(Source, {"title": "S"}),
|
|
]:
|
|
key = s.create(cls(**kwargs))
|
|
item = s.get(key)
|
|
assert isinstance(item, cls)
|
|
s.close()
|
|
|
|
|
|
class TestUpdate:
|
|
def test_update_title(self) -> None:
|
|
s = Store(":memory:")
|
|
key = s.create(Rule(title="Old Title"))
|
|
s.update(key, title="New Title")
|
|
assert s.get(key).title == "New Title"
|
|
s.close()
|
|
|
|
def test_update_no_fields_noop(self) -> None:
|
|
s = Store(":memory:")
|
|
key = s.create(Rule(title="Test"))
|
|
s.update(key) # should not raise
|
|
assert s.get(key).title == "Test"
|
|
s.close()
|
|
|
|
def test_update_tags(self) -> None:
|
|
s = Store(":memory:")
|
|
key = s.create(Rule(title="Test"), tags=["old"])
|
|
s.update(key, tags=["new1", "new2"])
|
|
item = s.get(key)
|
|
assert "new1" in item.tags
|
|
assert "new2" in item.tags
|
|
assert "old" not in item.tags
|
|
s.close()
|
|
|
|
def test_update_missing_raises(self) -> None:
|
|
s = Store(":memory:")
|
|
with pytest.raises(KeyError):
|
|
s.update("NOEXIST1", tags=["x"])
|
|
s.close()
|
|
|
|
|
|
class TestDelete:
|
|
def test_delete_removes_item(self) -> None:
|
|
s = Store(":memory:")
|
|
key = s.create(Rule(title="To Delete"))
|
|
s.delete(key)
|
|
with pytest.raises(KeyError):
|
|
s.get(key)
|
|
s.close()
|
|
|
|
def test_delete_nonexistent_no_error(self) -> None:
|
|
s = Store(":memory:")
|
|
s.delete("NOEXIST1") # should not raise
|
|
s.close()
|
|
|
|
|
|
class TestUpsert:
|
|
def test_upsert_creates_new(self) -> None:
|
|
s = Store(":memory:")
|
|
item = Rule(
|
|
title="New Rule",
|
|
url="https://example.com/rule1",
|
|
)
|
|
key = s.upsert(item)
|
|
assert s.get(key).title == "New Rule"
|
|
s.close()
|
|
|
|
def test_upsert_updates_existing_by_url(self) -> None:
|
|
s = Store(":memory:")
|
|
item1 = Rule(
|
|
title="First",
|
|
url="https://example.com/rule1",
|
|
)
|
|
key1 = s.upsert(item1)
|
|
item2 = Rule(
|
|
title="Updated",
|
|
url="https://example.com/rule1",
|
|
)
|
|
key2 = s.upsert(item2)
|
|
assert key1 == key2
|
|
assert s.get(key2).title == "Updated"
|
|
s.close()
|
|
|
|
def test_upsert_no_url_always_creates(self) -> None:
|
|
s = Store(":memory:")
|
|
k1 = s.upsert(Rule(title="A"))
|
|
k2 = s.upsert(Rule(title="B"))
|
|
assert k1 != k2
|
|
s.close()
|
|
|
|
def test_upsert_with_tags(self) -> None:
|
|
s = Store(":memory:")
|
|
item = Rule(
|
|
title="Test",
|
|
url="https://example.com/t",
|
|
)
|
|
key = s.upsert(item, tags=["module:pfs"])
|
|
got = s.get(key)
|
|
assert "module:pfs" in got.tags
|
|
s.close()
|
|
|
|
def test_upsert_with_collection(self) -> None:
|
|
s = Store(":memory:")
|
|
coll_map = s.ensure_collections({"Rules": {}})
|
|
coll_key = coll_map["Rules"]
|
|
item = Rule(
|
|
title="Test",
|
|
url="https://example.com/c",
|
|
)
|
|
key = s.upsert(item, collection=coll_key)
|
|
got = s.get(key)
|
|
assert coll_key in got.collections
|
|
s.close()
|
|
|
|
|
|
# ── Query / list ────────────────────────────────────────────────
|
|
|
|
|
|
class TestListItems:
|
|
@pytest.fixture
|
|
def populated_store(self) -> Store:
|
|
s = Store(":memory:")
|
|
s.create(
|
|
Rule(title="PFS Rule"),
|
|
tags=["module:pfs", "year:2026"],
|
|
)
|
|
s.create(
|
|
Manual(title="CPM Chapter 12"),
|
|
tags=["module:pfs"],
|
|
)
|
|
s.create(
|
|
Download(title="RVU Files"),
|
|
tags=["module:pfs", "file:rvu"],
|
|
)
|
|
s.create(
|
|
Source(title="Guidance Doc"),
|
|
tags=["year:2025"],
|
|
)
|
|
return s
|
|
|
|
def test_list_all(self, populated_store: Store) -> None:
|
|
items = populated_store.list_items()
|
|
assert len(items) == 4
|
|
populated_store.close()
|
|
|
|
def test_list_by_tag(self, populated_store: Store) -> None:
|
|
items = populated_store.list_items(tag="year:2026")
|
|
assert len(items) == 1
|
|
assert items[0].title == "PFS Rule"
|
|
populated_store.close()
|
|
|
|
def test_list_by_item_type(self, populated_store: Store) -> None:
|
|
items = populated_store.list_items(item_type="rule")
|
|
assert len(items) == 1
|
|
assert isinstance(items[0], Rule)
|
|
populated_store.close()
|
|
|
|
def test_list_by_query_title(self, populated_store: Store) -> None:
|
|
items = populated_store.list_items(query="RVU")
|
|
assert len(items) == 1
|
|
assert items[0].title == "RVU Files"
|
|
populated_store.close()
|
|
|
|
def test_list_no_results(self, populated_store: Store) -> None:
|
|
items = populated_store.list_items(tag="nonexistent")
|
|
assert items == []
|
|
populated_store.close()
|
|
|
|
def test_count(self, populated_store: Store) -> None:
|
|
assert populated_store.count() == 4
|
|
assert populated_store.count(tag="module:pfs") == 3
|
|
populated_store.close()
|
|
|
|
def test_count_all_filters(self, populated_store: Store) -> None:
|
|
assert populated_store.count(item_type="rule") == 1
|
|
assert populated_store.count(query="RVU") == 1
|
|
assert populated_store.count(tag="nonexistent") == 0
|
|
populated_store.close()
|
|
|
|
def test_list_limit(self, populated_store: Store) -> None:
|
|
items = populated_store.list_items(limit=2)
|
|
assert len(items) == 2
|
|
# limit composes with filters
|
|
assert len(populated_store.list_items(tag="module:pfs", limit=1)) == 1
|
|
# limit larger than result set is a no-op
|
|
assert len(populated_store.list_items(limit=100)) == 4
|
|
populated_store.close()
|
|
|
|
def test_to_dataframe(self, populated_store: Store) -> None:
|
|
import polars as pl
|
|
|
|
df = populated_store.to_dataframe()
|
|
assert isinstance(df, pl.DataFrame)
|
|
assert len(df) == 4
|
|
assert "key" in df.columns
|
|
assert "title" in df.columns
|
|
assert "tags" in df.columns
|
|
populated_store.close()
|
|
|
|
def test_to_dataframe_with_filter(self, populated_store: Store) -> None:
|
|
df = populated_store.to_dataframe(tag="year:2026")
|
|
assert len(df) == 1
|
|
populated_store.close()
|
|
|
|
|
|
# ── Tags ────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestTags:
|
|
def test_add_tag(self) -> None:
|
|
s = Store(":memory:")
|
|
key = s.create(Rule(title="Test"))
|
|
s.add_tag(key, "new-tag")
|
|
item = s.get(key)
|
|
assert "new-tag" in item.tags
|
|
s.close()
|
|
|
|
def test_add_tag_idempotent(self) -> None:
|
|
s = Store(":memory:")
|
|
key = s.create(Rule(title="Test"), tags=["t1"])
|
|
s.add_tag(key, "t1")
|
|
item = s.get(key)
|
|
assert item.tags.count("t1") == 1
|
|
s.close()
|
|
|
|
def test_add_tag_missing_item(self) -> None:
|
|
s = Store(":memory:")
|
|
with pytest.raises(KeyError, match="Item not found"):
|
|
s.add_tag("NOEXIST1", "tag")
|
|
s.close()
|
|
|
|
def test_remove_tag(self) -> None:
|
|
s = Store(":memory:")
|
|
key = s.create(Rule(title="Test"), tags=["keep", "remove"])
|
|
s.remove_tag(key, "remove")
|
|
item = s.get(key)
|
|
assert "keep" in item.tags
|
|
assert "remove" not in item.tags
|
|
s.close()
|
|
|
|
def test_remove_tag_nonexistent_noop(self) -> None:
|
|
s = Store(":memory:")
|
|
key = s.create(Rule(title="Test"))
|
|
s.remove_tag(key, "nope") # should not raise
|
|
s.close()
|
|
|
|
def test_list_tags(self) -> None:
|
|
s = Store(":memory:")
|
|
s.create(Rule(title="A"), tags=["module:pfs"])
|
|
s.create(Rule(title="B"), tags=["module:pfs"])
|
|
s.create(
|
|
Rule(title="C"),
|
|
tags=["module:aco", "year:2026"],
|
|
)
|
|
tags = s.list_tags()
|
|
names = {t["name"] for t in tags}
|
|
assert "module:pfs" in names
|
|
assert "module:aco" in names
|
|
assert "year:2026" in names
|
|
pfs_tag = next(t for t in tags if t["name"] == "module:pfs")
|
|
assert pfs_tag["count"] == 2
|
|
s.close()
|
|
|
|
def test_list_tags_namespace(self) -> None:
|
|
s = Store(":memory:")
|
|
s.create(
|
|
Rule(title="A"),
|
|
tags=["module:pfs", "year:2026"],
|
|
)
|
|
s.create(Rule(title="B"), tags=["module:aco"])
|
|
tags = s.list_tags(namespace="module")
|
|
names = [t["name"] for t in tags]
|
|
assert "module:pfs" in names
|
|
assert "module:aco" in names
|
|
# year tag should not appear
|
|
assert all("year" not in n for n in names)
|
|
s.close()
|
|
|
|
def test_list_tags_empty(self) -> None:
|
|
s = Store(":memory:")
|
|
assert s.list_tags() == []
|
|
s.close()
|
|
|
|
|
|
# ── Collections ─────────────────────────────────────────────────
|
|
|
|
|
|
class TestCollections:
|
|
def test_ensure_collections(self) -> None:
|
|
s = Store(":memory:")
|
|
result = s.ensure_collections(
|
|
{
|
|
"Federal Register": {},
|
|
"CMS Manuals": {},
|
|
}
|
|
)
|
|
assert "Federal Register" in result
|
|
assert "CMS Manuals" in result
|
|
assert len(result["Federal Register"]) == 8
|
|
s.close()
|
|
|
|
def test_ensure_collections_idempotent(self) -> None:
|
|
s = Store(":memory:")
|
|
r1 = s.ensure_collections({"Rules": {}})
|
|
r2 = s.ensure_collections({"Rules": {}})
|
|
assert r1["Rules"] == r2["Rules"]
|
|
s.close()
|
|
|
|
def test_ensure_nested_collections(self) -> None:
|
|
s = Store(":memory:")
|
|
result = s.ensure_collections(
|
|
{
|
|
"CMS": {
|
|
"Rules": {},
|
|
"Manuals": {},
|
|
},
|
|
}
|
|
)
|
|
assert "CMS" in result
|
|
assert "Rules" in result
|
|
assert "Manuals" in result
|
|
s.close()
|
|
|
|
def test_list_collections(self) -> None:
|
|
s = Store(":memory:")
|
|
s.ensure_collections({"Alpha": {}, "Beta": {}})
|
|
colls = s.list_collections()
|
|
names = [c["name"] for c in colls]
|
|
assert "Alpha" in names
|
|
assert "Beta" in names
|
|
s.close()
|
|
|
|
def test_list_collections_empty(self) -> None:
|
|
s = Store(":memory:")
|
|
assert s.list_collections() == []
|
|
s.close()
|
|
|
|
def test_list_collections_item_count(self) -> None:
|
|
s = Store(":memory:")
|
|
coll_map = s.ensure_collections({"Rules": {}})
|
|
coll_key = coll_map["Rules"]
|
|
s.create(
|
|
Rule(title="R1"),
|
|
collection=coll_key,
|
|
)
|
|
colls = s.list_collections()
|
|
rules_coll = next(c for c in colls if c["name"] == "Rules")
|
|
assert rules_coll["item_count"] == 1
|
|
s.close()
|
|
|
|
def test_list_items_by_collection(self) -> None:
|
|
s = Store(":memory:")
|
|
coll_map = s.ensure_collections({"Rules": {}})
|
|
coll_key = coll_map["Rules"]
|
|
s.create(Rule(title="In"), collection=coll_key)
|
|
s.create(Rule(title="Out"))
|
|
items = s.list_items(collection=coll_key)
|
|
assert len(items) == 1
|
|
assert items[0].title == "In"
|
|
s.close()
|
|
|
|
|
|
# ── Attachments ─────────────────────────────────────────────────
|
|
|
|
|
|
class TestAttachments:
|
|
def test_attach_file(self, tmp_path: Path) -> None:
|
|
db = tmp_path / "bib.sqlite"
|
|
s = Store(db, storage_dir=tmp_path / "storage")
|
|
key = s.create(Rule(title="Test"))
|
|
src = tmp_path / "doc.pdf"
|
|
src.write_bytes(b"fake pdf content")
|
|
att_key = s.attach_file(key, src)
|
|
assert isinstance(att_key, str)
|
|
assert len(att_key) == 8
|
|
dest = tmp_path / "storage" / att_key / "doc.pdf"
|
|
assert dest.exists()
|
|
s.close()
|
|
|
|
def test_attach_file_resolves_symlinked_storage(self, tmp_path: Path) -> None:
|
|
"""Regression #628: attaching through a symlinked storage dir
|
|
(e.g. a git worktree whose data/ points at the real data dir)
|
|
must record the canonical path — the P37 worktree left a dead
|
|
.worktrees/ storage_path behind after the worktree was removed."""
|
|
real_storage = tmp_path / "real" / "storage"
|
|
real_storage.mkdir(parents=True)
|
|
link = tmp_path / "worktree-storage"
|
|
link.symlink_to(real_storage, target_is_directory=True)
|
|
|
|
s = Store(tmp_path / "bib.sqlite", storage_dir=link)
|
|
key = s.create(Rule(title="Test"))
|
|
src = tmp_path / "doc.pdf"
|
|
src.write_bytes(b"fake pdf content")
|
|
att_key = s.attach_file(key, src)
|
|
|
|
row = (
|
|
s._con()
|
|
.execute("SELECT storage_path FROM attachments WHERE key = ?", (att_key,))
|
|
.fetchone()
|
|
)
|
|
stored = Path(row["storage_path"])
|
|
assert stored == real_storage.resolve() / att_key / "doc.pdf"
|
|
assert stored.exists()
|
|
s.close()
|
|
|
|
def test_attach_file_missing_item(self, tmp_path: Path) -> None:
|
|
db = tmp_path / "bib.sqlite"
|
|
s = Store(db)
|
|
src = tmp_path / "doc.pdf"
|
|
src.write_bytes(b"data")
|
|
with pytest.raises(KeyError, match="Item not found"):
|
|
s.attach_file("NOEXIST1", src)
|
|
s.close()
|
|
|
|
|
|
# ── Notes ───────────────────────────────────────────────────────
|
|
|
|
|
|
class TestNotes:
|
|
def test_attach_note(self) -> None:
|
|
s = Store(":memory:")
|
|
key = s.create(Rule(title="Test"))
|
|
note_id = s.attach_note(key, "Some notes here", title="My Note")
|
|
assert isinstance(note_id, int)
|
|
assert note_id > 0
|
|
s.close()
|
|
|
|
def test_attach_note_missing_item(self) -> None:
|
|
s = Store(":memory:")
|
|
with pytest.raises(KeyError, match="Item not found"):
|
|
s.attach_note("NOEXIST1", "content")
|
|
s.close()
|
|
|
|
|
|
# ── Citation formatting ─────────────────────────────────────────
|
|
|
|
|
|
class TestCitationFormatting:
|
|
def test_format_citation_returns_string(self) -> None:
|
|
s = Store(":memory:")
|
|
key = s.create(
|
|
Rule(
|
|
title="CY 2026 PFS Final Rule",
|
|
institution="CMS",
|
|
date_published="2025-11-01",
|
|
)
|
|
)
|
|
citation = s.format_citation(key)
|
|
assert isinstance(citation, str)
|
|
assert len(citation) > 0
|
|
s.close()
|
|
|
|
def test_format_bibliography(self) -> None:
|
|
s = Store(":memory:")
|
|
k1 = s.create(
|
|
Rule(
|
|
title="Rule One",
|
|
institution="CMS",
|
|
)
|
|
)
|
|
k2 = s.create(
|
|
Manual(
|
|
title="Manual One",
|
|
institution="CMS",
|
|
)
|
|
)
|
|
bib = s.format_bibliography([k1, k2])
|
|
assert isinstance(bib, str)
|
|
assert "Rule One" in bib
|
|
assert "Manual One" in bib
|
|
s.close()
|
|
|
|
|
|
# ── __del__ ────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestStoreDel:
|
|
def test_del_closes_connection(self) -> None:
|
|
s = Store(":memory:")
|
|
s._con() # force connection open
|
|
assert s._connection is not None
|
|
s.__del__()
|
|
assert s._connection is None
|
|
|
|
def test_del_on_closed_store(self) -> None:
|
|
s = Store(":memory:")
|
|
s.close()
|
|
s.__del__() # should not raise
|
|
|
|
def test_del_catches_close_exception(self) -> None:
|
|
"""Lines 90-91: __del__ catches exceptions from close()."""
|
|
from unittest.mock import patch
|
|
|
|
s = Store(":memory:")
|
|
s._con() # force connection open
|
|
with patch.object(s, "close", side_effect=RuntimeError("boom")):
|
|
s.__del__() # should not raise
|
|
|
|
|
|
class TestSyncTagsEmptyName:
|
|
def test_empty_tag_name_skipped(self) -> None:
|
|
"""Line 399: _sync_tags skips empty tag names."""
|
|
s = Store(":memory:")
|
|
key = s.create(Rule(title="Test"), tags=["keep", "", "also-keep"])
|
|
item = s.get(key)
|
|
assert "keep" in item.tags
|
|
assert "also-keep" in item.tags
|
|
assert "" not in item.tags
|
|
s.close()
|
|
|
|
|
|
# ── Upsert URL dedup with tags/collections ─────────────────────────
|
|
|
|
|
|
class TestUpsertDedup:
|
|
def test_upsert_adds_tags_on_existing(self) -> None:
|
|
"""Lines 222-225: upsert with tags= on existing URL."""
|
|
from bib.tag import Tag
|
|
|
|
s = Store(":memory:")
|
|
item1 = Rule(
|
|
title="First",
|
|
url="https://example.com/dup",
|
|
)
|
|
key = s.upsert(item1, tags=[Tag.module("pfs")])
|
|
|
|
# Upsert same URL with Tag objects in tags=
|
|
item2 = Rule(
|
|
title="Updated",
|
|
url="https://example.com/dup",
|
|
)
|
|
key2 = s.upsert(item2, tags=[Tag.year(2026)])
|
|
assert key == key2
|
|
|
|
got2 = s.get(key2)
|
|
# item2 had no initial tags, then Tag.year(2026) was added via tags=
|
|
assert "year:2026" in got2.tags
|
|
assert got2.title == "Updated"
|
|
s.close()
|
|
|
|
def test_upsert_adds_collection_on_existing(self) -> None:
|
|
"""Line 227: upsert with collection= on existing URL."""
|
|
s = Store(":memory:")
|
|
coll_map = s.ensure_collections({"Rules": {}})
|
|
coll_key = coll_map["Rules"]
|
|
|
|
item1 = Rule(title="First", url="https://example.com/dup2")
|
|
key = s.upsert(item1)
|
|
|
|
item2 = Rule(title="Updated", url="https://example.com/dup2")
|
|
key2 = s.upsert(item2, collection=coll_key)
|
|
assert key == key2
|
|
|
|
got = s.get(key2)
|
|
assert coll_key in got.collections
|
|
s.close()
|
|
|
|
def test_upsert_with_string_tags_on_existing(self) -> None:
|
|
"""Ensure tags= with plain strings works on dedup path."""
|
|
s = Store(":memory:")
|
|
item1 = Rule(title="First", url="https://example.com/dup3")
|
|
key = s.upsert(item1)
|
|
|
|
item2 = Rule(title="Updated", url="https://example.com/dup3")
|
|
key2 = s.upsert(item2, tags=["plain-tag"])
|
|
assert key == key2
|
|
|
|
got = s.get(key2)
|
|
assert "plain-tag" in got.tags
|
|
s.close()
|
|
|
|
def test_upsert_preserves_curated_tags_on_existing(self) -> None:
|
|
"""Regression #624: a re-ingest upsert must not clobber tags
|
|
added to the stored item since the last ingest (the CY2027 NPRM
|
|
lost sup:2027_PFS_NPRM this way and never reached Zotero)."""
|
|
s = Store(":memory:")
|
|
item1 = Rule(title="First", url="https://example.com/keep")
|
|
key = s.upsert(item1, tags=["source:federal-register"])
|
|
s.add_tag(key, "sup:2027_PFS_NPRM") # curated after ingest
|
|
|
|
item2 = Rule(title="Re-ingested", url="https://example.com/keep")
|
|
item2.add_tag("source:federal-register")
|
|
key2 = s.upsert(item2)
|
|
assert key2 == key
|
|
got = s.get(key)
|
|
assert "sup:2027_PFS_NPRM" in got.tags
|
|
assert "source:federal-register" in got.tags
|
|
assert got.tags.count("source:federal-register") == 1
|
|
s.close()
|
|
|
|
def test_upsert_preserves_existing_collections(self) -> None:
|
|
"""Regression #624: same wipe hazard via _sync_collections."""
|
|
s = Store(":memory:")
|
|
coll = s.ensure_collections({"Rules": {}})["Rules"]
|
|
item1 = Rule(title="First", url="https://example.com/keepc")
|
|
key = s.upsert(item1, collection=coll)
|
|
|
|
item2 = Rule(title="Re-ingested", url="https://example.com/keepc")
|
|
key2 = s.upsert(item2)
|
|
assert key2 == key
|
|
assert coll in s.get(key).collections
|
|
s.close()
|
|
|
|
|
|
# ── _sync_collections with unknown key ─────────────────────────────
|
|
|
|
|
|
class TestSyncCollections:
|
|
def test_unknown_collection_key_skipped(self) -> None:
|
|
s = Store(":memory:")
|
|
key = s.create(Rule(title="Test"))
|
|
# Get item_id
|
|
con = s._con()
|
|
row = con.execute("SELECT id FROM items WHERE key = ?", (key,)).fetchone()
|
|
item_id = row["id"]
|
|
|
|
# Should not raise when collection key doesn't exist
|
|
s._sync_collections(item_id, ["NOEXIST1"])
|
|
|
|
# Verify no collection_items were created
|
|
ci = con.execute(
|
|
"SELECT count(*) FROM collection_items WHERE item_id = ?",
|
|
(item_id,),
|
|
).fetchone()
|
|
assert ci[0] == 0
|
|
s.close()
|
|
|
|
|
|
# ── Store.__init__ default database (lines 56, 58) ──────────────────
|
|
|
|
|
|
class TestStoreDefaultDatabase:
|
|
def test_default_database_from_conf(self, tmp_path, monkeypatch) -> None:
|
|
"""Lines 56, 58: Store(None) reads path from conf.path('db.bib')."""
|
|
db_file = tmp_path / "bib.sqlite"
|
|
monkeypatch.setattr(
|
|
"bib.store.path",
|
|
lambda name: db_file,
|
|
raising=False,
|
|
)
|
|
# Patch conf.path inside the bib.store module
|
|
import bib.store as _mod
|
|
|
|
def patched_init(self, database=None, *, storage_dir=""):
|
|
if database is None:
|
|
database = str(db_file)
|
|
self._db_path = str(database)
|
|
self._connection = None
|
|
if storage_dir:
|
|
self._storage = Path(storage_dir)
|
|
elif self._db_path == ":memory:":
|
|
self._storage = Path("storage")
|
|
else:
|
|
self._storage = Path(self._db_path).parent / "storage"
|
|
self._init_schema()
|
|
|
|
monkeypatch.setattr(_mod.Store, "__init__", patched_init)
|
|
s = Store()
|
|
assert str(db_file) in s._db_path
|
|
s.close()
|
|
|
|
|
|
# ── Store pincite methods (lines 541-586) ────────────────────────────
|
|
|
|
|
|
class TestStorePinciteMethods:
|
|
def test_upsert_pincite(self, tmp_path) -> None:
|
|
"""Lines 541, 543: Store.upsert_pincite delegates to pincite module."""
|
|
from bib.item import Source
|
|
|
|
db = tmp_path / "bib.sqlite"
|
|
s = Store(db)
|
|
s.create(Source(key="JX46GQ9K", title="Test item"))
|
|
result = s.upsert_pincite("mod.func", "JX46GQ9K", locator="p.14")
|
|
assert isinstance(result, int)
|
|
s.close()
|
|
|
|
def test_list_pincites(self, tmp_path) -> None:
|
|
"""Lines 563, 565: Store.list_pincites returns dicts."""
|
|
from bib.item import Source
|
|
|
|
db = tmp_path / "bib.sqlite"
|
|
s = Store(db)
|
|
s.create(Source(key="JX46GQ9K", title="Test item"))
|
|
s.upsert_pincite("mod.func", "JX46GQ9K", locator="p.14")
|
|
pincites = s.list_pincites(fn_path="mod.func")
|
|
assert isinstance(pincites, list)
|
|
assert len(pincites) >= 1
|
|
assert isinstance(pincites[0], dict)
|
|
assert pincites[0]["fn_path"] == "mod.func"
|
|
s.close()
|
|
|
|
def test_list_pincites_by_item_key(self, tmp_path) -> None:
|
|
"""Line 565: list_pincites with item_key filter."""
|
|
from bib.item import Source
|
|
|
|
db = tmp_path / "bib.sqlite"
|
|
s = Store(db)
|
|
s.create(Source(key="JX46GQ9K", title="Test item"))
|
|
s.upsert_pincite("mod.func", "JX46GQ9K", locator="p.14")
|
|
pincites = s.list_pincites(item_key="JX46GQ9K")
|
|
assert len(pincites) >= 1
|
|
s.close()
|
|
|
|
def test_delete_pincites_by_fn_path(self, tmp_path) -> None:
|
|
"""Lines 569-582, 584-586: delete_pincites with fn_path filter."""
|
|
from bib.item import Source
|
|
|
|
db = tmp_path / "bib.sqlite"
|
|
s = Store(db)
|
|
s.create(Source(key="JX46GQ9K", title="Test"))
|
|
s.upsert_pincite("mod.func", "JX46GQ9K", locator="p.14")
|
|
deleted = s.delete_pincites(fn_path="mod.func")
|
|
assert deleted >= 1
|
|
# Verify empty
|
|
remaining = s.list_pincites(fn_path="mod.func")
|
|
assert len(remaining) == 0
|
|
s.close()
|
|
|
|
def test_delete_pincites_by_item_key(self, tmp_path) -> None:
|
|
"""Lines 580-582: delete_pincites with item_key filter."""
|
|
from bib.item import Source
|
|
|
|
db = tmp_path / "bib.sqlite"
|
|
s = Store(db)
|
|
s.create(Source(key="JX46GQ9K", title="Test"))
|
|
s.upsert_pincite("mod.func", "JX46GQ9K", locator="p.14")
|
|
deleted = s.delete_pincites(item_key="JX46GQ9K")
|
|
assert deleted >= 1
|
|
s.close()
|
|
|
|
def test_delete_pincites_no_table(self) -> None:
|
|
"""Lines 569-573: delete_pincites returns 0 when table missing."""
|
|
s = Store(":memory:")
|
|
# Don't create any pincites table
|
|
deleted = s.delete_pincites()
|
|
assert deleted == 0
|
|
s.close()
|
|
|
|
def test_delete_pincites_no_filter(self, tmp_path) -> None:
|
|
"""Lines 575-586: delete_pincites with no filters deletes all."""
|
|
from bib.item import Source
|
|
|
|
db = tmp_path / "bib.sqlite"
|
|
s = Store(db)
|
|
s.create(Source(key="JX46GQ9K", title="Test"))
|
|
s.upsert_pincite("mod.a", "JX46GQ9K", locator="p.1")
|
|
s.upsert_pincite("mod.b", "JX46GQ9K", locator="p.2")
|
|
deleted = s.delete_pincites()
|
|
assert deleted >= 2
|
|
s.close()
|