"""Regressions for the defects found by the 0.11 audit."""
import json
from pathlib import Path
from urllib.parse import quote
import app
import conftest
import network_core
CDROM_XML = conftest.DUMPXML.replace(
"",
"",
)
def recording(monkeypatch, module, overrides=()):
calls: list[list[str]] = []
def run(cmd, timeout=12, **kwargs):
calls.append(list(cmd))
for prefix, result in overrides:
if cmd[: len(prefix)] == list(prefix):
return result
return conftest.fake_run_cmd(cmd, timeout, **kwargs)
monkeypatch.setattr(module, "run_cmd", run)
return calls
def find(calls, *prefix):
return [cmd for cmd in calls if cmd[: len(prefix)] == list(prefix)]
# ---------------------------------------------------------------- disk image format (app.py:620)
def test_disk_image_format_asks_qemu_img(data_dirs, monkeypatch):
# Ubuntu cloud images are qcow2 files named *.img: the suffix must not decide the format.
assert app.disk_image_format(Path("/tmp/noble-server-cloudimg-amd64.img")) == "qcow2"
recording(monkeypatch, app, [(["qemu-img", "info"], conftest.fake_result(ok=False, stderr="no such file"))])
assert app.disk_image_format(Path("/tmp/disk.img")) == "raw"
assert app.disk_image_format(Path("/tmp/disk.qcow2")) == "qcow2"
# ---------------------------------------------------------------- VM actions (app.py:1977, 1981)
def test_vm_action_reports_virsh_error(logged_in, data_dirs, monkeypatch):
recording(monkeypatch, app, [(["virsh", "start"], conftest.fake_result(ok=False, stderr="error: Domain is already active"))])
response = logged_in.post("/vm/web01/start", follow_redirects=False)
assert response.status_code == 303
assert response.headers["location"] == "/vm/web01?error=" + quote("Машина уже запущена")
assert "Машина уже запущена" in logged_in.get("/vm/web01?error=" + quote("Машина уже запущена")).text
def test_vm_action_success_redirects_without_error(logged_in, data_dirs, monkeypatch):
recording(monkeypatch, app)
assert logged_in.post("/vm/web01/shutdown", follow_redirects=False).headers["location"] == "/vm/web01"
def test_vm_delete_undefines_uefi_machines(logged_in, data_dirs, monkeypatch):
calls = recording(monkeypatch, app)
response = logged_in.post("/vm/web01/delete", follow_redirects=False)
assert response.headers["location"].startswith("/?message=")
assert find(calls, "virsh", "destroy", "web01")
undefine = find(calls, "virsh", "undefine", "web01")[0]
assert "--nvram" in undefine and "--remove-all-storage" in undefine and "--snapshots-metadata" in undefine
def test_vm_delete_failure_is_reported(logged_in, data_dirs, monkeypatch):
recording(monkeypatch, app, [(["virsh", "undefine"], conftest.fake_result(ok=False, stderr="error: Requested operation is not valid: cannot undefine domain with nvram"))])
response = logged_in.post("/vm/web01/delete", follow_redirects=False)
assert response.headers["location"].startswith("/vm/web01?error=")
# ---------------------------------------------------------------- ISO in the CD-ROM drive (app.py:1150, 1158)
def test_iso_mount_uses_change_media_for_existing_drive(logged_in, data_dirs, monkeypatch):
iso = data_dirs["iso"] / "debian.iso"
iso.write_bytes(b"iso")
calls = recording(monkeypatch, app, [(["virsh", "dumpxml"], conftest.fake_result(CDROM_XML))])
ok, message = app.mount_vm_iso("web01", str(iso))
assert ok, message
assert find(calls, "virsh", "change-media", "web01", "sda", str(iso), "--update", "--config", "--live")
assert not find(calls, "virsh", "detach-disk") and not find(calls, "virsh", "attach-disk")
ok, message = app.detach_vm_iso("web01")
assert ok and find(calls, "virsh", "change-media", "web01", "sda", "--eject", "--config", "--live")
def test_iso_unmount_never_guesses_targets(data_dirs, monkeypatch):
# web01 has only a virtio data disk: nothing may be detached blindly.
calls = recording(monkeypatch, app)
ok, message = app.detach_vm_iso("web01")
assert not ok and message == "Подключенный ISO не найден."
assert not find(calls, "virsh", "detach-disk") and not find(calls, "virsh", "change-media")
def test_iso_mount_adds_drive_to_config_when_missing(data_dirs, monkeypatch):
iso = data_dirs["iso"] / "debian.iso"
iso.write_bytes(b"iso")
calls = recording(monkeypatch, app)
ok, message = app.mount_vm_iso("web01", str(iso))
assert ok and "перезапуска" in message
attach = find(calls, "virsh", "attach-disk", "web01")[0]
assert "--config" in attach and "--live" not in attach
# ---------------------------------------------------------------- port forwards leave no stale firewall rules (network_core.py:459)
def test_iptables_fallback_removes_stale_rules(data_dirs, monkeypatch):
listing = "\n".join([
"-A FORWARD -i eth0 -o virbr1 -p tcp -d 192.168.100.51 -m tcp --dport 22 -m comment --comment virtuality-forward -j ACCEPT",
"-A FORWARD -j SOMETHING_ELSE",
])
calls = recording(monkeypatch, network_core, [(["iptables", "-t", "filter", "-S", "FORWARD"], conftest.fake_result(listing))])
network_core.apply_iptables_fallback([])
deleted = find(calls, "iptables", "-t", "filter", "-D", "FORWARD")
assert deleted == [["iptables", "-t", "filter", "-D", "FORWARD", "-i", "eth0", "-o", "virbr1", "-p", "tcp", "-d", "192.168.100.51", "-m", "tcp", "--dport", "22", "-m", "comment", "--comment", "virtuality-forward", "-j", "ACCEPT"]]
checked = find(calls, "iptables", "-t", "nat", "-C", "POSTROUTING")
assert checked and "virtuality-forward" in checked[0]
def test_ufw_rules_follow_forwards(data_dirs, monkeypatch):
old_rule = ["allow", "2222/tcp"]
network_core.UFW_STATE_FILE.write_text(json.dumps([old_rule, ["route", "allow", "in", "on", "eth0", "out", "on", "virbr1", "to", "192.168.100.51", "port", "22", "proto", "tcp"]]))
calls = recording(monkeypatch, network_core, [(["ufw", "status"], conftest.fake_result("Status: active"))])
network_core.apply_ufw_route_rules([])
assert find(calls, "ufw", "delete", "allow", "2222/tcp")
assert find(calls, "ufw", "route", "delete", "allow", "in", "on", "eth0")
assert json.loads(network_core.UFW_STATE_FILE.read_text()) == []