985df970f8
Each linked overlay gets a checkbox on the blueprint detail page that opts
its server.cfg in as exec server_overlay_<id>. The web app builds the
spec with {path, alias} per overlay and prepends exec server_overlay_<id>
lines to the blueprint config in lowest-overlay-first order. The host
stages those copies in the overlayfs upper layer before mounting (avoids
copy-up writes against a sandbox-uid file). A live preview block above the
Config textarea shows what gets auto-executed.
Schema:
- alembic 0007: BlueprintOverlay.expose_server_cfg BOOLEAN
Spec contract:
- l4d2host OverlayRef(path, alias?). load_spec accepts both bare-string
and {path, alias} entries.
Side effects folded in (same file in l4d2_facade):
- start_server auto-initializes; the manual Initialize step is no longer
needed before Start.
- initialize_server no longer runs blueprint builders — builds happen on
overlay save, not on every server Start.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
277 lines
11 KiB
Python
277 lines
11 KiB
Python
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from l4d2host.instances import (
|
|
delete_instance,
|
|
initialize_instance,
|
|
reset_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")
|
|
(instance_dir / "spec.yaml").write_text("port: 27015\noverlays: [x, y]\n")
|
|
|
|
monkeypatch.setattr("l4d2host.fs.kernel_overlayfs.run_command", fake_run_command)
|
|
monkeypatch.setattr("l4d2host.service_control.run_command", fake_run_command)
|
|
|
|
start_instance("alpha", root=tmp_path)
|
|
|
|
assert calls[0] == [
|
|
"sudo",
|
|
"-n",
|
|
"/usr/local/libexec/left4me/left4me-overlay",
|
|
"mount",
|
|
"alpha",
|
|
]
|
|
assert calls[1] == ["sudo", "-n", "/usr/local/libexec/left4me/left4me-systemctl", "start", "alpha"]
|
|
|
|
|
|
def test_start_copies_per_overlay_aliases_and_sweeps_stale(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
def fake_run_command(cmd, **kwargs):
|
|
del cmd, kwargs
|
|
|
|
instance_dir = tmp_path / "instances" / "alpha"
|
|
runtime_dir = tmp_path / "runtime" / "alpha"
|
|
upper_cfg_dir = runtime_dir / "upper" / "left4dead2" / "cfg"
|
|
upper_cfg_dir.mkdir(parents=True, exist_ok=True)
|
|
(runtime_dir / "merged").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=\nL4D2_LOWERDIRS=/x\n"
|
|
)
|
|
(instance_dir / "server.cfg").write_text("exec server_overlay_5\n")
|
|
(instance_dir / "spec.yaml").write_text(
|
|
"port: 27015\n"
|
|
"overlays:\n"
|
|
" - {path: '5', alias: overlay_5}\n"
|
|
" - {path: '6', alias: overlay_6}\n"
|
|
" - path: '7'\n"
|
|
)
|
|
src_5 = tmp_path / "overlays" / "5" / "left4dead2" / "cfg"
|
|
src_5.mkdir(parents=True, exist_ok=True)
|
|
(src_5 / "server.cfg").write_text("sv_consistency 1\n")
|
|
src_7 = tmp_path / "overlays" / "7" / "left4dead2" / "cfg"
|
|
src_7.mkdir(parents=True, exist_ok=True)
|
|
(src_7 / "server.cfg").write_text("ignored: alias not set\n")
|
|
(upper_cfg_dir / "server_orphan.cfg").write_text("from previous start\n")
|
|
|
|
monkeypatch.setattr("l4d2host.fs.kernel_overlayfs.run_command", fake_run_command)
|
|
monkeypatch.setattr("l4d2host.service_control.run_command", fake_run_command)
|
|
|
|
start_instance("alpha", root=tmp_path)
|
|
|
|
assert (upper_cfg_dir / "server.cfg").read_text() == "exec server_overlay_5\n"
|
|
assert (upper_cfg_dir / "server_overlay_5.cfg").read_text() == "sv_consistency 1\n"
|
|
assert not (upper_cfg_dir / "server_overlay_6.cfg").exists(), "no source server.cfg → no alias"
|
|
assert not (upper_cfg_dir / "server_orphan.cfg").exists(), "stale alias must be swept"
|
|
assert not (upper_cfg_dir / "server_overlay_7.cfg").exists(), "no alias in spec → no copy"
|
|
|
|
|
|
def test_start_refuses_to_double_mount(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").mkdir(parents=True)
|
|
instance_dir.mkdir(parents=True)
|
|
(instance_dir / "instance.env").write_text("L4D2_PORT=27015\nL4D2_ARGS=\nL4D2_LOWERDIRS=/x\n")
|
|
(instance_dir / "server.cfg").write_text("")
|
|
|
|
merged = runtime_dir / "merged"
|
|
|
|
def fake_ismount(path):
|
|
return Path(path) == merged
|
|
|
|
monkeypatch.setattr("l4d2host.fs.kernel_overlayfs.run_command", fake_run_command)
|
|
monkeypatch.setattr("l4d2host.service_control.run_command", fake_run_command)
|
|
monkeypatch.setattr("l4d2host.instances.os.path.ismount", fake_ismount)
|
|
|
|
with pytest.raises(subprocess.CalledProcessError) as exc_info:
|
|
start_instance("alpha", root=tmp_path)
|
|
|
|
assert "already mounted" in (exc_info.value.stderr or "")
|
|
assert calls == [], "no mount/start commands must be issued when refusing"
|
|
|
|
|
|
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.fs.kernel_overlayfs.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, reset_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_reset_stops_unmounts_and_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))
|
|
|
|
instance_dir = tmp_path / "instances" / "alpha"
|
|
runtime_dir = tmp_path / "runtime" / "alpha"
|
|
instance_dir.mkdir(parents=True)
|
|
(runtime_dir / "merged").mkdir(parents=True)
|
|
(instance_dir / "instance.env").write_text("L4D2_PORT=27015\n")
|
|
(runtime_dir / "upper" / "logs").mkdir(parents=True)
|
|
(runtime_dir / "upper" / "logs" / "console.log").write_text("noise")
|
|
|
|
monkeypatch.setattr("l4d2host.fs.kernel_overlayfs.run_command", fake_run_command)
|
|
monkeypatch.setattr("l4d2host.service_control.run_command", fake_run_command)
|
|
|
|
reset_instance("alpha", root=tmp_path)
|
|
|
|
assert not instance_dir.exists()
|
|
assert not runtime_dir.exists()
|
|
assert any("left4me-systemctl" in arg for cmd in calls for arg in cmd)
|
|
assert any("stop" in cmd for cmd in calls)
|
|
|
|
|
|
def test_reset_on_never_initialized_is_noop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""No instance/runtime directories yet — reset should still attempt the
|
|
stop+unmount (both suppressed on failure) and not raise."""
|
|
def fake_run_command(cmd, **kwargs):
|
|
del kwargs
|
|
if "stop" in cmd:
|
|
raise subprocess.CalledProcessError(returncode=5, cmd=list(cmd), stderr="not loaded")
|
|
|
|
monkeypatch.setattr("l4d2host.fs.kernel_overlayfs.run_command", fake_run_command)
|
|
monkeypatch.setattr("l4d2host.service_control.run_command", fake_run_command)
|
|
|
|
reset_instance("alpha", root=tmp_path)
|
|
|
|
assert not (tmp_path / "instances" / "alpha").exists()
|
|
assert not (tmp_path / "runtime" / "alpha").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.fs.kernel_overlayfs.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_stop_succeeds_when_unmount_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
umount_calls: list[list[str]] = []
|
|
|
|
def fake_run_command(cmd, **kwargs):
|
|
del kwargs
|
|
if cmd[:4] == [
|
|
"sudo",
|
|
"-n",
|
|
"/usr/local/libexec/left4me/left4me-overlay",
|
|
"umount",
|
|
]:
|
|
umount_calls.append(list(cmd))
|
|
raise subprocess.CalledProcessError(
|
|
returncode=1,
|
|
cmd=list(cmd),
|
|
stderr="umount: /var/lib/left4me/runtime/alpha/merged: not mounted",
|
|
)
|
|
|
|
monkeypatch.setattr("l4d2host.fs.kernel_overlayfs.run_command", fake_run_command)
|
|
monkeypatch.setattr("l4d2host.service_control.run_command", fake_run_command)
|
|
|
|
stop_instance("alpha", root=tmp_path)
|
|
|
|
assert umount_calls, "stop must always attempt the overlay helper (no preflight)"
|
|
|
|
|
|
def test_delete_succeeds_when_unmount_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
umount_calls: list[list[str]] = []
|
|
|
|
def fake_run_command(cmd, **kwargs):
|
|
del kwargs
|
|
if cmd[:4] == [
|
|
"sudo",
|
|
"-n",
|
|
"/usr/local/libexec/left4me/left4me-overlay",
|
|
"umount",
|
|
]:
|
|
umount_calls.append(list(cmd))
|
|
raise subprocess.CalledProcessError(
|
|
returncode=1,
|
|
cmd=list(cmd),
|
|
stderr="umount: /var/lib/left4me/runtime/alpha/merged: not mounted",
|
|
)
|
|
|
|
(tmp_path / "instances" / "alpha").mkdir(parents=True)
|
|
(tmp_path / "runtime" / "alpha" / "merged").mkdir(parents=True)
|
|
|
|
monkeypatch.setattr("l4d2host.fs.kernel_overlayfs.run_command", fake_run_command)
|
|
monkeypatch.setattr("l4d2host.service_control.run_command", fake_run_command)
|
|
|
|
delete_instance("alpha", root=tmp_path)
|
|
|
|
assert umount_calls, "delete must always attempt the overlay helper (no preflight)"
|
|
assert not (tmp_path / "instances" / "alpha").exists()
|
|
assert not (tmp_path / "runtime" / "alpha").exists()
|