227 lines
7 KiB
Python
227 lines
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,
|
|
ServerPlayerSession,
|
|
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 == []
|
|
|
|
|
|
def _player(steam_id: str = "76561197960828710", name: str = "Alice",
|
|
connected: int = 21, ping: int = 60) -> PlayerRow:
|
|
return PlayerRow(
|
|
steam_id_64=steam_id, name=name, connected_seconds=connected, ping=ping,
|
|
)
|
|
|
|
|
|
def test_new_player_opens_session_with_backfilled_join(tmp_path, monkeypatch) -> None:
|
|
app, sid = _seed(tmp_path)
|
|
monkeypatch.setattr(
|
|
live_state_poller, "query_status",
|
|
lambda *a, **kw: _status(players=1, roster=[_player(connected=30)]),
|
|
)
|
|
|
|
with app.app_context():
|
|
live_state_poller.poll_once()
|
|
|
|
with session_scope() as db:
|
|
sessions = db.scalars(
|
|
select(ServerPlayerSession).where(ServerPlayerSession.server_id == sid)
|
|
).all()
|
|
assert len(sessions) == 1
|
|
s = sessions[0]
|
|
assert s.steam_id_64 == "76561197960828710"
|
|
assert s.name_at_join == "Alice"
|
|
assert s.left_at is None
|
|
assert s.min_ping == 60
|
|
assert s.max_ping == 60
|
|
# joined_at should be ~30s before now
|
|
delta = (datetime.now(UTC).replace(tzinfo=None) - s.joined_at).total_seconds()
|
|
assert 25 <= delta <= 60
|
|
|
|
|
|
def test_session_closes_when_player_no_longer_in_roster(tmp_path, monkeypatch) -> None:
|
|
app, sid = _seed(tmp_path)
|
|
snapshots = iter([
|
|
_status(players=1, roster=[_player()]),
|
|
_status(players=0, roster=[]),
|
|
])
|
|
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:
|
|
sessions = db.scalars(
|
|
select(ServerPlayerSession).where(ServerPlayerSession.server_id == sid)
|
|
).all()
|
|
assert len(sessions) == 1
|
|
assert sessions[0].left_at is not None
|
|
|
|
|
|
def test_session_ping_range_extends(tmp_path, monkeypatch) -> None:
|
|
app, sid = _seed(tmp_path)
|
|
snapshots = iter([
|
|
_status(players=1, roster=[_player(ping=60)]),
|
|
_status(players=1, roster=[_player(ping=200)]),
|
|
_status(players=1, roster=[_player(ping=40)]),
|
|
])
|
|
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()
|
|
live_state_poller.poll_once()
|
|
|
|
with session_scope() as db:
|
|
s = db.scalar(select(ServerPlayerSession).where(ServerPlayerSession.server_id == sid))
|
|
assert s.min_ping == 40
|
|
assert s.max_ping == 200
|
|
|
|
|
|
def test_session_skips_bots(tmp_path, monkeypatch) -> None:
|
|
# Bots have non-STEAM uniqueid; our parser already drops them, but verify
|
|
# that even if a non-STEAM steam_id_64 makes it through, sessions filter
|
|
# it out. We exercise the filter directly by giving a string that doesn't
|
|
# match the expected SteamID64 numeric form.
|
|
app, sid = _seed(tmp_path)
|
|
monkeypatch.setattr(
|
|
live_state_poller, "query_status",
|
|
lambda *a, **kw: _status(
|
|
players=0, roster=[_player(steam_id="BOT", name="Coach")]
|
|
),
|
|
)
|
|
|
|
with app.app_context():
|
|
live_state_poller.poll_once()
|
|
|
|
with session_scope() as db:
|
|
sessions = db.scalars(
|
|
select(ServerPlayerSession).where(ServerPlayerSession.server_id == sid)
|
|
).all()
|
|
assert sessions == []
|