cc418cefde
zot Db.__init__ leak on schema-parity raise; test fixtures not closing raw Db/create_db connections; cli comments _build_bib_helpers cache connection never closed. Gate: error::ResourceWarning (coverage's own sqlitedb warning ignored). Root causes tracemalloc-verified; most per-test attributions in #618 were GC-timing misattribution across xdist workers.
140 lines
5.0 KiB
Python
140 lines
5.0 KiB
Python
"""Tests for zot.schema — fresh Zotero DB creation from real schema."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from zot.db import CREATOR_TYPES, FIELD_MAP, TYPE_MAP, Db
|
|
from zot.schema import create_db
|
|
|
|
|
|
class TestCreateDb:
|
|
def test_creates_in_memory(self):
|
|
con = create_db()
|
|
tables = [
|
|
r[0]
|
|
for r in con.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table'"
|
|
).fetchall()
|
|
]
|
|
assert "items" in tables
|
|
assert "itemData" in tables
|
|
assert "fields" in tables
|
|
con.close()
|
|
|
|
def test_has_all_61_tables(self):
|
|
con = create_db()
|
|
tables = [
|
|
r[0]
|
|
for r in con.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table'"
|
|
).fetchall()
|
|
]
|
|
assert len(tables) >= 61
|
|
con.close()
|
|
|
|
def test_item_types_seeded(self):
|
|
con = create_db()
|
|
count = con.execute("SELECT count(*) FROM itemTypes").fetchone()[0]
|
|
assert count == 40
|
|
con.close()
|
|
|
|
def test_fields_seeded(self):
|
|
con = create_db()
|
|
count = con.execute("SELECT count(*) FROM fields").fetchone()[0]
|
|
assert count == 123
|
|
con.close()
|
|
|
|
def test_creator_types_seeded(self):
|
|
con = create_db()
|
|
count = con.execute("SELECT count(*) FROM creatorTypes").fetchone()[0]
|
|
assert count == 37
|
|
con.close()
|
|
|
|
def test_item_type_fields_seeded(self):
|
|
con = create_db()
|
|
count = con.execute("SELECT count(*) FROM itemTypeFields").fetchone()[0]
|
|
assert count == 770
|
|
con.close()
|
|
|
|
def test_base_field_mappings_seeded(self):
|
|
con = create_db()
|
|
count = con.execute("SELECT count(*) FROM baseFieldMappings").fetchone()[0]
|
|
assert count == 72
|
|
con.close()
|
|
|
|
def test_library_exists(self):
|
|
con = create_db()
|
|
row = con.execute("SELECT type FROM libraries WHERE libraryID = 1").fetchone()
|
|
assert row[0] == "user"
|
|
con.close()
|
|
|
|
def test_type_ids_match_constants(self):
|
|
con = create_db()
|
|
for name, expected_id in TYPE_MAP.items():
|
|
row = con.execute(
|
|
"SELECT itemTypeID FROM itemTypes WHERE typeName = ?", (name,)
|
|
).fetchone()
|
|
assert row is not None, f"type {name} not found"
|
|
assert row[0] == expected_id, f"type {name}: {row[0]} != {expected_id}"
|
|
con.close()
|
|
|
|
def test_field_ids_match_constants(self):
|
|
con = create_db()
|
|
for name, expected_id in FIELD_MAP.items():
|
|
row = con.execute(
|
|
"SELECT fieldID FROM fields WHERE fieldName = ?", (name,)
|
|
).fetchone()
|
|
assert row is not None, f"field {name} not found"
|
|
assert row[0] == expected_id, f"field {name}: {row[0]} != {expected_id}"
|
|
con.close()
|
|
|
|
def test_creator_type_ids_match_constants(self):
|
|
con = create_db()
|
|
for name, expected_id in CREATOR_TYPES.items():
|
|
row = con.execute(
|
|
"SELECT creatorTypeID FROM creatorTypes WHERE creatorType = ?",
|
|
(name,),
|
|
).fetchone()
|
|
assert row is not None, f"creator type {name} not found"
|
|
assert row[0] == expected_id
|
|
con.close()
|
|
|
|
def test_works_with_orm(self, tmp_path):
|
|
"""Full round-trip: create_db → Db → create item → read back."""
|
|
p = str(tmp_path / "schema.sqlite")
|
|
con = create_db(p)
|
|
con.close()
|
|
with Db(p) as db:
|
|
item_id = db.create_item(TYPE_MAP["statute"])
|
|
db.set_fields(item_id, {"nameOfAct": "Test Act", "code": "FR"})
|
|
db.sync_tags(item_id, ["test:schema"])
|
|
db.add_creators(item_id, [("Jane", "Doe")])
|
|
db.commit()
|
|
|
|
item = db.get_item(item_id)
|
|
assert item["itemType"] == "statute"
|
|
assert item["fields"]["nameOfAct"] == "Test Act"
|
|
assert "test:schema" in item["tags"]
|
|
assert item["creators"][0]["lastName"] == "Doe"
|
|
|
|
def test_validation_works_with_real_schema(self, tmp_path):
|
|
"""With full schema seed data, validation can detect invalid fields."""
|
|
p = str(tmp_path / "valid.sqlite")
|
|
con = create_db(p)
|
|
con.close()
|
|
with Db(p) as db:
|
|
# statute type — set a field that's NOT valid for statute
|
|
item_id = db.create_item(TYPE_MAP["statute"])
|
|
# publicationTitle (41) is not valid for statute (36)
|
|
pub_field_id = FIELD_MAP["publicationTitle"]
|
|
db.con.execute("INSERT INTO itemDataValues (value) VALUES ('Bad Journal')")
|
|
vid = db.con.execute(
|
|
"SELECT valueID FROM itemDataValues WHERE value = 'Bad Journal'"
|
|
).fetchone()[0]
|
|
db.con.execute(
|
|
"INSERT INTO itemData (itemID, fieldID, valueID) VALUES (?, ?, ?)",
|
|
(item_id, pub_field_id, vid),
|
|
)
|
|
issues = db.validate_item(item_id)
|
|
assert len(issues) == 1
|
|
assert "publicationTitle" in issues[0]
|