bvandeusen
31217498ce
Family-consistent CI for StashHandler (mirrors the NhenArchiver Python sibling, minus the integration lane — StashHandler is SQLite-only with no service containers). - .forgejo/workflows/ci.yml: lint (ruff) -> unit (pytest) -> publish. publish builds + pushes git.fabledsword.com/bvandeusen/stashhandler with the family rule-46 tag scheme (dev->:dev+:c-<sha>, main->:latest+:c-<sha>, v* tag->:<version>+:latest). Gated on lint+unit. Needs REGISTRY_TOKEN. - tests/: starter unit suite — hashing, SQLite dedup, media filter, transfer modes. - pyproject.toml: ruff target + pytest pythonpath/testpaths. - ci-requirements.md (family rule 39), .dockerignore. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016nyYC9dTQ78SqhpNFtpMvT
48 lines
1.7 KiB
Python
48 lines
1.7 KiB
Python
import db
|
|
|
|
|
|
def make_conn(tmp_path):
|
|
return db.init_db(str(tmp_path / "seen.db"))
|
|
|
|
|
|
def test_record_and_is_known(tmp_path):
|
|
conn = make_conn(tmp_path)
|
|
assert not db.is_known(conn, "deadbeef")
|
|
db.record_file(conn, "deadbeef", "a.mp4", "scan")
|
|
assert db.is_known(conn, "deadbeef")
|
|
|
|
|
|
def test_record_file_is_idempotent_on_hash(tmp_path):
|
|
conn = make_conn(tmp_path)
|
|
db.record_file(conn, "h1", "first.mp4", "scan")
|
|
db.record_file(conn, "h1", "second.mp4", "scan") # INSERT OR IGNORE
|
|
row = conn.execute("SELECT filename FROM seen_files WHERE hash='h1'").fetchone()
|
|
assert row[0] == "first.mp4" # first write wins, no duplicate row
|
|
|
|
|
|
def test_increment_seen(tmp_path):
|
|
conn = make_conn(tmp_path)
|
|
db.record_file(conn, "h1", "a.mp4", "scan") # times_seen defaults to 1
|
|
db.increment_seen(conn, "h1")
|
|
db.increment_seen(conn, "h1")
|
|
row = conn.execute("SELECT times_seen FROM seen_files WHERE hash='h1'").fetchone()
|
|
assert row[0] == 3
|
|
|
|
|
|
def test_processed_torrent_roundtrip(tmp_path):
|
|
conn = make_conn(tmp_path)
|
|
assert not db.is_torrent_processed(conn, "ih1")
|
|
db.record_torrent(conn, "ih1", "Some.Torrent.Name")
|
|
assert db.is_torrent_processed(conn, "ih1")
|
|
|
|
|
|
def test_processed_file_is_scoped_to_torrent_and_name(tmp_path):
|
|
conn = make_conn(tmp_path)
|
|
assert not db.is_file_processed(conn, "ih1", "f.mp4")
|
|
db.record_processed_file(conn, "ih1", "f.mp4")
|
|
assert db.is_file_processed(conn, "ih1", "f.mp4")
|
|
# A different file under the same torrent is tracked independently.
|
|
assert not db.is_file_processed(conn, "ih1", "other.mp4")
|
|
# Same file name under a different torrent is also independent.
|
|
assert not db.is_file_processed(conn, "ih2", "f.mp4")
|