d82bc4824f
- New deps: pymupdf>=1.24 (AGPL-3.0), python-docx>=1.1
- src/rex/comments/{__init__.py,extract.py} with ExtractResult dataclass
- PDF extraction via PyMuPDF with status taxonomy:
ok | ocr_needed | failed | unsupported
- Tests cover happy path, image-only (ocr_needed), and corrupted PDF
Also fixes 19 pre-existing test failures in tests/zot/test_{duck,extract,
table}.py — all were opening data/zotero/data/zotero.sqlite directly,
which fails with "database is locked" while the Zotero container holds
the WAL lock. New tests/zot/conftest.py provides a session-scoped
host_db fixture that snapshots the live DB once via shutil.copy2;
schema rows (itemTypes/fields/creatorTypes) are stable so a hot copy
is fine for these read-only schema-parity checks.
DOCX/text handlers and combine.py land in the next batch.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
388 lines
13 KiB
Python
388 lines
13 KiB
Python
"""Tests for zot.table — Pydantic models for all 61 Zotero SQLite tables.
|
|
|
|
Validates that models match the real Zotero schema by comparing against
|
|
the live zotero.sqlite database.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from conf.table_base import SQLTable
|
|
|
|
HOST_DB = Path("data/zotero/data/zotero.sqlite")
|
|
|
|
|
|
# ── Import smoke tests ──────────────────────────────────────────────
|
|
|
|
|
|
class TestImports:
|
|
"""All 8 submodules import cleanly."""
|
|
|
|
def test_core(self):
|
|
from zot.table.core import Items, Libraries
|
|
|
|
assert issubclass(Items, SQLTable)
|
|
assert issubclass(Libraries, SQLTable)
|
|
|
|
def test_fields(self):
|
|
from zot.table.fields import Fields, ItemData, ItemDataValues
|
|
|
|
assert issubclass(Fields, SQLTable)
|
|
assert issubclass(ItemData, SQLTable)
|
|
assert issubclass(ItemDataValues, SQLTable)
|
|
|
|
def test_creators(self):
|
|
from zot.table.creators import Creators, CreatorTypes, ItemCreators
|
|
|
|
assert issubclass(Creators, SQLTable)
|
|
assert issubclass(CreatorTypes, SQLTable)
|
|
assert issubclass(ItemCreators, SQLTable)
|
|
|
|
def test_collections(self):
|
|
from zot.table.collections import CollectionItems, Collections
|
|
|
|
assert issubclass(Collections, SQLTable)
|
|
assert issubclass(CollectionItems, SQLTable)
|
|
|
|
def test_tags(self):
|
|
from zot.table.tags import ItemTags, Tags
|
|
|
|
assert issubclass(Tags, SQLTable)
|
|
assert issubclass(ItemTags, SQLTable)
|
|
|
|
def test_attachments(self):
|
|
from zot.table.attachments import ItemAnnotations, ItemAttachments, ItemNotes
|
|
|
|
assert issubclass(ItemAttachments, SQLTable)
|
|
assert issubclass(ItemNotes, SQLTable)
|
|
assert issubclass(ItemAnnotations, SQLTable)
|
|
|
|
def test_search(self):
|
|
from zot.table.search import FulltextItems, FulltextWords, SavedSearches
|
|
|
|
assert issubclass(FulltextWords, SQLTable)
|
|
assert issubclass(FulltextItems, SQLTable)
|
|
assert issubclass(SavedSearches, SQLTable)
|
|
|
|
def test_sync(self):
|
|
from zot.table.sync import Settings, SyncCache, TranslatorCache, Version
|
|
|
|
assert issubclass(Settings, SQLTable)
|
|
assert issubclass(SyncCache, SQLTable)
|
|
assert issubclass(Version, SQLTable)
|
|
assert issubclass(TranslatorCache, SQLTable)
|
|
|
|
def test_wildcard_import(self):
|
|
import zot.table as zt
|
|
|
|
models = [
|
|
name
|
|
for name in dir(zt)
|
|
if not name.startswith("_")
|
|
and isinstance(getattr(zt, name), type)
|
|
and issubclass(getattr(zt, name), SQLTable)
|
|
and getattr(zt, name) is not SQLTable
|
|
]
|
|
assert len(models) == 61
|
|
|
|
|
|
# ── Model correctness ───────────────────────────────────────────────
|
|
|
|
|
|
class TestModelStructure:
|
|
"""Models have expected columns and types."""
|
|
|
|
def test_items_columns(self):
|
|
from zot.table.core import Items
|
|
|
|
cols = Items.column_names()
|
|
assert "itemID" in cols
|
|
assert "itemTypeID" in cols
|
|
assert "libraryID" in cols
|
|
assert "key" in cols
|
|
assert "synced" in cols
|
|
|
|
def test_items_instantiation(self):
|
|
from zot.table.core import Items
|
|
|
|
item = Items(itemTypeID=20, libraryID=1, key="TESTKEY1")
|
|
assert item.itemTypeID == 20
|
|
assert item.libraryID == 1
|
|
assert item.synced == 0
|
|
|
|
def test_fields_columns(self):
|
|
from zot.table.fields import Fields
|
|
|
|
cols = Fields.column_names()
|
|
assert cols == ["fieldID", "fieldName", "fieldFormatID"]
|
|
|
|
def test_item_data_eav(self):
|
|
from zot.table.fields import ItemData
|
|
|
|
row = ItemData(itemID=1, fieldID=110, valueID=42)
|
|
assert row.itemID == 1
|
|
assert row.fieldID == 110
|
|
|
|
def test_creators_columns(self):
|
|
from zot.table.creators import Creators
|
|
|
|
c = Creators(firstName="John", lastName="Doe", fieldMode=0)
|
|
assert c.firstName == "John"
|
|
assert c.lastName == "Doe"
|
|
|
|
def test_item_creators_defaults(self):
|
|
from zot.table.creators import ItemCreators
|
|
|
|
ic = ItemCreators(itemID=1, creatorID=2)
|
|
assert ic.creatorTypeID == 1 # author
|
|
assert ic.orderIndex == 0
|
|
|
|
def test_collections_columns(self):
|
|
from zot.table.collections import Collections
|
|
|
|
c = Collections(collectionName="Test", libraryID=1, key="ABCD1234")
|
|
assert c.collectionName == "Test"
|
|
assert c.synced == 0
|
|
|
|
def test_tags_columns(self):
|
|
from zot.table.tags import Tags
|
|
|
|
t = Tags(name="module:pfs")
|
|
assert t.name == "module:pfs"
|
|
|
|
def test_version_renamed_field(self):
|
|
from zot.table.sync import Version
|
|
|
|
v = Version(schema_name="userdata", version=127)
|
|
assert v.schema_name == "userdata"
|
|
assert v.version == 127
|
|
|
|
def test_ddl_generation(self):
|
|
from zot.table.core import Items
|
|
|
|
ddl = Items.to_ddl()
|
|
assert "CREATE TABLE" in ddl
|
|
assert "itemID" in ddl
|
|
assert "itemTypeID" in ddl
|
|
|
|
|
|
# ── Schema validation against real DB ────────────────────────────────
|
|
|
|
|
|
def _get_real_table_columns(db_path: Path, table_name: str) -> list[str]:
|
|
"""Get column names from a Zotero SQLite at *db_path*."""
|
|
con = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
|
|
cols = [r[1] for r in con.execute(f'PRAGMA table_info("{table_name}")').fetchall()]
|
|
con.close()
|
|
return cols
|
|
|
|
|
|
def _get_all_real_tables(db_path: Path) -> list[str]:
|
|
"""Get all table names from a Zotero SQLite at *db_path*."""
|
|
con = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
|
|
tables = [
|
|
r[0]
|
|
for r in con.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
|
|
).fetchall()
|
|
]
|
|
con.close()
|
|
return tables
|
|
|
|
|
|
@pytest.mark.skipif(not HOST_DB.exists(), reason="zotero.sqlite not available")
|
|
class TestSchemaMatchesRealDB:
|
|
"""Verify Pydantic models match the real Zotero SQLite schema.
|
|
|
|
Uses ``host_db`` (snapshot of live DB; see conftest.py) — opening
|
|
the live file directly fails with ``database is locked`` while the
|
|
Zotero container holds WAL.
|
|
"""
|
|
|
|
def test_all_real_tables_have_models(self, host_db):
|
|
import zot.table as zt
|
|
|
|
real_tables = set(_get_all_real_tables(host_db))
|
|
model_tables = set()
|
|
for name in dir(zt):
|
|
cls = getattr(zt, name)
|
|
if (
|
|
isinstance(cls, type)
|
|
and issubclass(cls, SQLTable)
|
|
and cls is not SQLTable
|
|
):
|
|
model_tables.add(cls.__tablename__)
|
|
|
|
missing = real_tables - model_tables
|
|
assert missing == set(), f"Tables without models: {missing}"
|
|
|
|
def test_items_columns_match(self, host_db):
|
|
from zot.table.core import Items
|
|
|
|
real = _get_real_table_columns(host_db, "items")
|
|
model = Items.column_names()
|
|
assert set(model) == set(real)
|
|
|
|
def test_fields_columns_match(self, host_db):
|
|
from zot.table.fields import Fields
|
|
|
|
real = _get_real_table_columns(host_db, "fields")
|
|
model = Fields.column_names()
|
|
assert set(model) == set(real)
|
|
|
|
def test_creators_columns_match(self, host_db):
|
|
from zot.table.creators import Creators
|
|
|
|
real = _get_real_table_columns(host_db, "creators")
|
|
model = Creators.column_names()
|
|
assert set(model) == set(real)
|
|
|
|
def test_item_data_columns_match(self, host_db):
|
|
from zot.table.fields import ItemData
|
|
|
|
real = _get_real_table_columns(host_db, "itemData")
|
|
model = ItemData.column_names()
|
|
assert set(model) == set(real)
|
|
|
|
def test_collections_columns_match(self, host_db):
|
|
from zot.table.collections import Collections
|
|
|
|
real = _get_real_table_columns(host_db, "collections")
|
|
model = Collections.column_names()
|
|
assert set(model) == set(real)
|
|
|
|
def test_all_models_columns_match(self, host_db):
|
|
"""Every model's columns must match the real table."""
|
|
import zot.table as zt
|
|
|
|
mismatches = []
|
|
for name in sorted(dir(zt)):
|
|
cls = getattr(zt, name)
|
|
if not (
|
|
isinstance(cls, type)
|
|
and issubclass(cls, SQLTable)
|
|
and cls is not SQLTable
|
|
):
|
|
continue
|
|
|
|
tbl = cls.__tablename__
|
|
try:
|
|
real = set(_get_real_table_columns(host_db, tbl))
|
|
except Exception:
|
|
mismatches.append(f"{name}: table {tbl} not found in DB")
|
|
continue
|
|
|
|
model = set(cls.column_names())
|
|
|
|
# Handle the renamed 'schema' -> 'schema_name' in Version
|
|
if tbl == "version" and "schema" in real and "schema_name" in model:
|
|
real = (real - {"schema"}) | {"schema_name"}
|
|
|
|
if model != real:
|
|
extra = model - real
|
|
missing = real - model
|
|
parts = []
|
|
if extra:
|
|
parts.append(f"extra={extra}")
|
|
if missing:
|
|
parts.append(f"missing={missing}")
|
|
mismatches.append(f"{name}({tbl}): {', '.join(parts)}")
|
|
|
|
assert mismatches == [], "Column mismatches:\n" + "\n".join(mismatches)
|
|
|
|
|
|
# ── Sync ID constants match real DB ─────────────────────────────────
|
|
|
|
|
|
@pytest.mark.skipif(not HOST_DB.exists(), reason="zotero.sqlite not available")
|
|
class TestSyncIDsMatchRealDB:
|
|
"""Verify bib.sync constants match the real Zotero schema."""
|
|
|
|
def test_type_map_ids_exist(self, host_db):
|
|
from bib.sync import _TYPE_MAP
|
|
|
|
con = sqlite3.connect(f"file:{host_db}?mode=ro", uri=True)
|
|
real_types = {
|
|
r[0]: r[1]
|
|
for r in con.execute("SELECT itemTypeID, typeName FROM itemTypes")
|
|
}
|
|
con.close()
|
|
|
|
for bib_type, zotero_id in _TYPE_MAP.items():
|
|
assert zotero_id in real_types, (
|
|
f"_TYPE_MAP[{bib_type!r}] = {zotero_id} not in itemTypes"
|
|
)
|
|
|
|
def test_type_map_names_correct(self, host_db):
|
|
from bib.sync import _TYPE_MAP
|
|
|
|
con = sqlite3.connect(f"file:{host_db}?mode=ro", uri=True)
|
|
real_types = dict(
|
|
con.execute("SELECT itemTypeID, typeName FROM itemTypes").fetchall()
|
|
)
|
|
con.close()
|
|
|
|
expected = {
|
|
"rule": "statute",
|
|
"regulation": "statute",
|
|
"manual": "report",
|
|
"download": "webpage",
|
|
"source": "document",
|
|
"journal-article": "journalArticle",
|
|
}
|
|
for bib_type, expected_name in expected.items():
|
|
zotero_id = _TYPE_MAP[bib_type]
|
|
assert real_types[zotero_id] == expected_name, (
|
|
f"_TYPE_MAP[{bib_type!r}] = {zotero_id} -> "
|
|
f"{real_types[zotero_id]!r}, expected {expected_name!r}"
|
|
)
|
|
|
|
def test_field_ids_exist(self, host_db):
|
|
from zot.db import FIELD_MAP as _FIELD_IDS
|
|
|
|
con = sqlite3.connect(f"file:{host_db}?mode=ro", uri=True)
|
|
real_fields = {
|
|
r[0]: r[1] for r in con.execute("SELECT fieldID, fieldName FROM fields")
|
|
}
|
|
con.close()
|
|
|
|
for field_name, field_id in _FIELD_IDS.items():
|
|
assert field_id in real_fields, (
|
|
f"_FIELD_IDS[{field_name!r}] = {field_id} not in fields table"
|
|
)
|
|
|
|
def test_field_ids_names_match(self, host_db):
|
|
from zot.db import FIELD_MAP as _FIELD_IDS
|
|
|
|
con = sqlite3.connect(f"file:{host_db}?mode=ro", uri=True)
|
|
real_fields = dict(
|
|
con.execute("SELECT fieldID, fieldName FROM fields").fetchall()
|
|
)
|
|
con.close()
|
|
|
|
for field_name, field_id in _FIELD_IDS.items():
|
|
actual_name = real_fields.get(field_id)
|
|
assert actual_name == field_name, (
|
|
f"_FIELD_IDS[{field_name!r}] = {field_id} -> {actual_name!r} in real DB"
|
|
)
|
|
|
|
def test_creator_type_artist_is_1(self, host_db):
|
|
con = sqlite3.connect(f"file:{host_db}?mode=ro", uri=True)
|
|
row = con.execute(
|
|
"SELECT creatorType FROM creatorTypes WHERE creatorTypeID = 1"
|
|
).fetchone()
|
|
con.close()
|
|
assert row[0] == "artist"
|
|
|
|
def test_creator_type_author_is_10(self, host_db):
|
|
con = sqlite3.connect(f"file:{host_db}?mode=ro", uri=True)
|
|
row = con.execute(
|
|
"SELECT creatorTypeID FROM creatorTypes WHERE creatorType = 'author'"
|
|
).fetchone()
|
|
con.close()
|
|
assert row[0] == 10
|