virtuality/tests/test_audit_fixes.py

"""Regressions for the defects found by the 0.11 audit."""
import json
from pathlib import Path
from urllib.parse import quote, unquote

import app
import conftest
import network_core

CDROM_XML = conftest.DUMPXML.replace(
    "<disk type='file' device='disk'><target dev='vda' bus='virtio'/></disk>",
    "<disk type='file' device='disk'><target dev='vda' bus='virtio'/></disk><disk type='file' device='cdrom'><source file='/var/lib/virtuality/iso/old.iso'/><target dev='sda' bus='sata'/></disk>",
)


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):
    import core

    calls = recording(monkeypatch, app)
    core_calls = recording(monkeypatch, core)
    disk = data_dirs["images"] / "web01.qcow2"
    disk.write_bytes(b"QFI\xfb")
    response = logged_in.post("/vm/web01/delete", follow_redirects=False)
    assert response.headers["location"].startswith("/?message=")
    assert find(calls, "virsh", "destroy", "web01")
    undefine = find(core_calls, "virsh", "undefine", "web01")[0]
    assert "--nvram" in undefine and "--snapshots-metadata" in undefine
    assert not disk.exists() and find(core_calls, "virsh", "pool-refresh", "virtuality-images")


def test_vm_delete_keeps_installer_iso_and_foreign_disks(logged_in, data_dirs, monkeypatch):
    """--remove-all-storage удалял вставленный ISO: загруженные образы пропадали вместе с машиной."""
    import core

    iso = data_dirs["iso"] / "smk-server-_1.iso"
    iso.write_bytes(b"ISO")
    disk = data_dirs["images"] / "web01.qcow2"
    disk.write_bytes(b"QFI\xfb")
    foreign = data_dirs["disk_images"] / "shared.raw"
    foreign.write_bytes(b"raw")
    listing = f" Type   Device   Target   Source\n----\n file   disk     vda      {disk}\n file   disk     vdb      {foreign}\n file   cdrom    sda      {iso}\n"
    recording(monkeypatch, app)
    core_calls = recording(monkeypatch, core, [(["virsh", "domblklist"], conftest.fake_result(listing))])
    response = logged_in.post("/vm/web01/delete", follow_redirects=False)
    assert not disk.exists() and iso.exists() and foreign.exists()
    assert "--remove-all-storage" not in find(core_calls, "virsh", "undefine", "web01")[0]
    assert str(foreign) in unquote(response.headers["location"])


