47 lines
1.7 KiB
Python
47 lines
1.7 KiB
Python
"""Tests for the l4d2web Flask CLI subcommands."""
|
|
from __future__ import annotations
|
|
from click.testing import CliRunner
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from l4d2web.app import create_app
|
|
from l4d2web.cli import workshop_refresh
|
|
from l4d2web.db import init_db, session_scope
|
|
from l4d2web.models import Job
|
|
|
|
|
|
@pytest.fixture
|
|
def app_env(tmp_path, monkeypatch):
|
|
db_url = f"sqlite:///{tmp_path/'cli.db'}"
|
|
monkeypatch.setenv("DATABASE_URL", db_url)
|
|
monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path))
|
|
app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
|
|
init_db()
|
|
return app
|
|
|
|
|
|
def test_workshop_refresh_enqueues_job(app_env):
|
|
runner = CliRunner()
|
|
with app_env.app_context():
|
|
result = runner.invoke(workshop_refresh, [])
|
|
assert result.exit_code == 0, result.output
|
|
assert "enqueued refresh_workshop_items job" in result.output
|
|
with session_scope() as db:
|
|
jobs = db.scalars(select(Job).where(Job.operation == "refresh_workshop_items")).all()
|
|
assert len(jobs) == 1
|
|
assert jobs[0].state == "queued"
|
|
assert jobs[0].user_id is None
|
|
assert jobs[0].server_id is None
|
|
|
|
|
|
def test_workshop_refresh_is_idempotent_when_job_queued(app_env):
|
|
runner = CliRunner()
|
|
with app_env.app_context():
|
|
first = runner.invoke(workshop_refresh, [])
|
|
second = runner.invoke(workshop_refresh, [])
|
|
assert first.exit_code == 0
|
|
assert second.exit_code == 0
|
|
assert "already queued" in second.output
|
|
with session_scope() as db:
|
|
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"
|