virtuality/tests/test_app.py

import uuid
from pathlib import Path
from urllib.parse import unquote

import pytest

import app

PAGES = ["/", "/help", "/host", "/iso", "/disk-images", "/network", "/operations", "/logs", "/update", "/vm/create", "/vm/web01", "/vm/web01/console"]


def test_healthz_is_public(client):
    response = client.get("/healthz")
    assert response.status_code == 200
    assert response.json()["ok"] is True
    assert response.json()["version"] == (app.BASE_DIR.parent / "VERSION").read_text().strip()


def test_security_headers(client):
    response = client.get("/login")
    assert response.headers["x-frame-options"] == "SAMEORIGIN"
    assert response.headers["x-content-type-options"] == "nosniff"


@pytest.mark.parametrize("path", PAGES)
def test_pages_require_login(client, path):
    response = client.get(path, follow_redirects=False)
    assert response.status_code == 303
    assert response.headers["location"] == "/login"


@pytest.mark.parametrize("path", PAGES)
def test_pages_render(logged_in, path):
    response = logged_in.get(path)
    assert response.status_code == 200, response.text[:500]


def test_wrong_user_is_rejected(client, monkeypatch):
    monkeypatch.setattr(app.auth, "verify_password", lambda user, password: True)
    response = client.post("/login", data={"username": "intruder", "password": "x"}, follow_redirects=False)
    assert response.status_code == 401


def test_login_throttle(client, monkeypatch):
    monkeypatch.setattr(app.auth, "verify_password", lambda user, password: False)
    for _ in range(5):
        assert client.post("/login", data={"username": "tester", "password": "bad"}).status_code == 401
    response = client.post("/login", data={"username": "tester", "password": "bad"})
    assert response.status_code == 429
    assert int(response.headers["retry-after"]) > 0


def test_session_cookie_flags(client, monkeypatch):
    monkeypatch.setattr(app.auth, "verify_password", lambda user, password: True)
    response = client.post("/login", data={"username": "tester", "password": "x"}, follow_redirects=False)
    cookie = response.headers["set-cookie"].lower()
    assert "httponly" in cookie and "samesite=lax" in cookie


def test_forged_cookie_is_rejected(client):
    client.cookies.set("virtuality_session", "forged.token.value")
    assert client.get("/", follow_redirects=False).status_code == 303


def test_cross_origin_post_blocked(logged_in):
    response = logged_in.post("/vm/web01/start", headers={"Origin": "http://evil.example"}, follow_redirects=False)
    assert response.status_code == 403


def test_same_origin_post_allowed(logged_in):
    response = logged_in.post("/vm/web01/start", headers={"Origin": "http://panel.test"}, follow_redirects=False)
    assert response.status_code == 303


def test_vm_action_rejects_bad_name(logged_in):
    assert logged_in.post("/vm/-bad/start", follow_redirects=False).status_code == 400


def test_vm_action_unknown(logged_in):
    assert logged_in.post("/vm/web01/format-disk", follow_redirects=False).status_code == 400


def test_vm_details_include_autostart(data_dirs):
    details = app.vm_details("web01")
    assert details["autostart_enabled"] is True
    assert details["autostart_label"] == "enabled"


def test_vm_detail_shows_autostart_enabled(logged_in):
    html = logged_in.get("/vm/web01").text
    assert 'action="/vm/web01/autostart-disable"' in html
    assert 'role="switch" aria-checked="true"' in html


def test_dashboard_uses_plain_language(logged_in):
    html = logged_in.get("/").text
    assert "Работает" in html and "Выключена" in html
    assert "2 ядра" in html and "2 ГБ" in html
    assert "shut off" not in html


def test_vm_create_preselects_image(logged_in, data_dirs):
    (data_dirs["iso"] / "debian.iso").write_bytes(b"iso")
    html = logged_in.get("/vm/create?iso=debian.iso").text
    assert f'value="{data_dirs["iso"] / "debian.iso"}" data-label="debian.iso" data-os="" data-login="" checked' in html


def test_vm_create_disk_image_without_disk_size(logged_in, data_dirs, monkeypatch):
    image = data_dirs["disk_images"] / "cloud.qcow2"
    image.write_bytes(b"qcow")
    captured = {}
    monkeypatch.setattr(app, "start_background_operation", lambda operation, cmd: captured.update(cmd=cmd))
    monkeypatch.setattr(app.network_core, "create_nat_network", lambda: {})
    form = {"name": "cloud-vm", "memory": 1024, "vcpus": 1, "disk_image_path": str(image), "source_type": "disk_image", "network_mode": "nat"}
    response = logged_in.post("/vm/create", data=form, follow_redirects=False)
    assert response.status_code == 303, response.text[:300]
    assert "--import" in captured["cmd"][-1]


def test_help_page(logged_in):
    assert "Как создать машину" in logged_in.get("/help").text


