437 lines
16 KiB
Python
437 lines
16 KiB
Python
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 ")
|
|
|
|
|
|
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]
|
|
|
|
|