667 lines
22 KiB
Python
667 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import time
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from textual.app import App, ComposeResult
|
|
from textual.containers import ScrollableContainer
|
|
from textual.widgets import Button, Checkbox, Input, Select, Static
|
|
|
|
from pve_vm_setup.app import PveVmSetupApp
|
|
from pve_vm_setup.models.workflow import WorkflowState
|
|
from pve_vm_setup.screens.env_setup import EnvSetupModal, MissingEnvSetupPromptModal
|
|
from pve_vm_setup.screens.login import LoginView
|
|
from pve_vm_setup.screens.wizard import NO_DISK_SELECTED, AutoStartConfirmModal, WizardView
|
|
from pve_vm_setup.services.factory import ProxmoxServiceFactory
|
|
from pve_vm_setup.services.fake import FakeProxmoxService
|
|
from pve_vm_setup.settings import AppSettings
|
|
|
|
|
|
class LoginHarnessApp(App[None]):
|
|
def compose(self) -> ComposeResult:
|
|
yield LoginView(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
WorkflowState(),
|
|
FakeProxmoxService(),
|
|
)
|
|
|
|
|
|
async def wait_for_wizard_ready(
|
|
pilot,
|
|
app: App[None],
|
|
*,
|
|
attempts: int = 12,
|
|
delay: float = 0.1,
|
|
) -> None:
|
|
for _ in range(attempts):
|
|
await pilot.pause(delay)
|
|
if (
|
|
app.query_one("#general-vmid", Input).value == "123"
|
|
and app.query_one("#general-node", Select).value == "fake-node-01"
|
|
and app.query_one("#os-storage", Select).value == "cephfs"
|
|
):
|
|
return
|
|
raise AssertionError("Timed out waiting for wizard reference data to load.")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_view_authenticates_with_pilot() -> None:
|
|
app = LoginHarnessApp()
|
|
|
|
async with app.run_test() as pilot:
|
|
await pilot.pause()
|
|
assert str(app.query_one("#title", Static).renderable) == "Proxmox Login"
|
|
assert app.focused is app.query_one("#username", Input)
|
|
|
|
app.query_one("#username", Input).value = "junior"
|
|
app.query_one("#password", Input).value = "secret"
|
|
app.query_one("#connect", Button).press()
|
|
await pilot.pause()
|
|
|
|
assert "Authenticated as junior@pam." == str(app.query_one("#status", Static).renderable)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_main_app_mounts_wizard_only_after_login() -> None:
|
|
service = FakeProxmoxService()
|
|
app = PveVmSetupApp(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
service=service,
|
|
)
|
|
|
|
async with app.run_test() as pilot:
|
|
await pilot.pause()
|
|
assert app.query(LoginView)
|
|
assert not app.query(WizardView)
|
|
|
|
login = app.query_one(LoginView)
|
|
login.post_message(LoginView.Authenticated("junior@pam", "pam"))
|
|
await pilot.pause()
|
|
await pilot.pause()
|
|
await wait_for_wizard_ready(pilot, app)
|
|
|
|
assert not app.query(LoginView)
|
|
assert app.query(WizardView)
|
|
assert app.focused is app.query_one("#general-name", Input)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_main_app_prompts_for_env_setup_when_no_dotenv_exists(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
home = tmp_path / "home"
|
|
monkeypatch.chdir(tmp_path)
|
|
monkeypatch.setenv("HOME", str(home))
|
|
|
|
app = PveVmSetupApp(
|
|
AppSettings.from_env({}, load_dotenv_file=True),
|
|
prompt_for_missing_env_setup=True,
|
|
)
|
|
|
|
async with app.run_test() as pilot:
|
|
await pilot.pause()
|
|
|
|
assert isinstance(app.screen_stack[-1], MissingEnvSetupPromptModal)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_main_app_can_save_env_setup_and_refresh_login_mode(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
class LiveLikeService(FakeProxmoxService):
|
|
mode = "live"
|
|
|
|
home = tmp_path / "home"
|
|
config_path = home / ".config" / "pve-vm-setup" / ".env"
|
|
monkeypatch.chdir(tmp_path)
|
|
monkeypatch.setenv("HOME", str(home))
|
|
monkeypatch.delenv("PROXMOX_URL", raising=False)
|
|
monkeypatch.delenv("PROXMOX_USER", raising=False)
|
|
|
|
def create_service(settings: AppSettings):
|
|
if settings.proxmox_url:
|
|
return LiveLikeService()
|
|
return FakeProxmoxService()
|
|
|
|
monkeypatch.setattr(ProxmoxServiceFactory, "create", staticmethod(create_service))
|
|
|
|
app = PveVmSetupApp(
|
|
AppSettings.from_env({}, load_dotenv_file=True),
|
|
prompt_for_missing_env_setup=True,
|
|
)
|
|
|
|
async with app.run_test() as pilot:
|
|
await pilot.pause()
|
|
assert isinstance(app.screen_stack[-1], MissingEnvSetupPromptModal)
|
|
|
|
app.query_one("#missing-env-setup", Button).press()
|
|
await pilot.pause()
|
|
assert isinstance(app.screen_stack[-1], EnvSetupModal)
|
|
|
|
app.query_one("#env-proxmox-url", Input).value = "https://pve.example.invalid:8006"
|
|
app.query_one("#env-proxmox-user", Input).value = "root"
|
|
app.query_one("#env-setup-save", Button).press()
|
|
|
|
for _ in range(10):
|
|
await pilot.pause(0.05)
|
|
if not app.screen_stack or not isinstance(app.screen_stack[-1], EnvSetupModal):
|
|
break
|
|
|
|
assert config_path.exists()
|
|
content = config_path.read_text(encoding="utf-8")
|
|
assert "PROXMOX_URL=https://pve.example.invalid:8006" in content
|
|
assert "PROXMOX_USER=root" in content
|
|
assert "PROXMOX_PASSWORD" not in content
|
|
|
|
assert "Mode: live on pve.example.invalid:8006" == str(
|
|
app.query_one("#mode", Static).renderable
|
|
)
|
|
assert app.query_one("#username", Input).value == "root"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wizard_activation_focuses_first_editable_field() -> None:
|
|
service = FakeProxmoxService()
|
|
app = PveVmSetupApp(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
service=service,
|
|
)
|
|
|
|
async with app.run_test() as pilot:
|
|
app.query_one(LoginView).remove()
|
|
wizard = WizardView(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
WorkflowState(),
|
|
service,
|
|
)
|
|
await app.query_one("#app-body").mount(wizard)
|
|
wizard.activate()
|
|
await wait_for_wizard_ready(pilot, app)
|
|
|
|
assert app.focused is app.query_one("#general-name", Input)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wizard_initial_activation_does_not_duplicate_live_reference_loads() -> None:
|
|
class CountingService(FakeProxmoxService):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.calls: list[str] = []
|
|
|
|
def load_nodes(self):
|
|
self.calls.append("load_nodes")
|
|
return super().load_nodes()
|
|
|
|
def load_pools(self):
|
|
self.calls.append("load_pools")
|
|
return super().load_pools()
|
|
|
|
def load_existing_tags(self):
|
|
self.calls.append("load_existing_tags")
|
|
return super().load_existing_tags()
|
|
|
|
def load_next_vmid(self):
|
|
self.calls.append("load_next_vmid")
|
|
return super().load_next_vmid()
|
|
|
|
def load_storages(self, node: str):
|
|
self.calls.append(f"load_storages:{node}")
|
|
return super().load_storages(node)
|
|
|
|
def load_bridges(self, node: str):
|
|
self.calls.append(f"load_bridges:{node}")
|
|
return super().load_bridges(node)
|
|
|
|
def load_isos(self, node: str, storage: str):
|
|
self.calls.append(f"load_isos:{node}:{storage}")
|
|
return super().load_isos(node, storage)
|
|
|
|
service = CountingService()
|
|
app = PveVmSetupApp(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
service=service,
|
|
)
|
|
|
|
async with app.run_test() as pilot:
|
|
app.query_one(LoginView).remove()
|
|
wizard = WizardView(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
WorkflowState(),
|
|
service,
|
|
)
|
|
await app.query_one("#app-body").mount(wizard)
|
|
wizard.activate()
|
|
for _ in range(6):
|
|
await pilot.pause()
|
|
|
|
assert Counter(service.calls) == Counter(
|
|
[
|
|
"load_nodes",
|
|
"load_pools",
|
|
"load_existing_tags",
|
|
"load_next_vmid",
|
|
"load_storages:fake-node-01",
|
|
"load_bridges:fake-node-01",
|
|
"load_isos:fake-node-01:cephfs",
|
|
]
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wizard_initial_activation_loads_reference_data_concurrently() -> None:
|
|
class SlowService(FakeProxmoxService):
|
|
delay = 0.15
|
|
|
|
def load_nodes(self):
|
|
time.sleep(self.delay)
|
|
return super().load_nodes()
|
|
|
|
def load_pools(self):
|
|
time.sleep(self.delay)
|
|
return super().load_pools()
|
|
|
|
def load_existing_tags(self):
|
|
time.sleep(self.delay)
|
|
return super().load_existing_tags()
|
|
|
|
def load_next_vmid(self):
|
|
time.sleep(self.delay)
|
|
return super().load_next_vmid()
|
|
|
|
def load_storages(self, node: str):
|
|
time.sleep(self.delay)
|
|
return super().load_storages(node)
|
|
|
|
def load_bridges(self, node: str):
|
|
time.sleep(self.delay)
|
|
return super().load_bridges(node)
|
|
|
|
def load_isos(self, node: str, storage: str):
|
|
time.sleep(self.delay)
|
|
return super().load_isos(node, storage)
|
|
|
|
service = SlowService()
|
|
app = PveVmSetupApp(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
service=service,
|
|
)
|
|
|
|
async with app.run_test() as pilot:
|
|
app.query_one(LoginView).remove()
|
|
wizard = WizardView(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
WorkflowState(),
|
|
service,
|
|
)
|
|
await app.query_one("#app-body").mount(wizard)
|
|
|
|
started_at = time.perf_counter()
|
|
wizard.activate()
|
|
|
|
await wait_for_wizard_ready(pilot, app)
|
|
|
|
elapsed = time.perf_counter() - started_at
|
|
|
|
assert elapsed < 1.0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wizard_uses_scrollable_sections_with_border_titles() -> None:
|
|
service = FakeProxmoxService()
|
|
app = PveVmSetupApp(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
service=service,
|
|
)
|
|
|
|
async with app.run_test() as pilot:
|
|
app.query_one(LoginView).remove()
|
|
wizard = WizardView(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
WorkflowState(),
|
|
service,
|
|
)
|
|
await app.query_one("#app-body").mount(wizard)
|
|
wizard.activate()
|
|
await wait_for_wizard_ready(pilot, app)
|
|
|
|
general_section = app.query_one("#general-section", ScrollableContainer)
|
|
os_section = app.query_one("#os-section", ScrollableContainer)
|
|
|
|
assert str(general_section.border_title).strip() == "General"
|
|
assert str(os_section.border_title).strip() == "Operating System"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wizard_hides_os_fields_based_on_media_choice() -> None:
|
|
service = FakeProxmoxService()
|
|
app = PveVmSetupApp(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
service=service,
|
|
)
|
|
|
|
async with app.run_test() as pilot:
|
|
app.query_one(LoginView).remove()
|
|
wizard = WizardView(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
WorkflowState(),
|
|
service,
|
|
)
|
|
await app.query_one("#app-body").mount(wizard)
|
|
wizard.activate()
|
|
await wait_for_wizard_ready(pilot, app)
|
|
|
|
assert app.query_one("#os-storage", Select).display is True
|
|
assert app.query_one("#os-iso", Select).display is True
|
|
assert app.query_one("#os-physical-drive", Input).display is False
|
|
|
|
app.query_one("#os-media-choice", Select).value = "physical"
|
|
await pilot.pause()
|
|
|
|
assert app.query_one("#os-storage", Select).display is False
|
|
assert app.query_one("#os-iso", Select).display is False
|
|
assert app.query_one("#os-physical-drive", Input).display is True
|
|
|
|
app.query_one("#os-media-choice", Select).value = "none"
|
|
await pilot.pause()
|
|
|
|
assert app.query_one("#os-storage", Select).display is False
|
|
assert app.query_one("#os-iso", Select).display is False
|
|
assert app.query_one("#os-physical-drive", Input).display is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wizard_hides_dependent_system_memory_and_network_fields() -> None:
|
|
service = FakeProxmoxService()
|
|
app = PveVmSetupApp(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
service=service,
|
|
)
|
|
|
|
async with app.run_test() as pilot:
|
|
app.query_one(LoginView).remove()
|
|
wizard = WizardView(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
WorkflowState(),
|
|
service,
|
|
)
|
|
await app.query_one("#app-body").mount(wizard)
|
|
wizard.activate()
|
|
await wait_for_wizard_ready(pilot, app)
|
|
|
|
assert app.query_one("#system-efi-storage", Select).display is True
|
|
assert app.query_one("#system-pre-enroll", Checkbox).display is True
|
|
|
|
app.query_one("#system-add-efi", Checkbox).value = False
|
|
await pilot.pause()
|
|
|
|
assert app.query_one("#system-efi-storage", Select).display is False
|
|
assert app.query_one("#system-pre-enroll", Checkbox).display is False
|
|
|
|
app.query_one("#system-tpm", Checkbox).value = True
|
|
await pilot.pause()
|
|
|
|
assert app.query_one("#system-efi-storage", Select).display is True
|
|
assert app.query_one("#system-pre-enroll", Checkbox).display is False
|
|
|
|
assert app.query_one("#memory-min-size", Input).display is True
|
|
assert app.query_one("#memory-ksm", Checkbox).display is True
|
|
|
|
app.query_one("#memory-ballooning", Checkbox).value = False
|
|
await pilot.pause()
|
|
|
|
assert app.query_one("#memory-min-size", Input).display is False
|
|
assert app.query_one("#memory-ksm", Checkbox).display is False
|
|
|
|
assert app.query_one("#network-bridge", Select).display is True
|
|
assert app.query_one("#network-rate", Input).display is True
|
|
|
|
app.query_one("#network-none", Checkbox).value = True
|
|
await pilot.pause()
|
|
|
|
assert app.query_one("#network-bridge", Select).display is False
|
|
assert app.query_one("#network-rate", Input).display is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wizard_tag_rows_keep_input_and_button_visible() -> None:
|
|
service = FakeProxmoxService()
|
|
app = PveVmSetupApp(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
service=service,
|
|
)
|
|
|
|
async with app.run_test() as pilot:
|
|
app.query_one(LoginView).remove()
|
|
wizard = WizardView(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
WorkflowState(),
|
|
service,
|
|
)
|
|
await app.query_one("#app-body").mount(wizard)
|
|
wizard.activate()
|
|
await wait_for_wizard_ready(pilot, app)
|
|
|
|
assert app.query_one("#general-tag-input", Input).display is True
|
|
assert app.query_one("#general-tag-add", Button).display is True
|
|
assert app.query_one("#general-tag-existing", Select).display is True
|
|
assert app.query_one("#general-tag-use", Button).display is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wizard_add_tag_button_updates_current_tags() -> None:
|
|
service = FakeProxmoxService()
|
|
app = PveVmSetupApp(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
service=service,
|
|
)
|
|
|
|
async with app.run_test() as pilot:
|
|
app.query_one(LoginView).remove()
|
|
wizard = WizardView(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
WorkflowState(),
|
|
service,
|
|
)
|
|
await app.query_one("#app-body").mount(wizard)
|
|
wizard.activate()
|
|
await wait_for_wizard_ready(pilot, app)
|
|
|
|
app.query_one("#general-tag-input", Input).value = "alpha"
|
|
app.query_one("#general-tag-add", Button).press()
|
|
await pilot.pause()
|
|
|
|
assert wizard._workflow.config.general.tags == ["alpha"]
|
|
current_tags = app.query_one("#general-tag-current", Select)
|
|
assert current_tags.display is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wizard_hiding_select_collapses_open_overlay() -> None:
|
|
service = FakeProxmoxService()
|
|
app = PveVmSetupApp(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
service=service,
|
|
)
|
|
|
|
async with app.run_test() as pilot:
|
|
app.query_one(LoginView).remove()
|
|
wizard = WizardView(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
WorkflowState(),
|
|
service,
|
|
)
|
|
await app.query_one("#app-body").mount(wizard)
|
|
wizard.activate()
|
|
await wait_for_wizard_ready(pilot, app)
|
|
|
|
storage = app.query_one("#os-storage", Select)
|
|
storage.expanded = True
|
|
await pilot.pause()
|
|
assert storage.expanded is True
|
|
|
|
app.query_one("#os-media-choice", Select).value = "physical"
|
|
await pilot.pause()
|
|
|
|
assert storage.display is False
|
|
assert storage.expanded is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disk_toolbar_buttons_render_left_of_disk_selector() -> None:
|
|
service = FakeProxmoxService()
|
|
app = PveVmSetupApp(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
service=service,
|
|
)
|
|
|
|
async with app.run_test() as pilot:
|
|
app.query_one(LoginView).remove()
|
|
wizard = WizardView(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
WorkflowState(),
|
|
service,
|
|
)
|
|
await app.query_one("#app-body").mount(wizard)
|
|
wizard.activate()
|
|
await wait_for_wizard_ready(pilot, app)
|
|
|
|
wizard._workflow.current_step_index = 3
|
|
wizard._show_step()
|
|
await pilot.pause()
|
|
|
|
add_button = app.query_one("#disks-add", Button)
|
|
remove_button = app.query_one("#disks-remove", Button)
|
|
selector = app.query_one("#disks-select", Select)
|
|
|
|
assert add_button.region.x < remove_button.region.x
|
|
assert selector.region.x == add_button.region.x
|
|
assert selector.region.width > remove_button.region.width * 3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disk_selector_switches_between_configured_disks_without_blank_option() -> None:
|
|
service = FakeProxmoxService()
|
|
app = PveVmSetupApp(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
service=service,
|
|
)
|
|
|
|
async with app.run_test() as pilot:
|
|
app.query_one(LoginView).remove()
|
|
wizard = WizardView(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
WorkflowState(),
|
|
service,
|
|
)
|
|
await app.query_one("#app-body").mount(wizard)
|
|
wizard.activate()
|
|
await wait_for_wizard_ready(pilot, app)
|
|
|
|
wizard._workflow.current_step_index = 3
|
|
wizard._show_step()
|
|
await pilot.pause()
|
|
|
|
app.query_one("#disks-add", Button).press()
|
|
await pilot.pause()
|
|
|
|
selector = app.query_one("#disks-select", Select)
|
|
option_values = [value for _, value in selector._options]
|
|
|
|
assert NO_DISK_SELECTED not in option_values
|
|
assert selector.disabled is False
|
|
assert selector.value == "1"
|
|
|
|
await asyncio.wait_for(pilot.click("#disks-select"), timeout=2)
|
|
await pilot.pause()
|
|
assert selector.expanded is True
|
|
|
|
await asyncio.wait_for(pilot.press("up"), timeout=2)
|
|
await pilot.pause()
|
|
await asyncio.wait_for(pilot.press("enter"), timeout=2)
|
|
await pilot.pause()
|
|
|
|
assert selector.expanded is False
|
|
assert selector.value == "0"
|
|
assert wizard._selected_disk_index == 0
|
|
assert app.focused is selector
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_confirm_step_replaces_create_with_exit_after_success() -> None:
|
|
service = FakeProxmoxService()
|
|
app = PveVmSetupApp(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
service=service,
|
|
)
|
|
|
|
async with app.run_test() as pilot:
|
|
app.query_one(LoginView).remove()
|
|
wizard = WizardView(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
WorkflowState(),
|
|
service,
|
|
)
|
|
await app.query_one("#app-body").mount(wizard)
|
|
wizard.activate()
|
|
await wait_for_wizard_ready(pilot, app)
|
|
|
|
wizard._workflow.current_step_index = 7
|
|
wizard._workflow.submission.phase = "success"
|
|
wizard._workflow.submission.message = "VM 123 created."
|
|
wizard._show_step()
|
|
await pilot.pause()
|
|
|
|
create_button = app.query_one("#wizard-create", Button)
|
|
assert str(create_button.label) == "Exit"
|
|
assert app.focused is create_button
|
|
|
|
exited: list[bool] = []
|
|
app.exit = lambda *args, **kwargs: exited.append(True) # type: ignore[method-assign]
|
|
create_button.press()
|
|
await pilot.pause()
|
|
|
|
assert exited == [True]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_confirm_step_asks_whether_to_start_vm_before_submitting() -> None:
|
|
service = FakeProxmoxService()
|
|
app = PveVmSetupApp(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
service=service,
|
|
)
|
|
|
|
async with app.run_test() as pilot:
|
|
app.query_one(LoginView).remove()
|
|
wizard = WizardView(
|
|
AppSettings.from_env({}, load_dotenv_file=False),
|
|
WorkflowState(),
|
|
service,
|
|
)
|
|
await app.query_one("#app-body").mount(wizard)
|
|
wizard.activate()
|
|
await wait_for_wizard_ready(pilot, app)
|
|
|
|
app.query_one("#general-name", Input).value = "demo"
|
|
wizard._workflow.current_step_index = 7
|
|
wizard._show_step()
|
|
await pilot.pause()
|
|
|
|
app.query_one("#wizard-create", Button).press()
|
|
await pilot.pause()
|
|
|
|
assert isinstance(app.screen_stack[-1], AutoStartConfirmModal)
|
|
assert service.created_vms == []
|
|
|
|
app.query_one("#auto-start-no", Button).press()
|
|
|
|
for _ in range(20):
|
|
await pilot.pause(0.05)
|
|
if service.created_vms:
|
|
break
|
|
|
|
assert len(service.created_vms) == 1
|
|
assert service.start_after_create_requests == [False]
|