Jonathan
27 lines
784 B
Python
27 lines
784 B
Python
from lyra_worker.claim import claim_next
|
|
from tests.conftest import insert_request
|
|
|
|
|
|
def test_claim_returns_none_when_empty(conn):
|
|
assert claim_next(conn) is None
|
|
|
|
|
|
def test_claim_advances_job_to_matching(conn):
|
|
job_id = insert_request(conn)
|
|
claimed = claim_next(conn)
|
|
assert claimed == job_id
|
|
|
|
with conn.cursor() as cur:
|
|
cur.execute('SELECT state, "currentStage", attempts, "claimedAt" FROM "Job" WHERE id = %s', (job_id,))
|
|
state, stage, attempts, claimed_at = cur.fetchone()
|
|
assert state == "matching"
|
|
assert stage == "match"
|
|
assert attempts == 1
|
|
assert claimed_at is not None
|
|
|
|
|
|
def test_claim_does_not_repick_claimed_job(conn):
|
|
insert_request(conn)
|
|
assert claim_next(conn) is not None
|
|
assert claim_next(conn) is None
|