virtuality/tests/test_core.py

import json
from pathlib import Path

import pytest

import network_core
from conftest import ROOT
import restore_network
import update_core


def test_parse_port_range():
    assert network_core.parse_port_range("80") == (80, 80)
    assert network_core.parse_port_range("10000-10010") == (10000, 10010)
    assert network_core.parse_port_range("10000:10010") == (10000, 10010)
    for bad in ("", "0", "70000", "20-10", "abc", "80;rm"):
        with pytest.raises(network_core.NetworkError):
            network_core.parse_port_range(bad)


def test_add_port_forward_rejects_overlap(data_dirs):
    network_core.add_port_forward("web01", "192.168.100.51", "8000-8010", "8000-8010", "tcp")
    with pytest.raises(network_core.NetworkError):
        network_core.add_port_forward("web01", "192.168.100.52", "8005", "80", "tcp")
    network_core.add_port_forward("web01", "192.168.100.52", "8005", "80", "udp")


def test_add_port_forward_rejects_mismatched_ranges(data_dirs):
    with pytest.raises(network_core.NetworkError):
        network_core.add_port_forward("web01", "192.168.100.51", "8000-8010", "80", "tcp")


def test_render_nft_rules_range(data_dirs):
    network_core.add_port_forward("web01", "192.168.100.51", "10000-10002", "20000-20002", "udp")
    rules = network_core.render_nft_rules()
    assert 'iifname "eth0" udp dport 10000-10002 dnat to 192.168.100.51:20000-20002' in rules
    assert "masquerade" in rules


def test_restore_network_noop_without_config(data_dirs, capsys):
    assert restore_network.main() == 0
    assert "nothing to restore" in capsys.readouterr().out


def test_restore_network_applies_rules(data_dirs):
    data_dirs["network"].joinpath("port_forwards.json").write_text(json.dumps([
        {"id": "1", "vm_name": "web01", "guest_ip": "192.168.100.51", "external_port_start": 2222, "external_port_end": 2222, "guest_port_start": 22, "guest_port_end": 22, "protocol": "tcp"},
    ]))
    assert restore_network.main() == 0
    assert "dport 2222 dnat to 192.168.100.51:22" in (data_dirs["nft"] / "virtuality.nft").read_text()


def test_version_tuple_ordering():
    assert update_core.version_tuple("0.9.6") < update_core.version_tuple("0.10.0")
    assert update_core.version_tuple("v1.0") == update_core.version_tuple("1.0.0")


def test_missing_versions():
    manifest = {"versions": [{"version": v} for v in ("0.9.5", "0.9.6", "0.10.0", "0.10.1")]}
    missed = update_core.missing_versions("0.9.6", "0.10.1", manifest)
    assert [item["version"] for item in missed] == ["0.10.0", "0.10.1"]


def test_versions_manifest_matches_version_file():
    root = Path(__file__).resolve().parents[1]
    version = (root / "VERSION").read_text().strip()
    manifest = json.loads((root / "updates" / "versions.json").read_text())
    assert version in [item["version"] for item in manifest["versions"]]


def test_presenters_parse_virsh_tables():
    import presenters

    disks = presenters.parse_domblklist(" Type   Device   Target   Source\n------------------------------------\n file   disk     vda      /var/lib/virtuality/images/web01.qcow2\n file   cdrom    sda      -\n")
    assert [d["name"] for d in disks] == ["web01.qcow2", "Пусто"]
    assert disks[1]["is_cdrom"]
    nics = presenters.parse_domiflist(" Interface   Type      Source           Model    MAC\n---\n vnet0       network   virtuality-nat   virtio   52:54:00:aa:bb:cc\n")
    assert nics[0]["mac"] == "52:54:00:aa:bb:cc"
    info = presenters.parse_dominfo("State: shut off\nCPU(s): 4\nMax memory: 8388608 KiB\nUsed memory: 8388608 KiB\nAutostart: disable\n")
    assert (info["vcpus"], info["memory_mb"], info["autostart"]) == (4, 8192, False)
    assert presenters.vm_state("shut off")["label"] == "Выключена"
    assert presenters.format_mb(1536) == "1.5 ГБ"
    assert presenters.plural(22, "ядро", "ядра", "ядер") == "ядра"


def test_update_not_offered_when_local_is_newer(tmp_path, monkeypatch, data_dirs):
    import subprocess

    def git(*args, cwd):
        subprocess.run(["git", *args], cwd=cwd, check=True, capture_output=True)

    remote, local = tmp_path / "remote", tmp_path / "local"
    remote.mkdir()
    git("init", "-q", "-b", "main", cwd=remote)
    git("-c", "user.name=t", "-c", "user.email=t@t", "commit", "-q", "--allow-empty", "-m", "base", cwd=remote)
    git("clone", "-q", str(remote), str(local), cwd=tmp_path)
    (local / "VERSION").write_text("0.10.0\n")
    git("add", "VERSION", cwd=local)
    git("-c", "user.name=t", "-c", "user.email=t@t", "commit", "-q", "-m", "newer local build", cwd=local)
    monkeypatch.setattr(update_core, "SOURCE_DIR", local)
    assert update_core.check_updates(fetch=True)["has_update"] is False

    git("-c", "user.name=t", "-c", "user.email=t@t", "commit", "-q", "--allow-empty", "-m", "upstream", cwd=remote)
    git("reset", "-q", "--hard", "origin/main", cwd=local)
    assert update_core.check_updates(fetch=True)["has_update"] is True


