6f49efd44a
- _console_line.html: command + reply, error variant, "(no reply)" placeholder. - server_detail.html: console section between Live State and Files, replays last 50 history rows server-side; HTMX form appends new lines via hx-swap. - console-history.js: ArrowUp/Down recall against /console/history JSON; scroll-to-bottom on load and after each new line. - CSS: fixed-height scrolling transcript, terminal-ish styling, spinner via HTMX in-flight class. - test_console_routes.py: update 4 assertions from legacy [ERROR] literal to console-error CSS class (matches new semantic markup). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
476 lines
17 KiB
Python
476 lines
17 KiB
Python
import json
|
|
from datetime import UTC, datetime
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
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, CommandHistory, Server, User
|
|
from l4d2web.services.rcon import RconAuthError, RconError
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers / fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_app(tmp_path, monkeypatch, db_suffix="console.db"):
|
|
db_url = f"sqlite:///{tmp_path / db_suffix}"
|
|
monkeypatch.setenv("DATABASE_URL", db_url)
|
|
app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
|
|
init_db()
|
|
return app
|
|
|
|
|
|
def _seed(app, *, owner_username="alice", other_username=None):
|
|
"""Seed owner + blueprint + server (port 27015). Optionally seed a second user."""
|
|
with session_scope() as s:
|
|
owner = User(username=owner_username, password_digest=hash_password("x"), admin=False)
|
|
s.add(owner)
|
|
if other_username:
|
|
other = User(username=other_username, password_digest=hash_password("x"), admin=False)
|
|
s.add(other)
|
|
s.flush()
|
|
bp = Blueprint(user_id=owner.id, name="bp", arguments="[]", config="[]")
|
|
s.add(bp)
|
|
s.flush()
|
|
server = Server(user_id=owner.id, blueprint_id=bp.id, name="srv", port=27015)
|
|
s.add(server)
|
|
s.flush()
|
|
result = {"owner_id": owner.id, "server_id": server.id}
|
|
if other_username:
|
|
result["other_id"] = other.id
|
|
return result
|
|
|
|
|
|
def _login(client, user_id):
|
|
with client.session_transaction() as sess:
|
|
sess["user_id"] = user_id
|
|
sess["pw_changed_at"] = datetime.now(UTC).isoformat()
|
|
sess["csrf_token"] = "test-token"
|
|
|
|
|
|
def _post_command(client, server_id, command, *, csrf="test-token"):
|
|
return client.post(
|
|
f"/servers/{server_id}/console",
|
|
data={"command": command, "csrf_token": csrf},
|
|
headers={"X-CSRF-Token": csrf},
|
|
)
|
|
|
|
|
|
def _get_history(client, server_id, **params):
|
|
return client.get(f"/servers/{server_id}/console/history", query_string=params)
|
|
|
|
|
|
def _history_count():
|
|
with session_scope() as s:
|
|
return s.query(CommandHistory).count()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 1. Not logged in → redirect to login
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_not_logged_in_post_returns_csrf_error_or_redirect(tmp_path, monkeypatch):
|
|
# The CSRF before_request runs before load_current_user. An anonymous POST
|
|
# without a session CSRF token will be rejected with 400 (CSRF mismatch)
|
|
# before the require_login decorator can redirect. This is the correct
|
|
# project behavior — anonymous users cannot reach the endpoint at all.
|
|
app = _make_app(tmp_path, monkeypatch, "anon_post.db")
|
|
data = _seed(app)
|
|
client = app.test_client()
|
|
|
|
resp = _post_command(client, data["server_id"], "status")
|
|
# 400 (CSRF) or 302 (auth redirect) are both valid rejection responses.
|
|
assert resp.status_code in (302, 400)
|
|
|
|
|
|
def test_not_logged_in_get_history_redirects(tmp_path, monkeypatch):
|
|
app = _make_app(tmp_path, monkeypatch, "anon_get.db")
|
|
data = _seed(app)
|
|
client = app.test_client()
|
|
|
|
resp = _get_history(client, data["server_id"])
|
|
assert resp.status_code == 302
|
|
assert "/login" in resp.headers["Location"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 2. Logged in but not the owner → 404
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_non_owner_post_returns_404(tmp_path, monkeypatch):
|
|
app = _make_app(tmp_path, monkeypatch, "nonowner_post.db")
|
|
data = _seed(app, other_username="bob")
|
|
client = app.test_client()
|
|
_login(client, data["other_id"])
|
|
|
|
resp = _post_command(client, data["server_id"], "status")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_non_owner_get_history_returns_404(tmp_path, monkeypatch):
|
|
app = _make_app(tmp_path, monkeypatch, "nonowner_get.db")
|
|
data = _seed(app, other_username="bob")
|
|
client = app.test_client()
|
|
_login(client, data["other_id"])
|
|
|
|
resp = _get_history(client, data["server_id"])
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_nonexistent_server_post_returns_404(tmp_path, monkeypatch):
|
|
app = _make_app(tmp_path, monkeypatch, "noserver_post.db")
|
|
data = _seed(app)
|
|
client = app.test_client()
|
|
_login(client, data["owner_id"])
|
|
|
|
resp = _post_command(client, 9999, "status")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 3. Valid command, RCON returns reply
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_valid_command_inserts_history_row(tmp_path, monkeypatch):
|
|
app = _make_app(tmp_path, monkeypatch, "ok.db")
|
|
data = _seed(app)
|
|
client = app.test_client()
|
|
_login(client, data["owner_id"])
|
|
|
|
with patch("l4d2web.routes.console_routes.rcon.execute_command", return_value="pong") as mock_exec:
|
|
resp = _post_command(client, data["server_id"], "ping")
|
|
|
|
assert resp.status_code == 200
|
|
text = resp.get_data(as_text=True)
|
|
assert "ping" in text
|
|
assert "pong" in text
|
|
assert "[ERROR]" not in text
|
|
|
|
mock_exec.assert_called_once_with("127.0.0.1", 27015, mock_exec.call_args[0][2], "ping")
|
|
|
|
with session_scope() as s:
|
|
row = s.query(CommandHistory).one()
|
|
assert row.command == "ping"
|
|
assert row.reply == "pong"
|
|
assert row.is_error is False
|
|
assert row.user_id == data["owner_id"]
|
|
assert row.server_id == data["server_id"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 4. Valid command, RCON returns empty string
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_valid_command_empty_reply(tmp_path, monkeypatch):
|
|
app = _make_app(tmp_path, monkeypatch, "empty_reply.db")
|
|
data = _seed(app)
|
|
client = app.test_client()
|
|
_login(client, data["owner_id"])
|
|
|
|
with patch("l4d2web.routes.console_routes.rcon.execute_command", return_value=""):
|
|
resp = _post_command(client, data["server_id"], "noop")
|
|
|
|
assert resp.status_code == 200
|
|
assert "[ERROR]" not in resp.get_data(as_text=True)
|
|
|
|
with session_scope() as s:
|
|
row = s.query(CommandHistory).one()
|
|
assert row.reply == ""
|
|
assert row.is_error is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 5. RCON raises RconError
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_rcon_error_inserts_error_row(tmp_path, monkeypatch):
|
|
app = _make_app(tmp_path, monkeypatch, "rconerr.db")
|
|
data = _seed(app)
|
|
client = app.test_client()
|
|
_login(client, data["owner_id"])
|
|
|
|
with patch(
|
|
"l4d2web.routes.console_routes.rcon.execute_command",
|
|
side_effect=RconError("connection refused"),
|
|
):
|
|
resp = _post_command(client, data["server_id"], "status")
|
|
|
|
assert resp.status_code == 200
|
|
assert "connection refused" in resp.get_data(as_text=True)
|
|
assert "console-error" in resp.get_data(as_text=True)
|
|
|
|
with session_scope() as s:
|
|
row = s.query(CommandHistory).one()
|
|
assert row.is_error is True
|
|
assert "connection refused" in row.reply
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 6. RCON raises RconAuthError
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_rcon_auth_error_inserts_error_row(tmp_path, monkeypatch):
|
|
app = _make_app(tmp_path, monkeypatch, "autherr.db")
|
|
data = _seed(app)
|
|
client = app.test_client()
|
|
_login(client, data["owner_id"])
|
|
|
|
with patch(
|
|
"l4d2web.routes.console_routes.rcon.execute_command",
|
|
side_effect=RconAuthError("bad rcon password"),
|
|
):
|
|
resp = _post_command(client, data["server_id"], "status")
|
|
|
|
assert resp.status_code == 200
|
|
assert "bad rcon password" in resp.get_data(as_text=True)
|
|
assert "console-error" in resp.get_data(as_text=True)
|
|
|
|
with session_scope() as s:
|
|
row = s.query(CommandHistory).one()
|
|
assert row.is_error is True
|
|
assert "bad rcon password" in row.reply
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 7. Input validation: empty command → no history row
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_empty_command_no_history_row(tmp_path, monkeypatch):
|
|
app = _make_app(tmp_path, monkeypatch, "empty_cmd.db")
|
|
data = _seed(app)
|
|
client = app.test_client()
|
|
_login(client, data["owner_id"])
|
|
|
|
# execute_command raises ValueError for empty commands; the route catches it
|
|
# before hitting rcon, but we patch anyway to be explicit.
|
|
with patch(
|
|
"l4d2web.routes.console_routes.rcon.execute_command",
|
|
side_effect=ValueError("command must not be empty or whitespace-only"),
|
|
):
|
|
resp = _post_command(client, data["server_id"], "")
|
|
|
|
assert resp.status_code == 200
|
|
assert "console-error" in resp.get_data(as_text=True)
|
|
assert _history_count() == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 8. Input validation: oversized command → no history row
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_oversized_command_no_history_row(tmp_path, monkeypatch):
|
|
app = _make_app(tmp_path, monkeypatch, "oversized.db")
|
|
data = _seed(app)
|
|
client = app.test_client()
|
|
_login(client, data["owner_id"])
|
|
|
|
with patch(
|
|
"l4d2web.routes.console_routes.rcon.execute_command",
|
|
side_effect=ValueError("command exceeds maximum byte length of 1000"),
|
|
):
|
|
resp = _post_command(client, data["server_id"], "x" * 1001)
|
|
|
|
assert resp.status_code == 200
|
|
assert "console-error" in resp.get_data(as_text=True)
|
|
assert _history_count() == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 9. CSRF token missing/invalid → 400
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_missing_csrf_token_rejected(tmp_path, monkeypatch):
|
|
app = _make_app(tmp_path, monkeypatch, "csrf.db")
|
|
data = _seed(app)
|
|
client = app.test_client()
|
|
_login(client, data["owner_id"])
|
|
|
|
# Send with a wrong CSRF token (session has "test-token" from _login).
|
|
resp = client.post(
|
|
f"/servers/{data['server_id']}/console",
|
|
data={"command": "status"},
|
|
# No X-CSRF-Token header and no csrf_token form field.
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 10. GET /console/history with rows in DB → newest-first JSON list
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_get_history_returns_newest_first(tmp_path, monkeypatch):
|
|
app = _make_app(tmp_path, monkeypatch, "hist_order.db")
|
|
data = _seed(app)
|
|
client = app.test_client()
|
|
_login(client, data["owner_id"])
|
|
|
|
with session_scope() as s:
|
|
for cmd in ["alpha", "beta", "gamma"]:
|
|
s.add(
|
|
CommandHistory(
|
|
user_id=data["owner_id"],
|
|
server_id=data["server_id"],
|
|
command=cmd,
|
|
reply="ok",
|
|
is_error=False,
|
|
)
|
|
)
|
|
|
|
resp = _get_history(client, data["server_id"])
|
|
assert resp.status_code == 200
|
|
rows = json.loads(resp.get_data(as_text=True))
|
|
assert [r["command"] for r in rows] == ["gamma", "beta", "alpha"]
|
|
for r in rows:
|
|
assert "id" in r
|
|
assert "command" in r
|
|
# reply should NOT be included
|
|
assert "reply" not in r
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 11. GET /console/history?before=<id> → only rows with id < before
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_get_history_before_pagination(tmp_path, monkeypatch):
|
|
app = _make_app(tmp_path, monkeypatch, "hist_before.db")
|
|
data = _seed(app)
|
|
client = app.test_client()
|
|
_login(client, data["owner_id"])
|
|
|
|
row_ids = []
|
|
with session_scope() as s:
|
|
for cmd in ["one", "two", "three"]:
|
|
row = CommandHistory(
|
|
user_id=data["owner_id"],
|
|
server_id=data["server_id"],
|
|
command=cmd,
|
|
reply="",
|
|
is_error=False,
|
|
)
|
|
s.add(row)
|
|
s.flush()
|
|
row_ids.append(row.id)
|
|
|
|
pivot = row_ids[2] # id of "three"
|
|
resp = _get_history(client, data["server_id"], before=pivot)
|
|
assert resp.status_code == 200
|
|
rows = json.loads(resp.get_data(as_text=True))
|
|
# Should only have "two" and "one" (ids < pivot), newest first.
|
|
assert len(rows) == 2
|
|
assert rows[0]["command"] == "two"
|
|
assert rows[1]["command"] == "one"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 12. GET /console/history?limit=10000 → clamped to ≤200
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_get_history_limit_clamped(tmp_path, monkeypatch):
|
|
app = _make_app(tmp_path, monkeypatch, "hist_clamp.db")
|
|
data = _seed(app)
|
|
client = app.test_client()
|
|
_login(client, data["owner_id"])
|
|
|
|
# Monkeypatch the max-limit constant to 3 so we can verify the clamp
|
|
# without inserting 200+ rows.
|
|
monkeypatch.setattr("l4d2web.routes.console_routes._HISTORY_MAX_LIMIT", 3)
|
|
|
|
# Insert 5 rows — more than the patched max of 3.
|
|
with session_scope() as s:
|
|
for i in range(5):
|
|
s.add(
|
|
CommandHistory(
|
|
user_id=data["owner_id"],
|
|
server_id=data["server_id"],
|
|
command=f"cmd{i}",
|
|
reply="",
|
|
is_error=False,
|
|
)
|
|
)
|
|
|
|
# Request with limit=100 (far above the patched max of 3).
|
|
resp = _get_history(client, data["server_id"], limit=100)
|
|
assert resp.status_code == 200
|
|
rows = json.loads(resp.get_data(as_text=True))
|
|
# Clamp must cap at _HISTORY_MAX_LIMIT (3), not at the requested 100.
|
|
assert len(rows) == 3
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 13 (in test_pages.py style, placed here): server_detail renders console_history
|
|
# scoped to current user, oldest-first
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_server_detail_console_history_scoped_and_ordered(tmp_path, monkeypatch):
|
|
"""Verify server_detail loads console_history scoped to the owner, oldest-first.
|
|
|
|
The server_detail template does not render console_history items yet (the
|
|
UI subagent will add that). We therefore verify the data contract by:
|
|
1. Checking the page loads (200).
|
|
2. Directly querying the DB to confirm only the owner's rows exist.
|
|
3. Testing ordering by verifying row IDs ascend (oldest-first) after the
|
|
route's list(reversed(...)) logic — done via a separate call to the
|
|
history endpoint which already has ordering tested above.
|
|
"""
|
|
app = _make_app(tmp_path, monkeypatch, "detail_hist.db")
|
|
data = _seed(app, other_username="bob")
|
|
client = app.test_client()
|
|
_login(client, data["owner_id"])
|
|
|
|
with session_scope() as s:
|
|
# Insert rows for alice (owner) — intentionally inserted newest first
|
|
# so that ordering logic is meaningful.
|
|
row_second = CommandHistory(
|
|
user_id=data["owner_id"], server_id=data["server_id"],
|
|
command="second", reply="r2", is_error=False,
|
|
)
|
|
s.add(row_second)
|
|
s.flush()
|
|
second_id = row_second.id
|
|
|
|
row_first = CommandHistory(
|
|
user_id=data["owner_id"], server_id=data["server_id"],
|
|
command="first", reply="r1", is_error=False,
|
|
)
|
|
s.add(row_first)
|
|
s.flush()
|
|
# Insert a row for the other user — must NOT be included.
|
|
s.add(CommandHistory(
|
|
user_id=data["other_id"], server_id=data["server_id"],
|
|
command="other_cmd", reply="x", is_error=False,
|
|
))
|
|
|
|
# The detail page must load successfully.
|
|
resp = client.get(f"/servers/{data['server_id']}")
|
|
assert resp.status_code == 200
|
|
|
|
# Verify scoping: the history endpoint is already tested for ordering above.
|
|
# Here we confirm via GET /console/history that only alice's rows are returned.
|
|
hist_resp = _get_history(client, data["server_id"])
|
|
assert hist_resp.status_code == 200
|
|
rows = json.loads(hist_resp.get_data(as_text=True))
|
|
commands = [r["command"] for r in rows]
|
|
# Only alice's commands appear.
|
|
assert "other_cmd" not in commands
|
|
assert "first" in commands
|
|
assert "second" in commands
|
|
# Newest-first from history endpoint; IDs must descend.
|
|
ids = [r["id"] for r in rows]
|
|
assert ids == sorted(ids, reverse=True)
|