309354942a
Drop .limit(20) from the recent_rows query so the full history window is available for the future recent-players modal; derive recent_players_overview (first 10) and recent_players_total_count from the unbounded result and pass both into _live_state.html alongside the existing recent_players key. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
109 lines
3.6 KiB
Python
109 lines
3.6 KiB
Python
import pytest
|
|
from datetime import UTC, datetime
|
|
|
|
from l4d2web.app import create_app
|
|
from l4d2web.auth import hash_password
|
|
from l4d2web.db import init_db, session_scope
|
|
from l4d2web.models import Blueprint, Server, User
|
|
|
|
|
|
@pytest.fixture
|
|
def owner_client_with_server(tmp_path, monkeypatch):
|
|
db_url = f"sqlite:///{tmp_path/'status.db'}"
|
|
monkeypatch.setenv("DATABASE_URL", db_url)
|
|
app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
|
|
init_db()
|
|
|
|
with session_scope() as session:
|
|
user = User(username="alice", password_digest=hash_password("secret"), admin=False)
|
|
session.add(user)
|
|
session.flush()
|
|
|
|
blueprint = Blueprint(user_id=user.id, name="default", arguments="[]", config="[]")
|
|
session.add(blueprint)
|
|
session.flush()
|
|
|
|
server = Server(user_id=user.id, blueprint_id=blueprint.id, name="alpha", port=27015)
|
|
session.add(server)
|
|
session.flush()
|
|
|
|
user_id = user.id
|
|
server_id = server.id
|
|
|
|
client = app.test_client()
|
|
with client.session_transaction() as sess:
|
|
sess["user_id"] = user_id
|
|
sess["pw_changed_at"] = datetime.now(UTC).isoformat()
|
|
|
|
return client, server_id
|
|
|
|
|
|
def test_owner_can_stream_server_logs(owner_client_with_server, monkeypatch) -> None:
|
|
client, server_id = owner_client_with_server
|
|
|
|
monkeypatch.setattr(
|
|
"l4d2web.services.l4d2_facade.stream_server_logs",
|
|
lambda name, lines=200, follow=True: iter(["first", "second"]),
|
|
)
|
|
|
|
response = client.get(f"/servers/{server_id}/logs/stream")
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_log_stream_translates_heartbeat_to_sse_keepalive(owner_client_with_server, monkeypatch) -> None:
|
|
client, server_id = owner_client_with_server
|
|
|
|
monkeypatch.setattr(
|
|
"l4d2web.services.l4d2_facade.stream_server_logs",
|
|
lambda name, lines=200, follow=True: iter(["first", "", "second"]),
|
|
)
|
|
|
|
response = client.get(f"/servers/{server_id}/logs/stream")
|
|
assert response.status_code == 200
|
|
body = response.get_data(as_text=True)
|
|
assert "data: first\n\n" in body
|
|
assert "data: second\n\n" in body
|
|
assert ": keepalive\n\n" in body
|
|
assert "data: \n\n" not in body
|
|
|
|
|
|
def test_status_precedence() -> None:
|
|
from l4d2web.services.status import compute_display_state
|
|
|
|
assert compute_display_state("start", "stopped") == "starting"
|
|
|
|
|
|
@pytest.mark.xfail(reason="template change in Task 3 will satisfy this", strict=True)
|
|
def test_live_state_exposes_recent_overview_and_total_count(owner_client_with_server) -> None:
|
|
"""Route must expose recent_players_overview (≤10) and recent_players_total_count.
|
|
|
|
The body-text assertions depend on the template change in Task 3.
|
|
The xfail marker will be removed once Task 3 lands.
|
|
"""
|
|
from datetime import timedelta
|
|
from l4d2web.models import ServerPlayerSession
|
|
|
|
client, server_id = owner_client_with_server
|
|
|
|
now = datetime.now(UTC)
|
|
total = 13
|
|
|
|
with session_scope() as db:
|
|
for i in range(total):
|
|
db.add(ServerPlayerSession(
|
|
server_id=server_id,
|
|
steam_id_64=str(76561190000000000 + i),
|
|
joined_at=now - timedelta(hours=i + 2),
|
|
left_at=now - timedelta(hours=i + 1),
|
|
name_at_join=f"Player{i}",
|
|
min_ping=10,
|
|
max_ping=50,
|
|
))
|
|
|
|
res = client.get(f"/servers/{server_id}/live-state")
|
|
assert res.status_code == 200
|
|
html = res.get_data(as_text=True)
|
|
|
|
# These assertions depend on the Task 3 template changes.
|
|
assert "13 Recent" in html
|
|
assert html.count("recent-chip") <= 10
|