Lyra/worker/tests/test_upgrade_e2e.py

from lyra_worker.adapters.qobuz import QobuzAdapter
from lyra_worker.claim import claim_next
from lyra_worker.monitor import MonitorConfig, enqueue_due, reconcile
from lyra_worker.pipeline import run_pipeline
from tests.conftest import insert_monitored_release
from tests.test_qobuz_adapter import FakeQobuzClient

CFG = MonitorConfig(enabled=True, quality_cutoff=2, retry_interval_hours=6, upgrade_window_days=14)


def _seed_library_item(conn, artist, album, quality_class, path):
    """Seed a throwaway Request (for its requestId) + a LibraryItem at the given quality,
    mirroring how test_monitor_enqueue.py's already-in-library test seeds a LibraryItem."""
    with conn.cursor() as cur:
        cur.execute(
            'INSERT INTO "Request" (id, artist, album, status, "createdAt") '
            "VALUES (gen_random_uuid()::text, %s, %s, 'completed', now()) RETURNING id",
            (artist, album),
        )
        req_id = cur.fetchone()[0]
        cur.execute(
            'INSERT INTO "LibraryItem" (id, "requestId", artist, album, path, source, format, '
            '"qualityClass", "importedAt") '
            "VALUES (gen_random_uuid()::text, %s, %s, %s, %s, 'qobuz', 'FLAC', %s, now())",
            (req_id, artist, album, path, quality_class),
        )
    conn.commit()


def _library_row(conn, artist, album):
    with conn.cursor() as cur:
        cur.execute(
            'SELECT path, "qualityClass", source FROM "LibraryItem" WHERE artist = %s AND album = %s',
            (artist, album),
        )
        return cur.fetchone()


def _job_state(conn, job_id):
    with conn.cursor() as cur:
        cur.execute('SELECT state FROM "Job" WHERE id = %s', (job_id,))
        return cur.fetchone()[0]


def _monitored_release_row(conn, rel_id):
    with conn.cursor() as cur:
        cur.execute(
            'SELECT state, "currentQualityClass" FROM "MonitoredRelease" WHERE id = %s', (rel_id,)
        )
        return cur.fetchone()


def _enqueue_monitor_job_directly(conn, rel_id, artist, album):
    """Insert a Request+Job tagged with monitoredReleaseId, the same shape enqueue_due's
    _enqueue_one produces. Used when enqueue_due itself would refuse to enqueue (e.g. the
    existing copy already meets the cutoff, in which case enqueue_due marks it fulfilled and
    skips it -- see test_monitor_enqueue.test_already_in_library_at_cutoff_is_fulfilled)."""
    with conn.cursor() as cur:
        cur.execute(
            'INSERT INTO "Request" (id, artist, album, status, "monitoredReleaseId", "createdAt") '
            "VALUES (gen_random_uuid()::text, %s, %s, 'pending', %s, now()) RETURNING id",
            (artist, album, rel_id),
        )
        req_id = cur.fetchone()[0]
        cur.execute(
            'INSERT INTO "Job" (id, "requestId", state, "currentStage", attempts, "createdAt", "updatedAt") '
            "VALUES (gen_random_uuid()::text, %s, 'requested', 'intake', 0, now(), now()) RETURNING id",
            (req_id,),
        )
        job_id = cur.fetchone()[0]
    conn.commit()
    return job_id


def test_below_cutoff_copy_is_upgraded_end_to_end(conn):
    rel_id = insert_monitored_release(
        conn, artist_name="John Mayer", album="Continuum", rg_mbid="rg-upgrade-below",
        monitored=True, state="grabbed", current_quality_class=1,
        first_grabbed_sql="now() - interval '1 day'",
    )
    _seed_library_item(conn, "John Mayer", "Continuum", quality_class=1, path="/m/old-lossy")

    # the grabbed-below-cutoff release is due and gets enqueued as a monitor-driven job.
    assert enqueue_due(conn, CFG) == 1

    job_id = claim_next(conn)
    assert job_id is not None

    # upgrade_cutoff=2 so intake does NOT short-circuit on the existing quality-1 copy.
    run_pipeline(conn, job_id, [QobuzAdapter(FakeQobuzClient())], upgrade_cutoff=2)

    assert _job_state(conn, job_id) == "imported"
    path, quality, source = _library_row(conn, "John Mayer", "Continuum")
    # Q: FakeQobuzClient's candidate is 24-bit/96kHz FLAC -> hi-res, qualityClass 3.
    assert quality == 3
    assert quality > 1
    assert source == "qobuz"
    assert path != "/m/old-lossy"  # the upsert replaced the below-cutoff copy

    reconcile(conn, job_id, CFG)
    state, cqc = _monitored_release_row(conn, rel_id)
    assert state == "fulfilled"  # Q(=3) >= cutoff(=2)
    assert cqc == 3


def test_already_at_cutoff_copy_is_not_replaced(conn):
    """Guard test pinning the fix: when the existing copy already meets the cutoff, a
    monitor-driven job must still short-circuit at intake (no needless re-download)."""
    rel_id = insert_monitored_release(
        conn, artist_name="John Mayer", album="Continuum", rg_mbid="rg-upgrade-at-cutoff",
        monitored=True, state="grabbed", current_quality_class=3,
        first_grabbed_sql="now() - interval '1 day'",
    )
    _seed_library_item(conn, "John Mayer", "Continuum", quality_class=3, path="/m/already-hires")

    # enqueue_due would mark this release fulfilled and refuse to enqueue it (already at
    # cutoff), so build the monitor-driven job directly to exercise run_pipeline's intake path.
    job_id = _enqueue_monitor_job_directly(conn, rel_id, "John Mayer", "Continuum")
    claimed = claim_next(conn)
    assert claimed == job_id

    run_pipeline(conn, job_id, [QobuzAdapter(FakeQobuzClient())], upgrade_cutoff=2)

    assert _job_state(conn, job_id) == "imported"
    # unchanged: still the original below-Q, already-at-cutoff copy, untouched.
    assert _library_row(conn, "John Mayer", "Continuum") == ("/m/already-hires", 3, "qobuz")

    with conn.cursor() as cur:
        cur.execute('SELECT count(*) FROM "Candidate" WHERE "jobId" = %s', (job_id,))
        assert cur.fetchone()[0] == 0  # intake short-circuited: match stage never ran, no candidates persisted