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.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_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): 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.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()