124 lines
3.7 KiB
Python
124 lines
3.7 KiB
Python
"""Live-state poller tests.
|
|
|
|
Each test seeds an app + DB, monkeypatches the RCON client to return a
|
|
canned StatusResponse, and asserts on what the poller writes.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, UTC
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from l4d2web.app import create_app
|
|
from l4d2web.db import init_db, session_scope
|
|
from l4d2web.models import (
|
|
Blueprint,
|
|
Server,
|
|
ServerLiveState,
|
|
User,
|
|
)
|
|
from l4d2web.services import live_state_poller
|
|
from l4d2web.services.rcon import PlayerRow, StatusResponse
|
|
|
|
|
|
def _seed(tmp_path):
|
|
db_url = f"sqlite:///{tmp_path/'p.db'}"
|
|
import os
|
|
os.environ["DATABASE_URL"] = db_url
|
|
app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "x"})
|
|
init_db()
|
|
with session_scope() as db:
|
|
u = User(username="u", password_digest="x"); db.add(u); db.flush()
|
|
bp = Blueprint(user_id=u.id, name="bp", arguments="[]", config="[]"); db.add(bp); db.flush()
|
|
s = Server(
|
|
user_id=u.id, blueprint_id=bp.id, name="s", port=27500,
|
|
rcon_password="pw", actual_state="running",
|
|
)
|
|
db.add(s); db.flush()
|
|
return app, s.id
|
|
|
|
|
|
def _status(players: int, map_: str = "c1m1_hotel", hibernating: bool = False,
|
|
roster: list[PlayerRow] | None = None) -> StatusResponse:
|
|
return StatusResponse(
|
|
map=map_, players=players, max_players=4, bots=0,
|
|
hibernating=hibernating, roster=roster or [],
|
|
)
|
|
|
|
|
|
def test_rle_bumps_last_seen_when_state_unchanged(tmp_path, monkeypatch) -> None:
|
|
app, sid = _seed(tmp_path)
|
|
monkeypatch.setattr(
|
|
live_state_poller, "query_status",
|
|
lambda host, port, password, timeout: _status(players=0),
|
|
)
|
|
|
|
with app.app_context():
|
|
live_state_poller.poll_once()
|
|
live_state_poller.poll_once()
|
|
|
|
with session_scope() as db:
|
|
rows = db.scalars(
|
|
select(ServerLiveState).where(ServerLiveState.server_id == sid)
|
|
.order_by(ServerLiveState.started_at)
|
|
).all()
|
|
assert len(rows) == 1
|
|
assert rows[0].players == 0
|
|
assert rows[0].last_seen_at >= rows[0].started_at
|
|
|
|
|
|
def test_rle_inserts_new_row_on_state_change(tmp_path, monkeypatch) -> None:
|
|
app, sid = _seed(tmp_path)
|
|
snapshots = iter([_status(players=0), _status(players=1)])
|
|
monkeypatch.setattr(
|
|
live_state_poller, "query_status",
|
|
lambda *a, **kw: next(snapshots),
|
|
)
|
|
|
|
with app.app_context():
|
|
live_state_poller.poll_once()
|
|
live_state_poller.poll_once()
|
|
|
|
with session_scope() as db:
|
|
rows = db.scalars(
|
|
select(ServerLiveState).where(ServerLiveState.server_id == sid)
|
|
.order_by(ServerLiveState.started_at)
|
|
).all()
|
|
assert [r.players for r in rows] == [0, 1]
|
|
|
|
|
|
def test_skips_servers_without_rcon_password(tmp_path, monkeypatch) -> None:
|
|
app, sid = _seed(tmp_path)
|
|
with session_scope() as db:
|
|
s = db.scalar(select(Server).where(Server.id == sid))
|
|
s.rcon_password = ""
|
|
|
|
called: list = []
|
|
monkeypatch.setattr(
|
|
live_state_poller, "query_status",
|
|
lambda *a, **kw: called.append(1) or _status(0),
|
|
)
|
|
|
|
with app.app_context():
|
|
live_state_poller.poll_once()
|
|
|
|
assert called == []
|
|
|
|
|
|
def test_skips_non_running_servers(tmp_path, monkeypatch) -> None:
|
|
app, sid = _seed(tmp_path)
|
|
with session_scope() as db:
|
|
s = db.scalar(select(Server).where(Server.id == sid))
|
|
s.actual_state = "stopped"
|
|
|
|
called: list = []
|
|
monkeypatch.setattr(
|
|
live_state_poller, "query_status",
|
|
lambda *a, **kw: called.append(1) or _status(0),
|
|
)
|
|
|
|
with app.app_context():
|
|
live_state_poller.poll_once()
|
|
|
|
assert called == []
|