memes_bot/test_db.py

"""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) == 3
    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")}
    assert statuses == {"approved"}, f"upgraded rows must stay live: {statuses}"
    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) == 3

    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 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):
            await test(tmp)
    print("\nAll database tests passed.")


if __name__ == "__main__":
    asyncio.run(main())