Jonathan
42 lines
1.4 KiB
Python
42 lines
1.4 KiB
Python
import os
|
|
|
|
import psycopg
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture()
|
|
def conn():
|
|
dsn = os.environ["DATABASE_URL"]
|
|
connection = psycopg.connect(dsn)
|
|
# Clean up any pre-existing rows before the test runs (Job cascades from Request).
|
|
with connection.cursor() as cur:
|
|
cur.execute('DELETE FROM "Job"')
|
|
cur.execute('DELETE FROM "Request"')
|
|
connection.commit()
|
|
yield connection
|
|
# Clean up rows created during the test (Job cascades from Request).
|
|
with connection.cursor() as cur:
|
|
cur.execute('DELETE FROM "Job"')
|
|
cur.execute('DELETE FROM "Request"')
|
|
connection.commit()
|
|
connection.close()
|
|
|
|
|
|
def insert_request(conn, artist="Artist", album="Album"):
|
|
"""Insert a Request + its Job (state 'requested') and return the job id."""
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
'INSERT INTO "Request" (id, artist, album, status, "createdAt") '
|
|
"VALUES (gen_random_uuid()::text, %s, %s, 'pending', now()) RETURNING id",
|
|
(artist, album),
|
|
)
|
|
request_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",
|
|
(request_id,),
|
|
)
|
|
job_id = cur.fetchone()[0]
|
|
conn.commit()
|
|
return job_id
|