Jonathan
51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
from lyra_worker.floor import (
|
|
press_paused,
|
|
job_delay_seconds,
|
|
kill_requested,
|
|
consume_kill_switch,
|
|
)
|
|
|
|
|
|
def test_press_paused_reads_truthy_values():
|
|
assert press_paused({"floor.paused": "true"}) is True
|
|
assert press_paused({"floor.paused": "ON"}) is True
|
|
assert press_paused({"floor.paused": "false"}) is False
|
|
assert press_paused({}) is False
|
|
|
|
|
|
def test_job_delay_seconds_parses_and_defaults():
|
|
assert job_delay_seconds({"floor.jobDelaySeconds": "30"}) == 30.0
|
|
assert job_delay_seconds({}) == 0.0
|
|
assert job_delay_seconds({"floor.jobDelaySeconds": "-5"}) == 0.0
|
|
assert job_delay_seconds({"floor.jobDelaySeconds": "junk"}) == 0.0
|
|
|
|
|
|
def test_kill_requested_reads_flag():
|
|
assert kill_requested({"floor.killRequested": "true"}) is True
|
|
assert kill_requested({}) is False
|
|
|
|
|
|
def test_consume_kill_switch_clears_flag_and_exits(conn):
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
'INSERT INTO "Config"(key, value, secret, "updatedAt") '
|
|
"VALUES ('floor.killRequested', 'true', false, now())"
|
|
)
|
|
conn.commit()
|
|
|
|
calls = []
|
|
fired = consume_kill_switch(conn, exit_fn=lambda code: calls.append(code))
|
|
|
|
assert fired is True
|
|
assert calls == [0]
|
|
with conn.cursor() as cur:
|
|
cur.execute("SELECT value FROM \"Config\" WHERE key = 'floor.killRequested'")
|
|
assert cur.fetchone()[0] == "false"
|
|
|
|
|
|
def test_consume_kill_switch_noop_when_unset(conn):
|
|
calls = []
|
|
fired = consume_kill_switch(conn, exit_fn=lambda code: calls.append(code))
|
|
assert fired is False
|
|
assert calls == []
|