def test_redirect_message_is_url_encoded(logged_in, monkeypatch):
    monkeypatch.setattr(app, "apply_vm_boot_order", lambda name, order: (True, "Готово: диск & ISO?"))
    response = logged_in.post("/vm/web01/boot-order", data={"boot_order": "disk"}, follow_redirects=False)
    location = response.headers["location"]
    assert " " not in location and "&" not in location.split("?", 1)[1]
    assert unquote(location.split("boot_message=", 1)[1]) == "Готово: диск & ISO?"


def test_vm_create_iso_builds_virt_install(logged_in, data_dirs, monkeypatch):
    iso = data_dirs["iso"] / "debian.iso"
    iso.write_bytes(b"iso")
    captured = {}
    monkeypatch.setattr(app, "start_background_operation", lambda operation, cmd: captured.update(operation=operation, cmd=cmd))
    monkeypatch.setattr(app.network_core, "create_nat_network", lambda: {})
    form = {"name": "new-vm", "memory": 2048, "vcpus": 2, "disk_size": 20, "iso_path": str(iso), "source_type": "iso", "guest_arch": "x86_64", "boot_order": "auto", "network_mode": "nat"}
    response = logged_in.post("/vm/create", data=form, follow_redirects=False)
    assert response.status_code == 303, response.text[:500]
    cmd = captured["cmd"]
    assert cmd[0] == "virt-install"
    assert "--cdrom" in cmd and str(iso) in cmd
    assert any(part.startswith("network=virtuality-nat,model=virtio,mac=52:54:00:") for part in cmd)
    assert captured["operation"]["boot_order"] == "disk_cdrom"
    # После установки гость перезагружается с диска: без особого режима установки и с диском первым в очереди.
    assert cmd[cmd.index("--boot") + 1] == "hd,cdrom"
    assert cmd[cmd.index("--install") + 1] == "no_install=yes"


def test_vm_create_validation_error(logged_in):
    form = {"name": "x", "memory": 2048, "vcpus": 2, "disk_size": 20, "source_type": "iso", "network_mode": "nat"}
    response = logged_in.post("/vm/create", data=form)
    assert response.status_code == 400


def test_iso_upload_and_delete(logged_in, data_dirs, monkeypatch):
    response = logged_in.post("/iso/upload", files={"iso_file": ("My Distro 1.0.iso", b"data", "application/octet-stream")}, follow_redirects=False)
    assert response.status_code == 303
    assert (data_dirs["iso"] / "My-Distro-1.0.iso").read_bytes() == b"data"
    logged_in.post("/iso/My-Distro-1.0.iso/delete", follow_redirects=False)
    assert not (data_dirs["iso"] / "My-Distro-1.0.iso").exists()


def test_iso_upload_rejects_other_types(logged_in):
    response = logged_in.post("/iso/upload", files={"iso_file": ("evil.sh", b"#!/bin/sh", "text/plain")})
    assert response.status_code == 400


def test_iso_delete_path_traversal(logged_in, data_dirs, tmp_path):
    victim = tmp_path / "victim.iso"
    victim.write_bytes(b"keep")
    logged_in.post("/iso/..%2Fvictim.iso/delete", follow_redirects=False)
    assert victim.exists()


def test_port_forward_add(logged_in, data_dirs):
    form = {"vm_name": "web01", "guest_ip": "192.168.100.51", "external_port": "2222", "guest_port": "22", "protocol": "tcp"}
    response = logged_in.post("/network/forward/add", data=form, follow_redirects=False)
    assert response.status_code == 303
    rules = (data_dirs["nft"] / "virtuality.nft").read_text()
    assert "tcp dport 2222 dnat to 192.168.100.51:22" in rules


def test_api_requires_auth(client):
    for path in ("/api/health", "/api/operations", "/live/status", "/api/logs"):
        assert client.get(path).status_code == 401


def test_live_status(logged_in):
    data = logged_in.get("/live/status").json()
    assert data["ok"] is True
    assert {vm["name"] for vm in data["vms"]} == {"web01", "db01"}


def test_uploader_collects_file_before_disabling_input():
    """Заблокированное поле браузер не отправляет: загрузка уходила пустой (44 байта) и получала 422."""
    script = (Path(__file__).resolve().parents[1] / "web" / "static" / "panel.js").read_text()
    uploader = script[script.index("function wireUploader"):script.index("/* ---------------------------------------------------- operation page */")]
    assert "input.disabled = active" in uploader
    assert uploader.index("new FormData(form)") < uploader.index("setActive(true)")
    assert "xhr.send(data)" in uploader


