Jonathan
45 lines
1.5 KiB
Python
45 lines
1.5 KiB
Python
from lyra_worker.main import _run_scan
|
|
from lyra_worker.types import MBTarget
|
|
|
|
|
|
class FakeProbe:
|
|
def probe(self, path):
|
|
from lyra_worker.probe import ProbeResult
|
|
return ProbeResult(artist="", album="", fmt="FLAC", quality_class=3)
|
|
|
|
|
|
class FakeResolver:
|
|
def resolve(self, artist, album):
|
|
return MBTarget(artist=artist, album=album, rg_mbid="rg1", artist_mbid="a1")
|
|
|
|
|
|
def _set(conn, key, value):
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
'INSERT INTO "Config"(key, value, secret, "updatedAt") VALUES (%s, %s, false, now()) '
|
|
'ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, "updatedAt" = now()',
|
|
(key, value),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def _get(conn, key):
|
|
with conn.cursor() as cur:
|
|
cur.execute('SELECT value FROM "Config" WHERE key = %s', (key,))
|
|
row = cur.fetchone()
|
|
return row[0] if row else None
|
|
|
|
|
|
def test_run_scan_writes_result_and_clears_flag(conn, tmp_path):
|
|
(tmp_path / "John Mayer" / "Continuum (2006)").mkdir(parents=True)
|
|
(tmp_path / "John Mayer" / "Continuum (2006)" / "01.flac").write_bytes(b"")
|
|
_set(conn, "scan.requested", "true")
|
|
|
|
_run_scan(conn, FakeResolver(), FakeProbe(), dest_root=str(tmp_path))
|
|
|
|
assert _get(conn, "scan.requested") == "false" # flag cleared
|
|
assert "imported" in (_get(conn, "scan.result") or "") # summary written
|
|
with conn.cursor() as cur:
|
|
cur.execute('SELECT count(*) FROM "LibraryItem"')
|
|
assert cur.fetchone()[0] == 1
|