cronekorkn/left4me
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
left4me/l4d2web/tests/test_l4d2_facade.py
mwiegand 730ef09967
feat(log-streaming): enable srcds log streaming + temp UDP capture listener 
Every managed server now auto-injects log on / mp_logdetail 3 / logaddress_add
into its generated server.cfg, streaming HL Log Standard events to a UDP
listener bundled with l4d2web. The listener is deliberately capture-only —
raw packets land in flat files per source address — so we can observe what
L4D2 actually emits on our servers before committing to a schema or event
vocabulary. Match/round/event model is a Phase 2 plan informed by that data.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-19 23:22:00 +02:00

472 lines
17 KiB
Python
Raw Blame History

from datetime import UTC, datetime
from pathlib import Path
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,
BlueprintOverlay,
Overlay,
OverlayWorkshopItem,
Server,
User,
WorkshopItem,
)
from l4d2web.services.host_commands import CommandResult
@pytest.fixture
def server_with_blueprint(tmp_path, monkeypatch):
db_url = f"sqlite:///{tmp_path/'facade.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()
with app.app_context():
with session_scope() as session:
user = User(username="alice", password_digest=hash_password("secret"), admin=False)
session.add(user)
session.flush()
overlay = Overlay(name="Standard Overlay", path="standard", type="workshop", user_id=user.id)
session.add(overlay)
session.flush()
blueprint = Blueprint(
user_id=user.id,
name="default",
arguments='["-tickrate 100"]',
config='["sv_consistency 1"]',
)
session.add(blueprint)
session.flush()
session.add(BlueprintOverlay(blueprint_id=blueprint.id, overlay_id=overlay.id, position=0))
server = Server(user_id=user.id, blueprint_id=blueprint.id, name="alpha", port=27015)
session.add(server)
session.flush()
server_id = server.id
user_id = user.id
return server_id, user_id
def test_initialize_uses_l4d2ctl_with_latest_blueprint_data(
monkeypatch: pytest.MonkeyPatch,
server_with_blueprint,
) -> None:
calls: list[list[str]] = []
def fake_run_command(cmd, **kwargs):
del kwargs
calls.append(list(cmd))
spec_path = Path(cmd[cmd.index("-f") + 1])
spec = spec_path.read_text()
assert "sv_consistency 1" in spec
assert "standard" in spec
assert "Standard Overlay" not in spec
return CommandResult(returncode=0, stdout="", stderr="")
monkeypatch.setattr("l4d2web.services.host_commands.run_command", fake_run_command)
monkeypatch.setattr(
"l4d2web.services.l4d2_facade.initialize_instance",
lambda *args, **kwargs: pytest.fail("facade must not call l4d2host.initialize_instance directly"),
raising=False,
)
from l4d2web.services.l4d2_facade import initialize_server
server_id, _ = server_with_blueprint
initialize_server(server_id)
assert calls[0][:3] == ["l4d2ctl", "initialize", str(server_id)]
assert calls[0][3] == "-f"
def test_install_and_lifecycle_commands_use_l4d2ctl(
monkeypatch: pytest.MonkeyPatch,
server_with_blueprint,
) -> None:
calls: list[list[str]] = []
def fake_run_command(cmd, **kwargs):
del kwargs
calls.append(list(cmd))
return CommandResult(returncode=0, stdout="", stderr="")
monkeypatch.setattr("l4d2web.services.host_commands.run_command", fake_run_command)
for name in ["SteamInstaller", "start_instance", "stop_instance", "delete_instance", "reset_instance"]:
monkeypatch.setattr(
f"l4d2web.services.l4d2_facade.{name}",
lambda *args, **kwargs: pytest.fail(f"facade must not call l4d2host {name} directly"),
raising=False,
)
from l4d2web.services.l4d2_facade import (
delete_server,
install_runtime,
reset_server,
start_server,
stop_server,
)
server_id, _ = server_with_blueprint
install_runtime()
start_server(server_id)
stop_server(server_id)
reset_server(server_id)
delete_server(server_id)
unit = str(server_id)
# start now auto-initializes before launching, so the call list interleaves
# an `initialize` before the `start`.
initialize_call = [c for c in calls if c[:2] == ["l4d2ctl", "initialize"]]
assert len(initialize_call) == 1
operational = [c for c in calls if c[:2] != ["l4d2ctl", "initialize"]]
assert operational == [
["l4d2ctl", "install"],
["l4d2ctl", "start", unit],
["l4d2ctl", "stop", unit],
["l4d2ctl", "reset", unit],
["l4d2ctl", "delete", unit],
]
def test_server_status_parses_l4d2ctl_json(monkeypatch: pytest.MonkeyPatch) -> None:
calls: list[list[str]] = []
def fake_run_command(cmd, **kwargs):
del kwargs
calls.append(list(cmd))
return CommandResult(
returncode=0,
stdout='{"state":"running","raw_active_state":"active","raw_sub_state":"running"}',
stderr="",
)
monkeypatch.setattr("l4d2web.services.host_commands.run_command", fake_run_command)
monkeypatch.setattr(
"l4d2web.services.l4d2_facade.get_instance_status",
lambda *args, **kwargs: pytest.fail("facade must not call l4d2host.get_instance_status directly"),
raising=False,
)
from l4d2web.services.l4d2_facade import server_status
status = server_status("alpha")
assert calls == [["l4d2ctl", "status", "alpha", "--json"]]
assert status.state == "running"
assert status.raw_active_state == "active"
assert status.raw_sub_state == "running"
def test_server_logs_stream_l4d2ctl_logs(monkeypatch: pytest.MonkeyPatch) -> None:
calls: list[list[str]] = []
def fake_stream_command(cmd):
calls.append(list(cmd))
return iter(["one", "two"])
monkeypatch.setattr("l4d2web.services.host_commands.stream_command", fake_stream_command)
monkeypatch.setattr(
"l4d2web.services.l4d2_facade.stream_instance_logs",
lambda *args, **kwargs: pytest.fail("facade must not call l4d2host.stream_instance_logs directly"),
raising=False,
)
from l4d2web.services.l4d2_facade import stream_server_logs
lines = list(stream_server_logs("alpha", lines=10, follow=False))
assert calls == [["l4d2ctl", "logs", "alpha", "--lines", "10", "--no-follow"]]
assert lines == ["one", "two"]
def _attach_workshop_overlay_to_blueprint(
server_id: int, user_id: int, *, item_cached: bool, tmp_path: Path
) -> tuple[int, str]:
"""Add a workshop overlay with a single workshop item to the server's
blueprint. Returns (overlay_id, steam_id)."""
with session_scope() as session:
server = session.query(Server).filter_by(id=server_id).one()
overlay = Overlay(name="ws", path="placeholder", type="workshop", user_id=user_id)
session.add(overlay)
session.flush()
# Path matches id, like the production create_overlay flow does.
overlay.path = str(overlay.id)
wi = WorkshopItem(
steam_id="1001",
title="A",
filename="a.vpk",
file_url="https://example.com/a.vpk",
file_size=3,
time_updated=1700000000,
last_downloaded_at=datetime.now(UTC) if item_cached else None,
)
session.add(wi)
session.flush()
session.add(
BlueprintOverlay(blueprint_id=server.blueprint_id, overlay_id=overlay.id, position=1)
)
session.add(OverlayWorkshopItem(overlay_id=overlay.id, workshop_item_id=wi.id))
overlay_id = overlay.id
if item_cached:
cache_root = tmp_path / "workshop_cache"
cache_root.mkdir(exist_ok=True)
(cache_root / "1001.vpk").write_bytes(b"abc")
return overlay_id, "1001"
def test_build_payload_emits_alias_and_exec_lines_for_exposed_overlays(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
db_url = f"sqlite:///{tmp_path/'payload.db'}"
monkeypatch.setenv("DATABASE_URL", db_url)
app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
init_db()
with app.app_context():
with session_scope() as session:
user = User(username="alice", password_digest=hash_password("secret"), admin=False)
session.add(user)
session.flush()
low = Overlay(name="low", path="42", type="script", user_id=user.id)
mid = Overlay(name="mid", path="43", type="script", user_id=user.id)
high = Overlay(name="high", path="44", type="script", user_id=user.id)
session.add_all([low, mid, high])
session.flush()
blueprint = Blueprint(
user_id=user.id,
name="bp",
arguments='[]',
config='["sv_consistency 1", "mp_disable_autokick 1"]',
)
session.add(blueprint)
session.flush()
session.add(BlueprintOverlay(
blueprint_id=blueprint.id, overlay_id=low.id, position=0, expose_server_cfg=True,
))
session.add(BlueprintOverlay(
blueprint_id=blueprint.id, overlay_id=mid.id, position=1, expose_server_cfg=False,
))
session.add(BlueprintOverlay(
blueprint_id=blueprint.id, overlay_id=high.id, position=2, expose_server_cfg=True,
))
server = Server(user_id=user.id, blueprint_id=blueprint.id, name="alpha", port=27020)
session.add(server)
session.flush()
server_id = server.id
low_id, high_id = low.id, high.id
from l4d2web.services.l4d2_facade import (
build_server_spec_payload,
load_server_blueprint_bundle,
)
with app.app_context():
server, blueprint, overlay_rows = load_server_blueprint_bundle(server_id)
payload = build_server_spec_payload(server, blueprint, overlay_rows)
assert payload["overlays"] == [
{"path": "42", "alias": f"overlay_{low_id}"},
{"path": "43"},
{"path": "44", "alias": f"overlay_{high_id}"},
]
# First in list = topmost = highest precedence, so its exec runs LAST.
# Position 0 (low_id) is first in list → exec'd last; position 2 (high_id)
# is last in list → exec'd first.
assert payload["config"] == [
f"exec server_overlay_{high_id}",
f"exec server_overlay_{low_id}",
"sv_consistency 1",
"mp_disable_autokick 1",
]
def test_initialize_does_not_run_overlay_builders(
monkeypatch: pytest.MonkeyPatch, server_with_blueprint, tmp_path
) -> None:
"""Initialize must NOT run overlay builders — those run on overlay save.
Running them on every server Start is expensive and redundant.
"""
server_id, user_id = server_with_blueprint
overlay_id, _steam_id = _attach_workshop_overlay_to_blueprint(
server_id, user_id, item_cached=True, tmp_path=tmp_path
)
monkeypatch.setattr(
"l4d2web.services.host_commands.run_command",
lambda *args, **kwargs: CommandResult(returncode=0, stdout="", stderr=""),
)
from l4d2web.services.l4d2_facade import initialize_server
initialize_server(server_id)
addons = tmp_path / "overlays" / str(overlay_id) / "left4dead2" / "addons"
assert not (addons / "1001.vpk").exists(), "initialize must not trigger workshop builder"
def test_initialize_fails_fast_on_uncached_workshop_items(
monkeypatch: pytest.MonkeyPatch, server_with_blueprint, tmp_path
) -> None:
server_id, user_id = server_with_blueprint
overlay_id, steam_id = _attach_workshop_overlay_to_blueprint(
server_id, user_id, item_cached=False, tmp_path=tmp_path
)
invocations: list[list[str]] = []
def fake_run_command(cmd, **kwargs):
invocations.append(list(cmd))
return CommandResult(returncode=0, stdout="", stderr="")
monkeypatch.setattr("l4d2web.services.host_commands.run_command", fake_run_command)
from l4d2web.services.l4d2_facade import initialize_server
with pytest.raises(Exception) as excinfo:
initialize_server(server_id)
msg = str(excinfo.value)
assert steam_id in msg
assert str(overlay_id) in msg or "ws" in msg
# l4d2ctl initialize MUST NOT run when uncached items are present.
assert all("initialize" not in cmd for cmd in invocations), invocations
# ---------------------------------------------------------------------------
# build_server_spec_payload — rcon_password injection
# ---------------------------------------------------------------------------
def _make_server_blueprint(rcon: str = "") -> tuple[Server, Blueprint]:
bp = Blueprint(
id=1, user_id=1, name="bp",
arguments='["-tickrate","60"]',
config='["sv_consistency 1","mp_gamemode coop"]',
)
srv = Server(
id=1, user_id=1, blueprint_id=1, name="s", port=27500,
rcon_password=rcon,
)
return srv, bp
def test_build_server_spec_payload_appends_rcon_password_last() -> None:
from l4d2web.services.l4d2_facade import build_server_spec_payload
srv, bp = _make_server_blueprint(rcon="topsecret123")
overlays = [(7, "/overlays/7", True), (8, "/overlays/8", False)]
spec = build_server_spec_payload(srv, bp, overlays)
cfg = spec["config"]
# rcon_password line is the LAST entry — overlay exec lines + blueprint
# config + rcon_password.
assert cfg[-1] == 'rcon_password "topsecret123"'
# Lines before our injection still contain the blueprint config.
assert "sv_consistency 1" in cfg
assert "mp_gamemode coop" in cfg
def test_build_server_spec_payload_omits_rcon_password_when_empty() -> None:
from l4d2web.services.l4d2_facade import build_server_spec_payload
srv, bp = _make_server_blueprint(rcon="")
spec = build_server_spec_payload(srv, bp, [])
for line in spec["config"]:
assert not line.startswith("rcon_password ")
# ---------------------------------------------------------------------------
# build_server_spec_payload — hostname injection
# ---------------------------------------------------------------------------
def test_build_server_spec_payload_injects_hostname() -> None:
from l4d2web.services.l4d2_facade import build_server_spec_payload
bp = Blueprint(id=1, user_id=1, name="bp", arguments="[]", config='["sv_consistency 1"]')
srv = Server(id=1, user_id=1, blueprint_id=1, name="alpha", port=27015, rcon_password="sekret")
spec = build_server_spec_payload(srv, bp, [], resolved_hostname="My Server")
cfg = spec["config"]
assert 'hostname "My Server"' in cfg
assert cfg[-1] == 'rcon_password "sekret"'
def test_build_server_spec_payload_omits_hostname_when_empty() -> None:
from l4d2web.services.l4d2_facade import build_server_spec_payload
bp = Blueprint(id=1, user_id=1, name="bp", arguments="[]", config="[]")
srv = Server(id=1, user_id=1, blueprint_id=1, name="alpha", port=27015, rcon_password="sekret")
spec = build_server_spec_payload(srv, bp, [])
for line in spec["config"]:
assert not line.startswith("hostname ")
# ---------------------------------------------------------------------------
# build_server_spec_payload — log streaming cvar injection
# ---------------------------------------------------------------------------
def test_build_server_spec_payload_injects_log_cvars() -> None:
from l4d2web.services.l4d2_facade import build_server_spec_payload
srv, bp = _make_server_blueprint(rcon="sekret")
spec = build_server_spec_payload(
srv, bp, [], log_listener_addr="127.0.0.1:28000",
)
cfg = spec["config"]
# All three log cvars emitted, AFTER rcon_password (so blueprint can't
# override them — Source cvars are last-wins).
assert "log on" in cfg
assert "mp_logdetail 3" in cfg
assert "logaddress_add 127.0.0.1:28000" in cfg
rcon_idx = cfg.index('rcon_password "sekret"')
log_idx = cfg.index("log on")
assert log_idx > rcon_idx, "log cvars must come after rcon_password"
assert cfg[-1] == "logaddress_add 127.0.0.1:28000"
def test_build_server_spec_payload_omits_log_cvars_when_addr_empty() -> None:
from l4d2web.services.l4d2_facade import build_server_spec_payload
srv, bp = _make_server_blueprint(rcon="sekret")
spec = build_server_spec_payload(srv, bp, []) # default "" → no inject
cfg = spec["config"]
assert "log on" not in cfg
assert "mp_logdetail 3" not in cfg
assert not any(line.startswith("logaddress_add ") for line in cfg)
def test_initialize_server_resolves_fallback_hostname(
monkeypatch: pytest.MonkeyPatch, server_with_blueprint,
) -> None:
"""When server.hostname is empty, deploy emits hostname "<username> <server.name>"."""
from l4d2web.services.l4d2_facade import initialize_server
spec_contents: list[str] = []
def fake_run_command(cmd, **kwargs):
nonlocal spec_contents
spec_path = cmd[cmd.index("-f") + 1]
spec_contents.append(Path(spec_path).read_text())
return CommandResult(returncode=0, stdout="", stderr="")
monkeypatch.setattr("l4d2web.services.host_commands.run_command", fake_run_command)
server_id, _ = server_with_blueprint
initialize_server(server_id)
assert len(spec_contents) == 1
# The fixture creates user "alice" and server named "alpha"
assert 'hostname "alice alpha"' in spec_contents[0]
Reference in a new issue View git blame Copy permalink