goyban
0732f2c34b
The bot-side half of TODO #4. Nothing federates yet; this establishes what would be eligible to, and proves the boundary before there is a hub that could leak across it. - Schema v5 adds visibility, defaulting to private. Upgrading an existing library therefore opts nothing into sharing, which a test on the legacy upgrade path pins down. - list_syncable() is the only function that will ever feed a hub, and it requires both public and approved. Tests force rows into states set_visibility refuses, so the filter is what protects the boundary rather than the caller being careful. - set_visibility refuses anything not approved: media still in the moderation queue cannot be marked shared. - /share lists memes with their current state and asks before making one public, since un-sharing stops future copies but cannot recall those already taken. The wording describes what sharing will mean once a hub exists, because by then warning is too late. - /list shows the same markers. Vault posts have no path into the sync set at all; being in a separate table is what guarantees it, and there is a test saying so. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
264 lines
10 KiB
Python
264 lines
10 KiB
Python
"""Migration tests. Run with: python test_db.py
|
|
|
|
The v1->v2 migration rebuilds the media table, so it drops the original.
|
|
These cases cover the paths that rebuild touches before it ever runs against
|
|
a real bot_db/media.db.
|
|
"""
|
|
import asyncio
|
|
import importlib
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
|
|
V1_SCHEMA = """
|
|
CREATE TABLE media (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
owner_id INTEGER NOT NULL,
|
|
file_id TEXT NOT NULL,
|
|
file_unique_id TEXT NOT NULL UNIQUE,
|
|
media_type TEXT NOT NULL,
|
|
keywords TEXT NOT NULL,
|
|
added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
);
|
|
"""
|
|
|
|
|
|
def user_version(path: str) -> int:
|
|
with sqlite3.connect(path) as c:
|
|
return c.execute("PRAGMA user_version").fetchone()[0]
|
|
|
|
|
|
def load_db(path: str):
|
|
os.environ["DB_PATH"] = path
|
|
import db
|
|
return importlib.reload(db)
|
|
|
|
|
|
async def test_legacy_upgrade(tmp: str):
|
|
path = os.path.join(tmp, "legacy.db")
|
|
with sqlite3.connect(path) as c:
|
|
c.executescript(V1_SCHEMA)
|
|
c.execute("INSERT INTO media (id,owner_id,file_id,file_unique_id,media_type,keywords)"
|
|
" VALUES (7,111,'FID_A','UNI_A','photo','cat,funny')")
|
|
c.execute("INSERT INTO media (id,owner_id,file_id,file_unique_id,media_type,keywords)"
|
|
" VALUES (9,111,'FID_B','UNI_B','gif','dog')")
|
|
|
|
db = load_db(path)
|
|
await db.init_db()
|
|
|
|
# Pinned to the module so adding a migration does not require editing this.
|
|
assert user_version(path) == db.SCHEMA_VERSION
|
|
rows = sorted((r["id"], r["keywords"]) for r in await db.search_media(""))
|
|
assert rows == [(7, "cat,funny"), (9, "dog")], f"rows not preserved: {rows}"
|
|
with sqlite3.connect(path) as c:
|
|
statuses = {r[0] for r in c.execute("SELECT status FROM media")}
|
|
tables = {r[0] for r in c.execute("SELECT name FROM sqlite_master WHERE type='table'")}
|
|
assert statuses == {"approved"}, f"upgraded rows must stay live: {statuses}"
|
|
assert "vault_items" in tables, "upgrading should create the vault table too"
|
|
with sqlite3.connect(path) as c:
|
|
visibilities = {r[0] for r in c.execute("SELECT visibility FROM media")}
|
|
assert visibilities == {"private"}, \
|
|
f"upgrading must not opt an existing library into sharing: {visibilities}"
|
|
print(" legacy v1 database upgrades with ids, keywords and visibility intact")
|
|
|
|
|
|
async def test_per_owner_uniqueness(tmp: str):
|
|
path = os.path.join(tmp, "owners.db")
|
|
db = load_db(path)
|
|
await db.init_db()
|
|
|
|
await db.add_media(111, "FID_A", "UNI_A", "photo", ["cat"])
|
|
await db.add_media(222, "FID_A2", "UNI_A", "photo", ["cat"])
|
|
assert len(await db.search_media("cat")) == 2, "two owners must each keep their own row"
|
|
assert len(await db.search_media("", owner_id=222)) == 1, "owner scoping failed"
|
|
print(" two owners can save the same file; owner scoping filters correctly")
|
|
|
|
|
|
async def test_readd_keeps_id(tmp: str):
|
|
path = os.path.join(tmp, "readd.db")
|
|
db = load_db(path)
|
|
await db.init_db()
|
|
|
|
await db.add_media(111, "FID_A", "UNI_A", "photo", ["cat"])
|
|
original = (await db.search_media("cat"))[0]["id"]
|
|
await db.add_media(111, "FID_NEW", "UNI_A", "photo", ["kitten"])
|
|
row = await db.get_media_by_id(original)
|
|
|
|
assert row is not None, "re-adding the same file orphaned its id"
|
|
assert row["keywords"] == "kitten" and row["file_id"] == "FID_NEW"
|
|
print(" re-adding a file updates in place and keeps its id")
|
|
|
|
|
|
async def test_fresh_and_idempotent(tmp: str):
|
|
path = os.path.join(tmp, "fresh.db")
|
|
db = load_db(path)
|
|
await db.init_db()
|
|
|
|
with sqlite3.connect(path) as c:
|
|
sql = c.execute("SELECT sql FROM sqlite_master WHERE name='media'").fetchone()[0]
|
|
assert "UNIQUE(owner_id, file_unique_id)" in sql
|
|
assert user_version(path) == db.SCHEMA_VERSION
|
|
|
|
await db.init_db()
|
|
await db.init_db()
|
|
print(" fresh install lands on the current version; init_db is idempotent")
|
|
|
|
|
|
async def test_pending_is_hidden(tmp: str):
|
|
path = os.path.join(tmp, "pending.db")
|
|
db = load_db(path)
|
|
await db.init_db()
|
|
|
|
await db.add_media(111, "F1", "U1", "photo", ["live"])
|
|
await db.add_media(222, "F2", "U2", "photo", ["waiting"], status="pending")
|
|
|
|
assert len(await db.search_media("")) == 1, "pending items must not be searchable"
|
|
assert await db.count_media() == 1
|
|
assert len(await db.list_media(0, 30)) == 1, "pending items must not appear in /list"
|
|
assert len(await db.search_media("", status=None)) == 2, "status=None should see everything"
|
|
assert await db.count_pending() == 1
|
|
print(" pending items stay out of search, /list and counts until approved")
|
|
|
|
|
|
async def test_queue_is_oldest_first(tmp: str):
|
|
path = os.path.join(tmp, "queue.db")
|
|
db = load_db(path)
|
|
await db.init_db()
|
|
|
|
for i in range(3):
|
|
await db.add_media(100 + i, f"F{i}", f"U{i}", "photo", [f"kw{i}"], status="pending")
|
|
|
|
first = await db.next_pending()
|
|
assert first["keywords"] == "kw0", f"queue must be oldest-first, got {first['keywords']}"
|
|
|
|
# Approving removes it from the queue and the next one steps up.
|
|
assert await db.set_status(first["id"], "approved", 999) is True
|
|
second = await db.next_pending()
|
|
assert second["keywords"] == "kw1"
|
|
assert await db.count_pending() == 2
|
|
print(" queue serves oldest first and advances as items are handled")
|
|
|
|
|
|
async def test_two_moderators_cannot_double_review(tmp: str):
|
|
path = os.path.join(tmp, "race.db")
|
|
db = load_db(path)
|
|
await db.init_db()
|
|
|
|
await db.add_media(111, "F1", "U1", "photo", ["kw"], status="pending")
|
|
item = await db.next_pending()
|
|
|
|
assert await db.set_status(item["id"], "approved", 111) is True
|
|
assert await db.set_status(item["id"], "rejected", 222) is False, \
|
|
"a second moderator must not overwrite a completed review"
|
|
|
|
row = await db.get_media_by_id(item["id"])
|
|
assert row["status"] == "approved", "first review must win"
|
|
print(" a second moderator cannot overturn an already-reviewed item")
|
|
|
|
|
|
async def test_rejected_stays_hidden(tmp: str):
|
|
path = os.path.join(tmp, "rejected.db")
|
|
db = load_db(path)
|
|
await db.init_db()
|
|
|
|
await db.add_media(111, "F1", "U1", "photo", ["nope"], status="pending")
|
|
item = await db.next_pending()
|
|
await db.set_status(item["id"], "rejected", 999)
|
|
|
|
assert await db.search_media("nope") == [], "rejected items must never surface"
|
|
assert await db.count_pending() == 0
|
|
assert await db.next_pending() is None
|
|
print(" rejected items leave the queue and never surface in search")
|
|
|
|
|
|
async def test_new_media_is_private(tmp: str):
|
|
path = os.path.join(tmp, "visibility.db")
|
|
db = load_db(path)
|
|
await db.init_db()
|
|
|
|
await db.add_media(111, "F1", "U1", "photo", ["kw"])
|
|
assert (await db.search_media("kw"))[0]["visibility"] == "private"
|
|
assert await db.list_syncable() == [], "nothing should be shared by default"
|
|
print(" media is private unless explicitly shared")
|
|
|
|
|
|
async def test_only_public_and_approved_can_sync(tmp: str):
|
|
path = os.path.join(tmp, "syncable.db")
|
|
db = load_db(path)
|
|
await db.init_db()
|
|
|
|
await db.add_media(111, "F1", "U1", "photo", ["private_approved"])
|
|
await db.add_media(111, "F2", "U2", "photo", ["public_approved"])
|
|
await db.add_media(111, "F3", "U3", "photo", ["public_pending"], status="pending")
|
|
await db.add_media(111, "F4", "U4", "photo", ["public_rejected"], status="rejected")
|
|
|
|
# Force the states set_visibility would refuse, to prove the query itself
|
|
# filters rather than relying on the writer being careful.
|
|
with sqlite3.connect(path) as c:
|
|
c.execute("UPDATE media SET visibility='public' WHERE file_unique_id IN ('U2','U3','U4')")
|
|
|
|
syncable = [r["keywords"] for r in await db.list_syncable()]
|
|
assert syncable == ["public_approved"], f"leaked: {syncable}"
|
|
assert await db.count_public() == 1
|
|
print(" only approved AND public media is ever syncable")
|
|
|
|
|
|
async def test_sharing_requires_approval(tmp: str):
|
|
path = os.path.join(tmp, "share_guard.db")
|
|
db = load_db(path)
|
|
await db.init_db()
|
|
|
|
await db.add_media(111, "F1", "U1", "photo", ["waiting"], status="pending")
|
|
pending = (await db.search_media("", status="pending"))[0]
|
|
assert await db.set_visibility(pending["id"], "public") is False, \
|
|
"unreviewed media must not be shareable"
|
|
assert await db.list_syncable() == []
|
|
print(" media awaiting moderation cannot be marked shared")
|
|
|
|
|
|
async def test_unsharing_stops_future_syncs(tmp: str):
|
|
path = os.path.join(tmp, "unshare.db")
|
|
db = load_db(path)
|
|
await db.init_db()
|
|
|
|
await db.add_media(111, "F1", "U1", "photo", ["kw"])
|
|
item = (await db.search_media("kw"))[0]
|
|
assert await db.set_visibility(item["id"], "public") is True
|
|
assert len(await db.list_syncable()) == 1
|
|
|
|
assert await db.set_visibility(item["id"], "private") is True
|
|
assert await db.list_syncable() == [], "unsharing should remove it from the sync set"
|
|
print(" unsharing takes an item back out of the sync set")
|
|
|
|
|
|
async def test_vault_never_syncs(tmp: str):
|
|
path = os.path.join(tmp, "vault_sync.db")
|
|
db = load_db(path)
|
|
await db.init_db()
|
|
|
|
await db.add_vault_item(111, 500, [1, 2], "a private post", ["personal"])
|
|
# The vault lives in its own table, so the sync query cannot reach it at all.
|
|
assert await db.list_syncable() == []
|
|
assert await db.count_public() == 0
|
|
print(" vault posts have no path into the sync set")
|
|
|
|
|
|
async def main():
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
for test in (test_legacy_upgrade, test_per_owner_uniqueness,
|
|
test_readd_keeps_id, test_fresh_and_idempotent,
|
|
test_pending_is_hidden, test_queue_is_oldest_first,
|
|
test_two_moderators_cannot_double_review,
|
|
test_rejected_stays_hidden, test_new_media_is_private,
|
|
test_only_public_and_approved_can_sync,
|
|
test_sharing_requires_approval,
|
|
test_unsharing_stops_future_syncs, test_vault_never_syncs):
|
|
await test(tmp)
|
|
print("\nAll database tests passed.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|