"""Migration 0102's data step, DB-lane: every existing job lands in its source folder, and a done file's expiry moves from the shared asset to the user's job. The schema half is checked by test_migrations.py; a row placed in the wrong folder is invisible there. """ from alembic import command from alembic.config import Config from sqlalchemy import text from app.db import engine def test_existing_jobs_are_placed_and_keep_their_expiry(db, make_user): alice, bob = make_user().id, make_user().id cfg = Config("alembic.ini") command.downgrade(cfg, "0101_plex_text_plain") try: with engine.begin() as conn: def asset(kind, ref, expires=None): return conn.execute( text( "INSERT INTO media_assets (source_kind, source_ref, format_sig, status," " ref_count, expires_at) VALUES (:k, :r, :sig, 'ready', 1, :e)" " RETURNING id" ), {"k": kind, "r": ref, "sig": kind + ref, "e": expires}, ).scalar_one() def job(user, kind, ref, asset_id, status="done", spec=None, parent=None): return conn.execute( text( "INSERT INTO download_jobs (user_id, asset_id, source_kind, source_ref," " profile_snapshot, status, edit_spec, source_job_id)" " VALUES (:u, :a, :k, :r, '{}', :s, CAST(:spec AS json), :p) RETURNING id" ), {"u": user, "a": asset_id, "k": kind, "r": ref, "s": status, "spec": spec, "p": parent}, ).scalar_one() # fmt: skip yt = asset("youtube", "abcdefghijk", "2030-01-02T00:00:00+00:00") tt = asset("url", "https://www.tiktok.com/@a/video/1", "2030-01-03T00:00:00+00:00") up = asset("upload", "u1") cl = asset("clip", "c1", "2030-01-04T00:00:00+00:00") j_yt = job(alice, "youtube", "abcdefghijk", yt) j_yt_bob = job(bob, "youtube", "abcdefghijk", yt, status="queued") j_tt = job(alice, "url", "https://www.tiktok.com/@a/video/1", tt) j_up = job(alice, "upload", "u1", up) j_cl = job(alice, "clip", "c1", cl, spec='{"source": "plex", "ref": "9"}') j_ed = job(alice, "edit", "e1", cl, parent=j_tt) finally: command.upgrade(cfg, "head") with engine.connect() as conn: rows = conn.execute( text( "SELECT j.id, f.user_id, f.source_key, f.retention_days, j.expires_at::date" " FROM download_jobs j JOIN file_folders f ON f.id = j.folder_id" ) ).all() folders = conn.execute(text("SELECT count(*) FROM file_folders")).scalar() got = {r[0]: (r[1], r[2], r[3], str(r[4]) if r[4] else None) for r in rows} assert got == { j_yt: (alice, "youtube", 30, "2030-01-02"), j_yt_bob: (bob, "youtube", 30, None), # not done yet: no expiry until it is j_tt: (alice, "site:tiktok.com", 30, "2030-01-03"), j_up: (alice, "upload", None, None), # an upload never expires j_cl: (alice, "plex", 30, "2030-01-04"), j_ed: (alice, "site:tiktok.com", 30, "2030-01-04"), } # alice: youtube, tiktok, upload, plex; bob: youtube. assert folders == 5 def test_the_admins_retention_is_kept_and_the_downgrade_gives_the_asset_back_its_expiry( db, make_user ): alice, bob = make_user().id, make_user().id cfg = Config("alembic.ini") command.downgrade(cfg, "0101_plex_text_plain") try: with engine.begin() as conn: conn.execute( text( "INSERT INTO system_config (key, value) VALUES ('download_retention_days', '7')" " ON CONFLICT (key) DO UPDATE SET value = '7'" ) ) yt = conn.execute( text( "INSERT INTO media_assets (source_kind, source_ref, format_sig, status," " ref_count, expires_at) VALUES ('youtube', 'abcdefghijk', 'sig', 'ready', 2," " '2030-01-02T00:00:00+00:00') RETURNING id" ) ).scalar_one() for user in (alice, bob): conn.execute( text( "INSERT INTO download_jobs (user_id, asset_id, source_kind, source_ref," " profile_snapshot, status) VALUES (:u, :a, 'youtube', 'abcdefghijk'," " '{}', 'done')" ), {"u": user, "a": yt}, ) command.upgrade(cfg, "head") with engine.connect() as conn: retention = conn.execute( text( "SELECT DISTINCT retention_days FROM file_folders WHERE source_key = 'youtube'" ) ).all() def asset_expiry_after_downgrade(alice_at, bob_at): with engine.begin() as conn: for user, at in ((alice, alice_at), (bob, bob_at)): conn.execute( text( "UPDATE download_jobs SET expires_at = CAST(:at AS timestamptz)" " WHERE asset_id = :a AND user_id = :u" ), {"at": at, "a": yt, "u": user}, ) command.downgrade(cfg, "0101_plex_text_plain") with engine.connect() as conn: got = conn.execute( text("SELECT expires_at::date FROM media_assets WHERE id = :a"), {"a": yt} ).scalar() command.upgrade(cfg, "head") return str(got) if got else None # The shared file goes when its last holder's copy would; one holder keeping it keeps it. latest = asset_expiry_after_downgrade( "2030-02-01T00:00:00+00:00", "2030-03-01T00:00:00+00:00" ) kept_by_one = asset_expiry_after_downgrade("2030-02-01T00:00:00+00:00", None) finally: command.upgrade(cfg, "head") assert retention == [(7,)] assert latest == "2030-03-01" assert kept_by_one is None