def test_vm_create_disk_bigger_than_free_space_needs_confirmation(logged_in, data_dirs, monkeypatch):
    """virt-install отказывал с английской ошибкой; теперь панель предупреждает сама и создаёт только с подтверждением."""
    iso = data_dirs["iso"] / "smk.iso"
    iso.write_bytes(b"iso")
    started = []
    monkeypatch.setattr(app, "start_background_operation", lambda operation, cmd: started.append(cmd))
    monkeypatch.setattr(app.network_core, "create_nat_network", lambda: {})
    monkeypatch.setattr(app, "images_free_bytes", lambda: 25 * 1024 ** 3)
    form = {"name": "smk", "memory": 4096, "vcpus": 2, "disk_size": 40, "iso_path": str(iso), "source_type": "iso", "network_mode": "nat"}

    page = logged_in.get("/vm/create").text
    assert "Свободно на сервере: 25" in page and 'name="allow_overcommit"' not in page
    response = logged_in.post("/vm/create", data=form)
    assert response.status_code == 400 and started == []
    assert "Диск на 40 ГБ больше, чем свободно на сервере" in response.text and 'name="allow_overcommit"' in response.text

    response = logged_in.post("/vm/create", data={**form, "allow_overcommit": "1"}, follow_redirects=False)
    assert response.status_code == 303, response.text[:300]
    assert started[0][-2:] == ["--check", "disk_size=off"]

    response = logged_in.post("/vm/create", data={**form, "name": "small", "disk_size": 20}, follow_redirects=False)
    assert response.status_code == 303 and "--check" not in started[1]


def failed_create(data_dirs, name, disk=True, **extra):
    """Операция создания машины, у которой virt-install завершился с ошибкой."""
    disk_path = data_dirs["images"] / f"{name}.qcow2"
    if disk:
        disk_path.write_bytes(b"QFI\xfb")
    operation = {"id": str(uuid.uuid4()), "type": "vm_create", "title": f"Создание VM {name}", "status": "queued", "progress": 0, "message": "", "created_at": app.utc_now(), "updated_at": app.utc_now(), "vm_name": name, "disk_path": str(disk_path), "network_mode": "bridge", **extra}
    app.write_operation(operation)
    app.run_operation_worker(operation["id"], ["bash", "-c", "echo 'ERROR    The requested volume capacity will exceed the available pool space'; exit 1"])
    return app.read_operation(operation["id"]), disk_path


def test_failed_vm_create_cleans_up_and_offers_retry(logged_in, data_dirs):
    """Страница упавшей задачи предлагала «Открыть экран машины», хотя машины нет; недосозданный диск мешал повтору."""
    operation, disk_path = failed_create(data_dirs, "smk", iso_path=str(data_dirs["iso"] / "smk-server-_1.iso"))
    assert operation["status"] == "error" and "ничего не осталось" in operation["message"]
    assert not disk_path.exists() and "Недосозданный диск smk.qcow2 удалён" in operation["log_tail"]
    html = logged_in.get(f"/operations/{operation['id']}").text
    assert 'data-op-done="success" hidden' in html
    assert 'href="/vm/create?iso=smk-server-_1.iso"' in html and "Создать заново" in html


def test_failed_create_of_existing_vm_keeps_its_disk(logged_in, data_dirs):
    operation, disk_path = failed_create(data_dirs, "web01")  # машина определена в libvirt: её диск не трогаем
    assert disk_path.exists() and "ничего не осталось" not in operation["message"]
    html = logged_in.get(f"/operations/{operation['id']}").text
    assert "Создать заново" not in html and 'href="/vm/web01">Перейти к машине' in html


def test_operation_page_reveals_only_buttons_for_its_result():
    script = (Path(__file__).resolve().parents[1] / "web" / "static" / "panel.js").read_text()
    assert '$$(`[data-op-done="${status}"]`)' in script and "$$('[data-op-done]')" not in script


def test_console_websocket_path_works_with_old_and_new_novnc(tmp_path, monkeypatch):
    """noVNC 1.6 (Ubuntu 26.04) считает path от /novnc/vnc.html: без "../" экран стучался в /novnc/console/ws/… и не подключался."""
    import posixpath
    from urllib.parse import parse_qs, urljoin, urlsplit

    (tmp_path / "vnc.html").write_text("<html></html>")
    monkeypatch.setattr(app, "NOVNC_DIR", tmp_path)
    monkeypatch.setattr(app, "vm_vnc_display", lambda name: "127.0.0.1:0")
    info = app.console_info("web01")
    page = "https://panel.test:8443" + info["url"]
    path = parse_qs(urlsplit(page).query)["path"][0]
    token = path.rsplit("/", 1)[1]
    assert app.console_serializer.loads(token) == {"vm": "web01", "port": 5900}
    assert urlsplit(urljoin(page, path)).path == f"/console/ws/{token}"  # noVNC 1.6: new URL(path, location.href)
    assert posixpath.normpath("/" + path) == f"/console/ws/{token}"  # noVNC 1.3: "wss://host:port/" + path