69 lines
2.9 KiB
Python
69 lines
2.9 KiB
Python
"""Characterize DB layer (donor: get_db, _migrate_schema, status transitions,
|
|
get_pending, purge_excluded, prune_missing, FTS sync)."""
|
|
import photo_analyzer as pa
|
|
from conftest import seed_analyzed, write_fixture
|
|
|
|
|
|
def test_schema_and_migration_idempotent(tmp_path):
|
|
db_path = str(tmp_path / "x.db")
|
|
conn = pa.get_db(db_path)
|
|
conn.close()
|
|
conn = pa.get_db(db_path) # re-open runs migration again — must not fail
|
|
cols = {r["name"] for r in conn.execute("PRAGMA table_info(photos)")}
|
|
assert {"path", "status", "phash", "file_sha1", "dup_of", "description",
|
|
"tags", "raw_response", "analyzed_at", "exif_written_at"} <= cols
|
|
conn.close()
|
|
|
|
|
|
def test_status_lifecycle_pending_analyzed_exif_written(db):
|
|
seed_analyzed(db, "/x/a.jpg")
|
|
assert db.execute("SELECT status FROM photos").fetchone()["status"] == "analyzed"
|
|
pa.mark_exif_written(db, "/x/a.jpg")
|
|
assert db.execute("SELECT status FROM photos").fetchone()["status"] == "exif_written"
|
|
|
|
|
|
def test_mark_error_and_retry_via_get_pending(db):
|
|
pa.upsert_pending(db, "/x/a.jpg")
|
|
pa.mark_error(db, "/x/a.jpg", "boom")
|
|
row = db.execute("SELECT status, error_message FROM photos").fetchone()
|
|
assert (row["status"], row["error_message"]) == ("error", "boom")
|
|
# error rows are retried on the next run
|
|
assert pa.get_pending(db, reanalyze=False) == ["/x/a.jpg"]
|
|
|
|
|
|
def test_mark_analyzed_clears_error(db):
|
|
pa.upsert_pending(db, "/x/a.jpg")
|
|
pa.mark_error(db, "/x/a.jpg", "boom")
|
|
seed_analyzed(db, "/x/a.jpg")
|
|
assert db.execute("SELECT error_message FROM photos").fetchone()["error_message"] is None
|
|
|
|
|
|
def test_upsert_pending_never_downgrades(db):
|
|
seed_analyzed(db, "/x/a.jpg")
|
|
pa.upsert_pending(db, "/x/a.jpg") # INSERT OR IGNORE — no reset to pending
|
|
assert db.execute("SELECT status FROM photos").fetchone()["status"] == "analyzed"
|
|
|
|
|
|
def test_fts_kept_in_sync_by_triggers(db):
|
|
seed_analyzed(db, "/x/beach.jpg", description="A sunny beach with palm trees.")
|
|
hit = db.execute(
|
|
"SELECT path FROM photos_fts WHERE photos_fts MATCH 'beach'").fetchall()
|
|
assert [r["path"] for r in hit] == ["/x/beach.jpg"]
|
|
|
|
|
|
def test_purge_excluded_removes_ignore_and_thumb_rows(db):
|
|
for p in ("/lib/_IGNORE/a.jpg", "/lib/x/.@__thumb/b.jpg", "/lib/keep.jpg"):
|
|
pa.upsert_pending(db, p)
|
|
assert pa.purge_excluded(db) == 2
|
|
assert [r["path"] for r in db.execute("SELECT path FROM photos")] == ["/lib/keep.jpg"]
|
|
|
|
|
|
def test_prune_missing_guards_unmounted_library(db, tmp_path):
|
|
pa.upsert_pending(db, "/gone/a.jpg")
|
|
# library root missing → refuses to prune (protects against unmounted volume)
|
|
assert pa.prune_missing(db, tmp_path / "not-there") == 0
|
|
assert db.execute("SELECT COUNT(*) FROM photos").fetchone()[0] == 1
|
|
# existing root → stale row is pruned
|
|
assert pa.prune_missing(db, tmp_path) == 1
|
|
assert db.execute("SELECT COUNT(*) FROM photos").fetchone()[0] == 0
|