GITEA = "https://git.stackworks.ru/viktor/virtuality"


def make_source(path):
    import subprocess

    def git(*args):
        return subprocess.run(["git", "-c", "user.name=t", "-c", "user.email=t@t", *args], cwd=path, check=True, capture_output=True, text=True).stdout.strip()

    path.mkdir()
    git("init", "-q", "-b", "main")
    git("commit", "-q", "--allow-empty", "-m", "base")
    git("remote", "add", "origin", "https://github.com/viktor138irk/virtuality.git")
    return git


def test_migrate_remote_moves_only_the_old_github_address(tmp_path, monkeypatch, data_dirs):
    git = make_source(tmp_path / "source")
    monkeypatch.setattr(update_core, "SOURCE_DIR", tmp_path / "source")
    assert update_core.migrate_remote() is True
    assert git("remote", "get-url", "origin") == GITEA + ".git"
    assert update_core.migrate_remote() is False
    git("remote", "set-url", "origin", "https://example.org/me/fork.git")
    assert update_core.migrate_remote() is False
    assert git("remote", "get-url", "origin") == "https://example.org/me/fork.git"


def test_update_target_script_downloads_from_gitea(tmp_path):
    import os
    import subprocess

    source = tmp_path / "source"
    git = make_source(source)

    def target(channel, script="resolve_update_target"):
        env = {**os.environ, "VIRTUALITY_UPDATE_CHANNEL": channel}
        code = f'SOURCE_DIR="{source}"; REMOTE=origin; BRANCH=main; . "{ROOT}/scripts/update_target.sh"; {script}; echo "$TARGET_LABEL $TARGET_ZIP_URL"'
        return subprocess.run(["bash", "-c", code], env=env, check=True, capture_output=True, text=True).stdout.strip().splitlines()

    moved, result = target("main", "migrate_update_remote; resolve_update_target")
    assert moved == f"update remote moved: https://github.com/viktor138irk/virtuality.git -> {GITEA}.git"
    assert result == f"main {GITEA}/archive/main.zip"
    assert git("remote", "get-url", "origin") == GITEA + ".git"
    git("tag", "v0.11.1")
    assert target("stable") == [f"v0.11.1 {GITEA}/archive/v0.11.1.zip"]
    assert target("main") == [f"main {GITEA}/archive/main.zip"]
    # A fork downloads from itself; a non-https remote falls back to the project's address.
    git("remote", "set-url", "origin", "https://example.org/me/fork.git")
    assert target("stable", "migrate_update_remote; resolve_update_target") == ["v0.11.1 https://example.org/me/fork/archive/v0.11.1.zip"]
    git("remote", "set-url", "origin", "git@example.org:me/fork.git")
    assert target("stable") == [f"v0.11.1 {GITEA}/archive/v0.11.1.zip"]


def test_stable_channel_follows_release_tags(tmp_path, monkeypatch, data_dirs):
    import subprocess

    def git(*args, cwd):
        subprocess.run(["git", "-c", "user.name=t", "-c", "user.email=t@t", *args], cwd=cwd, check=True, capture_output=True)

    remote, local = tmp_path / "remote", tmp_path / "local"
    remote.mkdir()
    git("init", "-q", "-b", "main", cwd=remote)
    (remote / "VERSION").write_text("0.10.0\n")
    git("add", "VERSION", cwd=remote)
    git("commit", "-q", "-m", "0.10.0", cwd=remote)
    git("clone", "-q", str(remote), str(local), cwd=tmp_path)
    monkeypatch.setattr(update_core, "SOURCE_DIR", local)
    monkeypatch.setattr(update_core, "NODE_ENV", tmp_path / "web.env")
    monkeypatch.delenv("VIRTUALITY_UPDATE_CHANNEL", raising=False)
    assert update_core.update_channel() == "stable"

    # Work on main without a release: stable follows main until the first tag exists.
    (remote / "VERSION").write_text("0.11.0\n")
    git("commit", "-q", "-am", "0.11.0", cwd=remote)
    data = update_core.check_updates(fetch=True)
    assert data["has_update"] and data["target"] == "main"

    # With a release tag, stable offers the tag and ignores newer commits on main.
    git("tag", "v0.11.0", cwd=remote)
    git("commit", "-q", "--allow-empty", "-m", "work in progress", cwd=remote)
    data = update_core.check_updates(fetch=True)
    assert data["channel"] == "stable" and data["target"] == "v0.11.0" and data["latest_version"] == "0.11.0" and data["has_update"]
    git("checkout", "-q", "v0.11.0", cwd=local)
    assert update_core.check_updates(fetch=True)["has_update"] is False

    # The early-access channel sees every change on main.
    (tmp_path / "web.env").write_text("VIRTUALITY_UPDATE_CHANNEL=main\n")
    data = update_core.check_updates(fetch=True)
    assert data["channel"] == "main" and data["target"] == "main" and data["has_update"]


@pytest.mark.parametrize("username", ["admin", "root", "kvm", "libvirt-qemu"])
def test_build_iso_refuses_names_the_ubuntu_installer_rejects(username):
    """Установщик Ubuntu не принимает системные имена: автоматический образ с таким именем падал на первом шаге."""
    import subprocess

    result = subprocess.run(["bash", str(ROOT / "image" / "build-iso.sh"), "--unattended", "--username", username, "--password-hash", "$6$salt$hash"], capture_output=True, text=True)
    assert result.returncode != 0 and "reserved by the Ubuntu installer" in result.stderr