def test_vm_delete_failure_is_reported(logged_in, data_dirs, monkeypatch):
    import core

    recording(monkeypatch, app)
    recording(monkeypatch, core, [(["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=")


def test_vm_delete_stops_paused_machine_first(logged_in, data_dirs, monkeypatch):
    import core

    calls = recording(monkeypatch, app)
    core_calls = recording(monkeypatch, core, [(["virsh", "domstate"], conftest.fake_result("paused"))])
    logged_in.post("/vm/web01/delete", follow_redirects=False)
    assert find(calls, "virsh", "destroy", "web01") and find(core_calls, "virsh", "undefine", "web01")


def test_vm_delete_waits_for_running_task(logged_in, data_dirs, monkeypatch):
    import core

    calls = recording(monkeypatch, app)
    backup = core.new_operation("backup", "Резервная копия web01", vm_name="web01")
    core.update_operation(backup, status="running")
    response = logged_in.post("/vm/web01/delete", follow_redirects=False)
    assert "Резервная копия web01" in unquote(response.headers["location"])
    assert not find(calls, "virsh", "undefine") and not find(calls, "virsh", "destroy")


def test_vm_delete_removes_its_port_forwards(logged_in, data_dirs, monkeypatch):
    recording(monkeypatch, app)
    rule = {"guest_ip": "192.168.100.51", "external_port_start": 2222, "external_port_end": 2222, "guest_port_start": 22, "guest_port_end": 22, "protocol": "tcp"}
    data_dirs["network"].joinpath("port_forwards.json").write_text(json.dumps([
        {"id": "1", "vm_name": "web01", **rule},
        {"id": "2", "vm_name": "db01", **rule, "external_port_start": 2223, "external_port_end": 2223, "guest_ip": "192.168.100.52"},
    ]))
    response = logged_in.post("/vm/web01/delete", follow_redirects=False)
    assert "убраны пробросы портов: 1" in unquote(response.headers["location"])
    assert [item["vm_name"] for item in network_core.load_port_forwards()] == ["db01"]
    assert "2222" not in (data_dirs["nft"] / "virtuality.nft").read_text()


def test_delete_is_offered_for_running_machine(logged_in):
    html = logged_in.get("/vm/web01").text  # в тестах web01 работает
    assert 'action="/vm/web01/delete"' in html and "она будет сразу выключена" in html
    dashboard = logged_in.get("/").text
    assert 'action="/vm/web01/delete"' in dashboard and 'action="/vm/db01/delete"' in dashboard


# ---------------------------------------------------------------- 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()) == []


# ---------------------------------------------------------------- operations orphaned by a restart
def test_orphaned_operations_are_closed_on_startup(data_dirs):
    import core

    running = core.new_operation("snapshot_create", "Снимок web01", vm_name="web01")
    core.update_operation(running, status="running")
    done = core.new_operation("backup", "Копия web01", vm_name="web01")
    core.finish_operation(done, True, "Готово")
    assert core.active_operations_for("web01")
    assert core.interrupt_orphaned_operations() == 1
    assert core.read_operation(running["id"])["status"] == "error" and core.read_operation(running["id"])["interrupted"]
    assert core.read_operation(done["id"])["status"] == "success"
    assert core.active_operations_for("web01") == []


def test_panel_startup_closes_orphaned_operations(data_dirs):
    """Стартовые задачи панели запускает lifespan (FastAPI on_event устарел)."""
    import core
    from fastapi.testclient import TestClient

    running = core.new_operation("snapshot_create", "Снимок web01", vm_name="web01")
    core.update_operation(running, status="running")
    with TestClient(app.app, base_url="http://panel.test"):
        assert core.read_operation(running["id"])["interrupted"] is True


def test_interrupt_leaves_operations_of_other_live_processes(data_dirs):
    """Копии по расписанию идут в отдельной службе: перезапуск панели не должен объявлять их прерванными."""
    import subprocess

    import core

    sleeper = subprocess.Popen(["sleep", "30"])
    finished = subprocess.Popen(["sleep", "0.1"])
    finished_owner = core.process_owner(finished.pid)
    finished.wait()
    try:
        scheduled = core.new_operation("backup", "Резервная копия web01 по расписанию", vm_name="web01")
        core.update_operation(scheduled, status="running", owner=core.process_owner(sleeper.pid))
        dead = core.new_operation("backup", "Резервная копия db01", vm_name="db01")
        core.update_operation(dead, status="running", owner=finished_owner)
        legacy = core.new_operation("backup", "Задача старой версии без владельца", vm_name="db01")
        core.update_operation(legacy, status="running", owner=None)
        assert core.interrupt_orphaned_operations() == 2
        assert core.read_operation(scheduled["id"])["status"] == "running"
        assert core.read_operation(dead["id"])["interrupted"] and core.read_operation(legacy["id"])["interrupted"]
    finally:
        sleeper.kill()
        sleeper.wait()


def test_snapshot_names_with_spaces_are_listed():
    from features import snapshots

    rows = snapshots.parse_snapshot_list(" Name              Creation Time               State\n------\n before update     2026-09-25 10:12:03 +0000   shutoff\n clean             2026-09-10 12:05:00 +0000   running\n")
    assert [row["name"] for row in rows] == ["before update", "clean"]


# ---------------------------------------------------------------- second audit round
def test_vm_names_that_virsh_reads_as_id_or_uuid_are_rejected():
    import core

    for module in (app, core, network_core):
        assert module.valid_vm_name("web01") and module.valid_vm_name("1c-server")
        assert not module.valid_vm_name("10") and not module.valid_vm_name("6c1b3c1e-0000-0000-0000-000000000001")


def test_subprocesses_use_c_locale():
    import os

    import core

    assert core.STORAGE_DIR and os.environ["LC_ALL"] == "C.UTF-8"  # set when core is imported


def test_lease_is_not_guessed_without_mac(data_dirs, monkeypatch):
    recording(monkeypatch, network_core, [
        (["virsh", "domifaddr"], conftest.fake_result(ok=False, stderr="domain is not running")),
        (["virsh", "domiflist"], conftest.fake_result(ok=False, stderr="failed to get domain")),
        (["virsh", "net-dhcp-leases"], conftest.fake_result(" 2026-09-26 10:00:00  52:54:00:11:22:33  ipv4  192.168.100.77/24  other  -")),
    ])
    assert network_core.resolve_vm_ip("gone") is None


def test_stopped_machines_have_no_stale_ip(data_dirs, monkeypatch):
    calls = recording(monkeypatch, app)
    rows = {row["name"]: row for row in app.parse_virsh_list()}
    assert rows["db01"]["ip"] == "—" and rows["web01"]["ip"] == "192.168.100.51"
    assert not [cmd for cmd in calls if cmd[:2] == ["virsh", "domifaddr"] and cmd[-1] == "db01"]


def test_system_disk_named_like_iso_is_not_shown_as_iso(data_dirs, monkeypatch):
    xml = conftest.DUMPXML.replace("<disk type='file' device='disk'><target dev='vda' bus='virtio'/></disk>", "<disk type='file' device='disk'><source file='/var/lib/virtuality/images/web.iso.qcow2'/><target dev='vda' bus='virtio'/></disk><disk type='file' device='cdrom'><target dev='sda' bus='sata'/></disk>")
    recording(monkeypatch, app, [(["virsh", "dumpxml"], conftest.fake_result(xml))])
    assert app.current_vm_iso("web01") == ""


def test_iso_in_use_is_not_deleted(logged_in, data_dirs, monkeypatch):
    iso = data_dirs["iso"] / "old.iso"
    iso.write_bytes(b"iso")
    xml = CDROM_XML.replace("/var/lib/virtuality/iso/old.iso", str(iso))
    recording(monkeypatch, app, [(["virsh", "dumpxml"], conftest.fake_result(xml)), (["virsh", "list", "--all", "--name"], conftest.fake_result("web01\n"))])
    response = logged_in.post("/iso/old.iso/delete", follow_redirects=False)
    assert "error=" in response.headers["location"] and iso.exists()


def test_arm_board_is_ready_without_vmx_flags(monkeypatch):
    import host_profile

    monkeypatch.setattr(host_profile.platform, "machine", lambda: "aarch64")
    monkeypatch.setattr(host_profile, "cpu_virtualization_flags_count", lambda: 0)
    profile = host_profile.detect_host_profile()
    assert not [check for check in profile["checks"] if check["name"] == "Поддержка виртуализации процессором"]