cronekorkn/left4me
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

tests/cli: cover running+cancelling idempotency, tighten app-context scope

Browse source
This commit is contained in:
mwiegand 2026-05-11 23:18:54 +02:00
parent 0ab54b4a7d
commit f614ac05f0
No known key found for this signature in database
1 changed files with 31 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

45
l4d2web/tests/test_cli.py
View file

@ -24,14 +24,14 @@ def test_workshop_refresh_enqueues_job(app_env):
runner = CliRunner() runner = CliRunner()
with app_env.app_context(): with app_env.app_context():
result = runner.invoke(workshop_refresh, []) result = runner.invoke(workshop_refresh, [])
assert result.exit_code == 0, result.output assert result.exit_code == 0, result.output
assert "enqueued refresh_workshop_items job" in result.output assert "enqueued refresh_workshop_items job" in result.output
with session_scope() as db: with session_scope() as db:
jobs = db.scalars(select(Job).where(Job.operation == "refresh_workshop_items")).all() jobs = db.scalars(select(Job).where(Job.operation == "refresh_workshop_items")).all()
assert len(jobs) == 1 assert len(jobs) == 1
assert jobs[0].state == "queued" assert jobs[0].state == "queued"
assert jobs[0].user_id is None assert jobs[0].user_id is None
assert jobs[0].server_id is None assert jobs[0].server_id is None
def test_workshop_refresh_is_idempotent_when_job_queued(app_env): def test_workshop_refresh_is_idempotent_when_job_queued(app_env):
@ -39,9 +39,26 @@ def test_workshop_refresh_is_idempotent_when_job_queued(app_env):
with app_env.app_context(): with app_env.app_context():
first = runner.invoke(workshop_refresh, []) first = runner.invoke(workshop_refresh, [])
second = runner.invoke(workshop_refresh, []) second = runner.invoke(workshop_refresh, [])
assert first.exit_code == 0 assert first.exit_code == 0
assert second.exit_code == 0 assert second.exit_code == 0
assert "already queued" in second.output assert "already queued" in second.output
with session_scope() as db: with session_scope() as db:
jobs = db.scalars(select(Job).where(Job.operation == "refresh_workshop_items")).all() jobs = db.scalars(select(Job).where(Job.operation == "refresh_workshop_items")).all()
assert len(jobs) == 1, "must not insert a second job when one is already queued" assert len(jobs) == 1, "must not insert a second job when one is already queued"
@pytest.mark.parametrize("active_state", ["running", "cancelling"])
def test_workshop_refresh_is_idempotent_when_job_active(app_env, active_state):
runner = CliRunner()
with app_env.app_context():
with session_scope() as db:
db.add(Job(
user_id=None, server_id=None,
operation="refresh_workshop_items", state=active_state,
))
result = runner.invoke(workshop_refresh, [])
assert result.exit_code == 0
assert f"already {active_state}" in result.output
with session_scope() as db:
jobs = db.scalars(select(Job).where(Job.operation == "refresh_workshop_items")).all()
assert len(jobs) == 1