goyban
b55cee1650
Step 0 of the roadmap: everything from the moderation queue onward adds columns, and init_db could only ever run CREATE TABLE IF NOT EXISTS, so it could not alter a live deployment. - Track schema state in PRAGMA user_version and run ordered migrations. Fresh installs create the v1 table and migrate forward, so the upgrade path runs on every new deployment rather than only on old ones. - Rebuild media with UNIQUE(owner_id, file_unique_id). The old global unique constraint stopped two users from each saving the same file, which blocks public mode. - Replace INSERT OR REPLACE with an upsert. Replacing deleted and reinserted the row, giving it a new id and silently invalidating any id the user had noted from /list. - Give search_media an optional owner_id scope, defaulting to the current shared-library behaviour. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
109 lines
3.6 KiB
Python
109 lines
3.6 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()
|
|
|
|
assert user_version(path) == 2
|
|
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}"
|
|
print(" legacy v1 database upgrades to v2 with ids and keywords 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) == 2
|
|
|
|
await db.init_db()
|
|
await db.init_db()
|
|
print(" fresh install lands on v2; init_db is idempotent")
|
|
|
|
|
|
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):
|
|
await test(tmp)
|
|
print("\nAll database tests passed.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|