f81e839ba2
- validate instance names at the host lib and web boundary against
[a-z0-9][a-z0-9_-]{0,63} to prevent path traversal via Server.name
- fail-closed on SECRET_KEY: load_config returns None when env unset,
create_app raises if missing or "dev" outside TESTING
- close login timing oracle by hashing a dummy digest when the user
is not found, equalizing response time
- set SESSION_COOKIE_SECURE outside TESTING
- delete_instance tolerates stop_service and fusermount3 failures so
partially-initialized instances clean up without contract breaks;
drops the is_mount() preflight that violated AGENTS.md
- document claim_next_job's single-process assumption
- clarify emit_step contract via docstring
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
122 lines
4.5 KiB
Python
122 lines
4.5 KiB
Python
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from l4d2host.instances import (
|
|
delete_instance,
|
|
initialize_instance,
|
|
start_instance,
|
|
stop_instance,
|
|
)
|
|
|
|
|
|
def test_start_order(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
calls: list[list[str]] = []
|
|
|
|
def fake_run_command(cmd, **kwargs):
|
|
del kwargs
|
|
calls.append(list(cmd))
|
|
|
|
instance_dir = tmp_path / "instances" / "alpha"
|
|
runtime_dir = tmp_path / "runtime" / "alpha"
|
|
(runtime_dir / "merged" / "left4dead2" / "cfg").mkdir(parents=True, exist_ok=True)
|
|
instance_dir.mkdir(parents=True, exist_ok=True)
|
|
(instance_dir / "instance.env").write_text(
|
|
"L4D2_PORT=27015\nL4D2_ARGS=-tickrate 100\nL4D2_LOWERDIRS=/x:/y\n"
|
|
)
|
|
(instance_dir / "server.cfg").write_text("sv_consistency 1")
|
|
|
|
monkeypatch.setattr("l4d2host.instances.run_command", fake_run_command)
|
|
monkeypatch.setattr("l4d2host.service_control.run_command", fake_run_command)
|
|
|
|
start_instance("alpha", root=tmp_path)
|
|
|
|
assert calls[0][0] == "fuse-overlayfs"
|
|
assert calls[1] == ["sudo", "-n", "/usr/local/libexec/left4me/left4me-systemctl", "start", "alpha"]
|
|
|
|
|
|
def test_delete_missing_is_noop(tmp_path: Path) -> None:
|
|
delete_instance("missing", root=tmp_path)
|
|
|
|
|
|
def test_delete_succeeds_when_stop_service_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
calls: list[list[str]] = []
|
|
|
|
def fake_run_command(cmd, **kwargs):
|
|
del kwargs
|
|
calls.append(list(cmd))
|
|
if cmd[:2] == ["sudo", "-n"] and "left4me-systemctl" in cmd[2] and "stop" in cmd:
|
|
raise subprocess.CalledProcessError(
|
|
returncode=5,
|
|
cmd=list(cmd),
|
|
stderr="Unit left4me-server@alpha.service not loaded.",
|
|
)
|
|
|
|
(tmp_path / "instances" / "alpha").mkdir(parents=True)
|
|
(tmp_path / "runtime" / "alpha" / "merged").mkdir(parents=True)
|
|
|
|
monkeypatch.setattr("l4d2host.instances.run_command", fake_run_command)
|
|
monkeypatch.setattr("l4d2host.service_control.run_command", fake_run_command)
|
|
|
|
delete_instance("alpha", root=tmp_path)
|
|
|
|
assert not (tmp_path / "instances" / "alpha").exists()
|
|
assert not (tmp_path / "runtime" / "alpha").exists()
|
|
|
|
|
|
@pytest.mark.parametrize("bad_name", ["..", "../escape", "foo/bar", " foo", "Foo"])
|
|
def test_lifecycle_rejects_unsafe_instance_names(tmp_path: Path, bad_name: str) -> None:
|
|
for func in (start_instance, stop_instance, delete_instance):
|
|
with pytest.raises(ValueError):
|
|
func(bad_name, root=tmp_path)
|
|
with pytest.raises(ValueError):
|
|
initialize_instance(bad_name, tmp_path / "spec.yaml", root=tmp_path)
|
|
assert not (tmp_path / "instances").exists()
|
|
assert not (tmp_path / "runtime").exists()
|
|
|
|
|
|
def test_delete_stopped_instance_removes_dirs(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
calls: list[list[str]] = []
|
|
|
|
def fake_run_command(cmd, **kwargs):
|
|
del kwargs
|
|
calls.append(list(cmd))
|
|
|
|
(tmp_path / "instances" / "alpha").mkdir(parents=True)
|
|
(tmp_path / "runtime" / "alpha" / "merged").mkdir(parents=True)
|
|
|
|
monkeypatch.setattr("l4d2host.instances.run_command", fake_run_command)
|
|
monkeypatch.setattr("l4d2host.service_control.run_command", fake_run_command)
|
|
|
|
delete_instance("alpha", root=tmp_path)
|
|
|
|
assert not (tmp_path / "instances" / "alpha").exists()
|
|
assert not (tmp_path / "runtime" / "alpha").exists()
|
|
assert ["sudo", "-n", "/usr/local/libexec/left4me/left4me-systemctl", "stop", "alpha"] in calls
|
|
|
|
|
|
def test_delete_succeeds_when_unmount_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
fusermount_calls: list[list[str]] = []
|
|
|
|
def fake_run_command(cmd, **kwargs):
|
|
del kwargs
|
|
if cmd and cmd[0] == "fusermount3":
|
|
fusermount_calls.append(list(cmd))
|
|
raise subprocess.CalledProcessError(
|
|
returncode=1,
|
|
cmd=list(cmd),
|
|
stderr="fusermount3: entry for merged not found in /etc/mtab",
|
|
)
|
|
|
|
(tmp_path / "instances" / "alpha").mkdir(parents=True)
|
|
(tmp_path / "runtime" / "alpha" / "merged").mkdir(parents=True)
|
|
|
|
monkeypatch.setattr("l4d2host.instances.run_command", fake_run_command)
|
|
monkeypatch.setattr("l4d2host.service_control.run_command", fake_run_command)
|
|
|
|
delete_instance("alpha", root=tmp_path)
|
|
|
|
assert fusermount_calls, "delete must always attempt fusermount3 -u (no preflight)"
|
|
assert not (tmp_path / "instances" / "alpha").exists()
|
|
assert not (tmp_path / "runtime" / "alpha").exists()
|