Claude
34684634af
Модуль web/features/vmops.py (router подключён двумя строками в app.py):
- Увеличение диска на вкладке «Диски и загрузка»: размер из qemu-img info,
только рост, проверка свободного места; выключенная машина — qemu-img resize,
работающая — virsh blockresize. Диски вне /var/lib/virtuality не трогаются.
- Клонирование: страница /vm/{name}/clone с проверками (машина выключена,
хватает места, имя свободно), фоновая операция virt-clone с явными --file
в хранилище образов, прогресс по росту файлов копии, pool-refresh после.
- Пауза, продолжение, спящий режим (managedsave как операция) и удаление
сохранённого состояния — /vm/{name}/ops/{action}; пункты меню на обзоре
и странице машины, кнопка «Продолжить» для приостановленной машины.
- Заметки: название и описание через virsh desc (--new-desc=ТЕКСТ одним
аргументом, чтобы текст вида «--edit» не стал опцией); название видно
под именем машины на обзоре (virsh list --all --title, парсер в presenters).
- Живая нагрузка: /live/stats по virsh domstats для всех активных машин
одним вызовом, CPU % по разнице cpu.time между опросами; шкалы в строках
обзора и карточка «Нагрузка сейчас» на странице машины (panel.js, 5 с).
Тесты для всех парсеров и маршрутов (tests/test_vmops.py), демо-данные
в dev_server.py (три машины в разных состояниях, живые шкалы), значки
notebook-pen и expand в спрайте.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019LEUXMcdZo7b87mjkUsPgg
438 строки
23 KiB
Python
438 строки
23 KiB
Python
"""vmops: увеличение диска, клонирование, пауза и спящий режим, заметки, живая нагрузка."""
|
|
import threading
|
|
from pathlib import Path
|
|
from urllib.parse import unquote
|
|
|
|
import pytest
|
|
|
|
import conftest
|
|
import core
|
|
import presenters
|
|
from features import vmops
|
|
|
|
FIXTURES = Path(__file__).resolve().parent / "fixtures"
|
|
DOMSTATS = (FIXTURES / "domstats.txt").read_text()
|
|
GIB = 1024 ** 3
|
|
|
|
|
|
class Recorder:
|
|
"""Записывает команды; отвечает по overrides {(префикс,): результат|функция}, иначе как fake_run_cmd."""
|
|
|
|
def __init__(self, overrides=None):
|
|
self.calls: list[list[str]] = []
|
|
self.overrides = overrides or {}
|
|
|
|
def __call__(self, cmd, timeout=12, **kwargs):
|
|
self.calls.append(list(cmd))
|
|
for prefix, result in self.overrides.items():
|
|
if tuple(cmd[: len(prefix)]) == prefix:
|
|
return result(cmd) if callable(result) else result
|
|
return conftest.fake_run_cmd(cmd, timeout, **kwargs)
|
|
|
|
def find(self, *prefix):
|
|
return [call for call in self.calls if tuple(call[: len(prefix)]) == prefix]
|
|
|
|
|
|
@pytest.fixture()
|
|
def record(data_dirs, monkeypatch):
|
|
"""Подменяет run_cmd и в core (vmops), и в app (обзор, страница машины)."""
|
|
import app
|
|
|
|
recorder = Recorder()
|
|
for module in (core, app):
|
|
monkeypatch.setattr(module, "run_cmd", recorder)
|
|
monkeypatch.setattr(vmops, "_samples", {})
|
|
return recorder
|
|
|
|
|
|
@pytest.fixture()
|
|
def inline_operations(monkeypatch):
|
|
"""Фоновые операции выполняются сразу, в этом же потоке."""
|
|
|
|
def run_now(operation, worker):
|
|
core.update_operation(operation, status="running")
|
|
worker(operation)
|
|
|
|
monkeypatch.setattr(vmops, "run_operation", run_now)
|
|
|
|
|
|
def shut_off(record):
|
|
record.overrides[("virsh", "domstate")] = conftest.fake_result("shut off")
|
|
|
|
|
|
def location(response) -> str:
|
|
assert response.status_code == 303, response.text[:300]
|
|
return unquote(response.headers["location"])
|
|
|
|
|
|
# ---------------------------------------------------------------- parsers
|
|
def test_parse_domstats_fixture():
|
|
stats = vmops.parse_domstats(DOMSTATS)
|
|
assert list(stats) == ["web01"]
|
|
assert stats["web01"]["cpu.time"] == "48123456789"
|
|
assert stats["web01"]["balloon.rss"] == "1534208"
|
|
assert stats["web01"]["vcpu.current"] == "2"
|
|
|
|
|
|
def test_parse_domstats_several_domains():
|
|
text = DOMSTATS + "Domain: 'db01'\n\nDomain: 'win 11'\n cpu.time=5\n vcpu.current=4\n"
|
|
stats = vmops.parse_domstats(text)
|
|
assert set(stats) == {"web01", "db01", "win 11"}
|
|
assert stats["db01"] == {}
|
|
assert stats["win 11"] == {"cpu.time": "5", "vcpu.current": "4"}
|
|
|
|
|
|
def test_compute_stats_cpu_percent_from_two_samples():
|
|
samples: dict = {}
|
|
first = vmops.compute_stats(vmops.parse_domstats(DOMSTATS), 100.0, samples)
|
|
assert first["web01"]["cpu_pct"] is None # первый замер — сравнивать не с чем
|
|
assert first["web01"]["mem_used_mb"] == 1498 and first["web01"]["mem_total_mb"] == 2048
|
|
later = DOMSTATS.replace("cpu.time=48123456789", "cpu.time=49123456789") # +1 с процессорного времени
|
|
second = vmops.compute_stats(vmops.parse_domstats(later), 101.0, samples)
|
|
assert second["web01"]["cpu_pct"] == 50.0 # 1 с на 2 ядра за 1 с стены
|
|
burst = DOMSTATS.replace("cpu.time=48123456789", "cpu.time=99123456789")
|
|
third = vmops.compute_stats(vmops.parse_domstats(burst), 101.5, samples)
|
|
assert third["web01"]["cpu_pct"] == 100.0 # не больше 100 %
|
|
|
|
|
|
def test_compute_stats_skips_shut_off_and_forgets_stale():
|
|
samples = {"old": (1.0, 5)}
|
|
stats = vmops.compute_stats({"db01": {}, "web01": {"cpu.time": "10", "balloon.current": "1024", "vcpu.current": "1"}}, 2.0, samples)
|
|
assert set(stats) == {"web01"}
|
|
assert stats["web01"]["mem_used_mb"] == 1 and stats["web01"]["mem_total_mb"] == 1
|
|
assert set(samples) == {"web01"}
|
|
|
|
|
|
def test_parse_qemu_img_info():
|
|
info = vmops.parse_qemu_img_info((FIXTURES / "qemu-img-info.json").read_text())
|
|
assert info == {"virtual_size": 42949672960, "actual_size": 6443499520, "format": "qcow2"}
|
|
assert vmops.parse_qemu_img_info("not json") == {}
|
|
assert vmops.parse_qemu_img_info("[1, 2]") == {}
|
|
|
|
|
|
def test_parse_virsh_list_titles():
|
|
titles = presenters.parse_virsh_list_titles((FIXTURES / "list-title.txt").read_text())
|
|
assert titles == {"web01": "Сайт компании", "win11": "Бухгалтерия 1С"}
|
|
assert presenters.parse_virsh_list_titles(" Id Name State\n---\n 1 web01 running") == {}
|
|
assert presenters.parse_virsh_list_titles("") == {}
|
|
|
|
|
|
def test_parse_desc_output():
|
|
assert vmops.parse_desc_output(conftest.fake_result("Сайт компании\n")) == "Сайт компании"
|
|
assert vmops.parse_desc_output(conftest.fake_result("No title for domain: web01")) == ""
|
|
assert vmops.parse_desc_output(conftest.fake_result(ok=False, stderr="error: failed to get domain")) == ""
|
|
|
|
|
|
def test_has_managed_save():
|
|
assert vmops.has_managed_save(conftest.DOMINFO + "Managed save: yes\n")
|
|
assert not vmops.has_managed_save(conftest.DOMINFO + "Managed save: no\n")
|
|
assert not vmops.has_managed_save(conftest.DOMINFO)
|
|
|
|
|
|
def test_clean_notes():
|
|
assert vmops.clean_title(" Сайт \n компании ") == "Сайт компании"
|
|
assert len(vmops.clean_title("x" * 200)) == vmops.TITLE_MAX
|
|
assert vmops.clean_description("строка 1\r\nстрока 2 \r\n\r\n") == "строка 1\nстрока 2"
|
|
|
|
|
|
def test_clone_names_and_targets(data_dirs):
|
|
assert vmops.suggest_clone_name("web01", {"web01"}) == "web01-copy"
|
|
assert vmops.suggest_clone_name("web01", {"web01", "web01-copy", "web01-copy2"}) == "web01-copy3"
|
|
disks = [
|
|
{"device": "disk", "source": "/var/lib/virtuality/images/web01.qcow2"},
|
|
{"device": "cdrom", "source": ""},
|
|
{"device": "disk", "source": "/var/lib/virtuality/images/data.img"},
|
|
]
|
|
targets = vmops.clone_targets("web02", disks)
|
|
assert targets == [data_dirs["images"] / "web02.qcow2", data_dirs["images"] / "web02-disk2.img"]
|
|
|
|
|
|
def test_explain_error():
|
|
assert "снимки" in vmops.explain_error("qemu-img: Can't resize an image which has snapshots", "x")
|
|
assert vmops.explain_error("", "запасной текст") == "запасной текст"
|
|
|
|
|
|
# ---------------------------------------------------------------- live stats
|
|
def test_live_stats_requires_auth(client):
|
|
assert client.get("/live/stats").status_code == 401
|
|
|
|
|
|
def test_live_stats_json(logged_in, record):
|
|
first = logged_in.get("/live/stats").json()
|
|
assert first["ok"] is True
|
|
assert first["stats"]["web01"]["cpu_pct"] is None
|
|
assert first["stats"]["web01"]["mem_used_mb"] == 1498 and first["stats"]["web01"]["mem_total_mb"] == 2048
|
|
second = logged_in.get("/live/stats").json()
|
|
assert second["stats"]["web01"]["cpu_pct"] == 0.0 # cpu.time в подделке не растёт
|
|
assert record.find("virsh", "domstats") == [["virsh", "domstats", "--cpu-total", "--balloon", "--vcpu", "web01"]] * 2 # только активные машины, один вызов на запрос
|
|
|
|
|
|
def test_live_stats_without_active_vms(logged_in, record):
|
|
record.overrides[("virsh", "list")] = conftest.fake_result("")
|
|
assert logged_in.get("/live/stats").json()["stats"] == {}
|
|
assert record.find("virsh", "domstats") == []
|
|
|
|
|
|
# ---------------------------------------------------------------- pages
|
|
def test_dashboard_shows_title_and_meters(logged_in, record):
|
|
html = logged_in.get("/").text
|
|
assert "Сайт компании" in html
|
|
assert 'data-live-stats="web01"' in html and 'data-live-stats="db01"' in html
|
|
assert "/vm/web01/ops/pause" in html and "/vm/web01/ops/hibernate" in html
|
|
assert "/vm/web01/clone" in html and "/vm/db01/clone" in html
|
|
assert record.find("virsh", "list", "--all", "--title")
|
|
|
|
|
|
def test_dashboard_paused_vm_offers_resume(logged_in, record):
|
|
record.overrides[("virsh", "list")] = conftest.fake_result(" Id Name State\n-----------------------\n 2 web01 paused\n - db01 shut off")
|
|
html = logged_in.get("/").text
|
|
assert 'action="/vm/web01/ops/resume"' in html
|
|
assert 'action="/vm/web01/start"' not in html
|
|
assert 'action="/vm/db01/start"' in html
|
|
|
|
|
|
def test_vm_page_shows_disks_notes_and_ops(logged_in, record):
|
|
html = logged_in.get("/vm/web01").text
|
|
assert "40 ГБ" in html and "занято 6.0 ГБ" in html
|
|
assert 'action="/vm/web01/disks/vda/resize"' in html and 'min="41"' in html
|
|
assert 'action="/vm/web01/disks/sda/resize"' not in html # привод не увеличивают
|
|
assert 'value="Сайт компании"' in html and "Nginx + база. Бэкап по пятницам." in html
|
|
assert "Приостановить" in html and "Спящий режим" in html and "Клонировать" in html
|
|
assert 'data-live-stats="web01"' in html # карточка «Нагрузка сейчас»
|
|
assert record.find("qemu-img", "info") == [["qemu-img", "info", "--output=json", "-U", str(core.IMAGES_DIR / "web01.qcow2")]]
|
|
|
|
|
|
def test_vm_page_paused_and_saved_states(logged_in, record):
|
|
record.overrides[("virsh", "dominfo")] = conftest.fake_result(conftest.DOMINFO.replace("State: running", "State: paused"))
|
|
html = logged_in.get("/vm/web01").text
|
|
assert 'action="/vm/web01/ops/resume"' in html and "Продолжить" in html
|
|
record.overrides[("virsh", "dominfo")] = conftest.fake_result(conftest.DOMINFO.replace("State: running", "State: shut off") + "Managed save: yes\n")
|
|
html = logged_in.get("/vm/web01").text
|
|
assert "Машина в спящем режиме" in html and 'action="/vm/web01/ops/forget-save"' in html
|
|
assert 'data-live-stats="web01"' not in html # выключенная машина — нагрузки нет
|
|
|
|
|
|
def test_vm_page_messages(logged_in, record):
|
|
html = logged_in.get("/vm/web01?disk_message=Диск+увеличен¬es_error=Не+вышло").text
|
|
assert "Диск увеличен" in html and "Не вышло" in html
|
|
|
|
|
|
# ---------------------------------------------------------------- disk resize
|
|
def test_disk_resize_shut_off_uses_qemu_img(logged_in, record):
|
|
shut_off(record)
|
|
response = logged_in.post("/vm/web01/disks/vda/resize", data={"size_gb": "50"}, follow_redirects=False)
|
|
assert location(response).startswith("/vm/web01?disk_message=Диск web01.qcow2 увеличен до 50 ГБ")
|
|
assert location(response).endswith("#storage")
|
|
assert record.find("qemu-img", "resize") == [["qemu-img", "resize", str(core.IMAGES_DIR / "web01.qcow2"), "50G"]]
|
|
assert record.find("virsh", "blockresize") == []
|
|
|
|
|
|
def test_disk_resize_running_uses_blockresize(logged_in, record):
|
|
response = logged_in.post("/vm/web01/disks/vda/resize", data={"size_gb": "41"}, follow_redirects=False)
|
|
assert "disk_message" in location(response)
|
|
assert record.find("virsh", "blockresize") == [["virsh", "blockresize", "web01", "vda", "41G"]]
|
|
assert record.find("qemu-img", "resize") == []
|
|
|
|
|
|
def test_disk_resize_only_grows(logged_in, record):
|
|
response = logged_in.post("/vm/web01/disks/vda/resize", data={"size_gb": "40"}, follow_redirects=False)
|
|
assert "не меньше 41 ГБ" in location(response)
|
|
response = logged_in.post("/vm/web01/disks/vda/resize", data={"size_gb": "999999"}, follow_redirects=False)
|
|
assert "Слишком большой" in location(response)
|
|
assert record.find("virsh", "blockresize") == [] and record.find("qemu-img", "resize") == []
|
|
|
|
|
|
def test_disk_resize_rejects_unknown_or_cdrom(logged_in, record):
|
|
assert "Такого диска" in location(logged_in.post("/vm/web01/disks/vdz/resize", data={"size_gb": "50"}, follow_redirects=False))
|
|
assert "нельзя увеличить" in location(logged_in.post("/vm/web01/disks/sda/resize", data={"size_gb": "50"}, follow_redirects=False))
|
|
assert logged_in.post("/vm/web01/disks/../etc/resize", data={"size_gb": "50"}, follow_redirects=False).status_code in (400, 404)
|
|
assert logged_in.post("/vm/-bad/disks/vda/resize", data={"size_gb": "50"}, follow_redirects=False).status_code == 400
|
|
|
|
|
|
def test_disk_resize_outside_storage_refused(logged_in, record):
|
|
record.overrides[("virsh", "domblklist")] = conftest.fake_result((FIXTURES / "domblklist-details.txt").read_text().replace("/var/lib/virtuality/images", "/root"))
|
|
assert "нельзя увеличить" in location(logged_in.post("/vm/web01/disks/vda/resize", data={"size_gb": "50"}, follow_redirects=False))
|
|
|
|
|
|
def test_disk_resize_checks_free_space(logged_in, record, monkeypatch):
|
|
monkeypatch.setattr(vmops, "free_space", lambda path: 2 * GIB)
|
|
response = logged_in.post("/vm/web01/disks/vda/resize", data={"size_gb": "50"}, follow_redirects=False)
|
|
assert "свободно только 2.0 ГБ" in location(response)
|
|
assert record.find("virsh", "blockresize") == []
|
|
|
|
|
|
def test_disk_resize_reports_tool_error(logged_in, record):
|
|
record.overrides[("virsh", "blockresize")] = conftest.fake_result(ok=False, stderr="error: internal error: Can't resize an image which has snapshots")
|
|
assert "удалите снимки" in location(logged_in.post("/vm/web01/disks/vda/resize", data={"size_gb": "50"}, follow_redirects=False))
|
|
|
|
|
|
# ---------------------------------------------------------------- pause / resume / hibernate
|
|
def test_ops_pause_and_resume(logged_in, record):
|
|
response = logged_in.post("/vm/web01/ops/pause", data={"next": "/"}, follow_redirects=False)
|
|
assert location(response).startswith("/?ops_message=Машина web01 приостановлена")
|
|
assert record.find("virsh", "suspend") == [["virsh", "suspend", "web01"]]
|
|
response = logged_in.post("/vm/web01/ops/resume", follow_redirects=False)
|
|
assert "ops_error=Продолжить можно только приостановленную" in location(response)
|
|
record.overrides[("virsh", "domstate")] = conftest.fake_result("paused")
|
|
response = logged_in.post("/vm/web01/ops/resume", data={"next": "http://evil.example/"}, follow_redirects=False)
|
|
assert location(response).startswith("/vm/web01?ops_message=") # чужой next не используется
|
|
assert record.find("virsh", "resume") == [["virsh", "resume", "web01"]]
|
|
|
|
|
|
def test_ops_validation(logged_in, record):
|
|
assert logged_in.post("/vm/web01/ops/explode", follow_redirects=False).status_code == 400
|
|
assert logged_in.post("/vm/-bad/ops/pause", follow_redirects=False).status_code == 400
|
|
assert location(logged_in.post("/vm/ghost/ops/pause", follow_redirects=False)) == "/"
|
|
shut_off(record)
|
|
assert "ops_error=Приостановить можно только работающую" in location(logged_in.post("/vm/web01/ops/pause", follow_redirects=False))
|
|
record.overrides[("virsh", "suspend")] = conftest.fake_result(ok=False, stderr="error: Timed out during operation: cannot acquire state change lock")
|
|
record.overrides[("virsh", "domstate")] = conftest.fake_result("running")
|
|
assert "занята другой операцией" in location(logged_in.post("/vm/web01/ops/pause", follow_redirects=False))
|
|
|
|
|
|
def test_ops_requires_login(client):
|
|
assert client.post("/vm/web01/ops/pause", follow_redirects=False).headers["location"] == "/login"
|
|
|
|
|
|
def test_hibernate_runs_operation(logged_in, record, inline_operations, data_dirs):
|
|
response = logged_in.post("/vm/web01/ops/hibernate", data={"next": "/"}, follow_redirects=False)
|
|
assert location(response).startswith("/?ops_message=Машина web01 засыпает")
|
|
assert record.find("virsh", "managedsave") == [["virsh", "managedsave", "web01"]]
|
|
operations = core.list_operations(5)
|
|
assert operations[0]["type"] == "vm_hibernate" and operations[0]["status"] == "success" and operations[0]["vm_name"] == "web01"
|
|
|
|
|
|
def test_hibernate_blocked(logged_in, record, monkeypatch):
|
|
shut_off(record)
|
|
assert "ops_error=В спящий режим можно" in location(logged_in.post("/vm/web01/ops/hibernate", follow_redirects=False))
|
|
record.overrides[("virsh", "domstate")] = conftest.fake_result("running")
|
|
monkeypatch.setattr(vmops, "free_space", lambda path: 0)
|
|
assert "свободного места" in location(logged_in.post("/vm/web01/ops/hibernate", follow_redirects=False))
|
|
assert record.find("virsh", "managedsave") == []
|
|
|
|
|
|
def test_hibernate_failure_is_reported(logged_in, record, inline_operations):
|
|
record.overrides[("virsh", "managedsave")] = conftest.fake_result(ok=False, stderr="error: Requested operation is not valid: domain is not running")
|
|
logged_in.post("/vm/web01/ops/hibernate", follow_redirects=False)
|
|
operation = core.list_operations(1)[0]
|
|
assert operation["status"] == "error" and "выключена" in operation["message"]
|
|
|
|
|
|
def test_forget_saved_state(logged_in, record):
|
|
shut_off(record)
|
|
response = logged_in.post("/vm/web01/ops/forget-save", follow_redirects=False)
|
|
assert "ops_message=Сохранённое состояние удалено" in location(response)
|
|
assert record.find("virsh", "managedsave-remove") == [["virsh", "managedsave-remove", "web01"]]
|
|
|
|
|
|
# ---------------------------------------------------------------- notes
|
|
def test_notes_save_running(logged_in, record):
|
|
response = logged_in.post("/vm/web01/notes/save", data={"title": " Сайт компании ", "description": "строка 1\r\nстрока 2"}, follow_redirects=False)
|
|
assert location(response) == "/vm/web01?notes_message=Заметки сохранены"
|
|
assert record.find("virsh", "desc") == [
|
|
["virsh", "desc", "web01", "--title", "--new-desc=Сайт компании", "--config", "--live"],
|
|
["virsh", "desc", "web01", "--new-desc=строка 1\nстрока 2", "--config", "--live"],
|
|
]
|
|
|
|
|
|
def test_notes_save_shut_off_and_dashes(logged_in, record):
|
|
shut_off(record)
|
|
logged_in.post("/vm/web01/notes/save", data={"title": "--edit", "description": ""}, follow_redirects=False)
|
|
calls = record.find("virsh", "desc")
|
|
assert calls[0] == ["virsh", "desc", "web01", "--title", "--new-desc=--edit", "--config"] # текст одним аргументом, без --live
|
|
assert calls[1] == ["virsh", "desc", "web01", "--new-desc=", "--config"] # пустое описание очищает
|
|
|
|
|
|
def test_notes_save_error(logged_in, record):
|
|
record.overrides[("virsh", "desc")] = conftest.fake_result(ok=False, stderr="error: failed to get domain 'web01'")
|
|
assert "notes_error=" in location(logged_in.post("/vm/web01/notes/save", data={"title": "x"}, follow_redirects=False))
|
|
assert logged_in.post("/vm/-bad/notes/save", data={"title": "x"}, follow_redirects=False).status_code == 400
|
|
|
|
|
|
# ---------------------------------------------------------------- clone
|
|
def test_clone_page_blocked_while_running(logged_in, record):
|
|
response = logged_in.get("/vm/web01/clone")
|
|
assert response.status_code == 200
|
|
assert "Сначала выключите машину" in response.text
|
|
assert 'action="/vm/web01/shutdown"' in response.text
|
|
assert "disabled" in response.text
|
|
|
|
|
|
def test_clone_page_ready(logged_in, record):
|
|
shut_off(record)
|
|
html = logged_in.get("/vm/web01/clone").text
|
|
assert "Сначала выключите" not in html
|
|
assert 'value="web01-copy"' in html
|
|
assert "6.0 ГБ" in html # нужно места — реальный размер диска
|
|
assert "web01.qcow2" in html
|
|
|
|
|
|
def test_clone_validation(logged_in, record):
|
|
shut_off(record)
|
|
assert "Имя копии" in location(logged_in.post("/vm/web01/clone/start", data={"new_name": "bad name!"}, follow_redirects=False))
|
|
assert "уже есть" in location(logged_in.post("/vm/web01/clone/start", data={"new_name": "db01"}, follow_redirects=False))
|
|
assert "уже есть" in location(logged_in.post("/vm/web01/clone/start", data={"new_name": "WEB01"}, follow_redirects=False))
|
|
record.overrides[("virsh", "domstate")] = conftest.fake_result("running")
|
|
assert "Сначала выключите" in location(logged_in.post("/vm/web01/clone/start", data={"new_name": "web02"}, follow_redirects=False))
|
|
assert record.find("virt-clone") == []
|
|
assert logged_in.post("/vm/-bad/clone/start", data={"new_name": "web02"}, follow_redirects=False).status_code == 400
|
|
|
|
|
|
def test_clone_checks_space_and_existing_file(logged_in, record, data_dirs, monkeypatch):
|
|
shut_off(record)
|
|
(data_dirs["images"] / "web02.qcow2").write_bytes(b"x")
|
|
assert "уже есть в хранилище" in location(logged_in.post("/vm/web01/clone/start", data={"new_name": "web02"}, follow_redirects=False))
|
|
monkeypatch.setattr(vmops, "free_space", lambda path: GIB)
|
|
assert "Не хватает места" in location(logged_in.post("/vm/web01/clone/start", data={"new_name": "web03"}, follow_redirects=False))
|
|
|
|
|
|
def test_clone_starts_operation(logged_in, record, inline_operations, data_dirs):
|
|
shut_off(record)
|
|
response = logged_in.post("/vm/web01/clone/start", data={"new_name": "web02"}, follow_redirects=False)
|
|
assert location(response).startswith("/operations/")
|
|
operation = core.read_operation(location(response).split("/")[-1])
|
|
assert operation["status"] == "success" and operation["type"] == "vm_clone"
|
|
assert operation["vm_name"] == "web02" and operation["source_vm"] == "web01"
|
|
assert "web02 создана" in operation["message"]
|
|
assert record.find("virt-clone") == [["virt-clone", "--original", "web01", "--name", "web02", "--file", str(data_dirs["images"] / "web02.qcow2")]]
|
|
assert record.find("virsh", "pool-refresh") == [["virsh", "pool-refresh", "virtuality-images"]]
|
|
assert "virt-clone --original web01" in operation["log_tail"]
|
|
|
|
|
|
def test_clone_failure_is_reported(logged_in, record, inline_operations):
|
|
shut_off(record)
|
|
record.overrides[("virt-clone",)] = conftest.fake_result(ok=False, stderr="ERROR Storage volume already exists")
|
|
response = logged_in.post("/vm/web01/clone/start", data={"new_name": "web02"}, follow_redirects=False)
|
|
operation = core.read_operation(location(response).split("/")[-1])
|
|
assert operation["status"] == "error" and "уже есть" in operation["message"]
|
|
assert record.find("virsh", "pool-refresh") == []
|
|
|
|
|
|
def test_clone_progress_watcher(data_dirs):
|
|
target = data_dirs["images"] / "copy.qcow2"
|
|
target.write_bytes(b"\0" * (512 * 1024))
|
|
operation = core.new_operation("vm_clone", "test", vm_name="copy", source_vm="web01")
|
|
|
|
class OneTick:
|
|
def __init__(self):
|
|
self.ticks = 0
|
|
|
|
def wait(self, timeout):
|
|
self.ticks += 1
|
|
return self.ticks > 1
|
|
|
|
vmops._watch_clone_progress(operation, [target], 1024 * 1024, OneTick())
|
|
fresh = core.read_operation(operation["id"])
|
|
assert 3 <= fresh["progress"] <= 95 and "Копируем диски" in fresh["message"]
|
|
assert vmops._allocated_bytes(data_dirs["images"] / "missing.qcow2") == 0
|
|
|
|
|
|
def test_clone_worker_stops_watcher(data_dirs, monkeypatch):
|
|
"""Даже если virt-clone упал, поток прогресса останавливается."""
|
|
monkeypatch.setattr(core, "run_cmd", lambda cmd, timeout=12: conftest.fake_result(ok=False, stderr="boom"))
|
|
operation = core.new_operation("vm_clone", "test", vm_name="web02", source_vm="web01", targets=[str(data_dirs["images"] / "web02.qcow2")], expected_bytes=10)
|
|
before = threading.active_count()
|
|
vmops.clone_worker(operation)
|
|
assert threading.active_count() <= before
|
|
assert core.read_operation(operation["id"])["status"] == "error"
|
© 2026 StackWorks