diff --git a/stackbox/cli/__main__.py b/stackbox/cli/__main__.py
index 72663b8..d72befb 100644
--- a/stackbox/cli/__main__.py
+++ b/stackbox/cli/__main__.py
@@ -322,10 +322,48 @@ def init(
click.echo(f"\nā Failed to start Ironic services: {e}", err=True)
sys.exit(1)
- # TODO: Phase 6 (Issue #8): Create virtual node
- # TODO: Phase 7 (Issue #11): Configure Tempest
+ # Phase 6: Create virtual node
+ click.echo("\nš» Creating virtual node...")
+ try:
+ from stackbox.core import vm
+
+ vm_manager = vm.LibvirtManager(config_dir_path)
+
+ # Ensure default network exists (required for VM connectivity)
+ vm_manager.ensure_default_network()
+
+ # Create single node (hardcoded for milestone 1)
+ node_name = "stackbox-node-0"
+ node_info = vm_manager.create_vm(
+ name=node_name,
+ memory_mb=4096,
+ vcpus=2,
+ disk_gb=20,
+ network="default",
+ )
+
+ # Save node info to YAML file
+ import yaml
+
+ nodes_dir = config_dir_path / "nodes"
+ nodes_dir.mkdir(parents=True, exist_ok=True)
+ node_file = nodes_dir / f"{node_name}.yaml"
+ node_file.write_text(yaml.dump(node_info))
+
+ click.echo(f"ā
Virtual node created: {node_name}")
+ click.echo(f" Node info saved: {node_file}")
+
+ except Exception as e:
+ click.echo(f"\nā Failed to create VM: {e}", err=True)
+ sys.exit(1)
+
+ # TODO: Phase 7 (Issue #9): Configure BMC emulator
+ # TODO: Phase 8 (Issue #10): Enroll node in Ironic
+ # TODO: Phase 9 (Issue #11): Configure Tempest
click.echo("\nā
Initialization complete!")
+ click.echo(" Ironic API: http://localhost:6385")
+ click.echo(f" Virtual node: {node_name}")
def rebuild(self, service: str, no_cache: bool) -> None:
"""Rebuild a service after code changes."""
diff --git a/stackbox/core/vm.py b/stackbox/core/vm.py
new file mode 100644
index 0000000..733a900
--- /dev/null
+++ b/stackbox/core/vm.py
@@ -0,0 +1,473 @@
+"""Virtual machine management for StackBox."""
+
+from pathlib import Path
+import random
+import subprocess
+from typing import Any
+import uuid
+import xml.etree.ElementTree as ET
+
+import click
+from jinja2 import Environment, FileSystemLoader
+
+
+class LibvirtManager:
+ """Manage libvirt VMs inside containerized libvirt service."""
+
+ def __init__(self, config_dir: Path, container_name: str = "stackbox-libvirt"):
+ """Initialize libvirt manager.
+
+ Args:
+ config_dir: Directory for storing VM configs (.stackbox)
+ container_name: Name of libvirt container
+ """
+ self.config_dir = Path(config_dir)
+ self.vm_dir = self.config_dir / "vms"
+ self.vm_dir.mkdir(parents=True, exist_ok=True)
+ self.container_name = container_name
+
+ # Setup Jinja2 for XML templates
+ template_dir = Path(__file__).parent.parent / "templates"
+ self.env = Environment(loader=FileSystemLoader(str(template_dir)))
+
+ def create_vm(
+ self,
+ name: str,
+ memory_mb: int = 4096,
+ vcpus: int = 2,
+ disk_gb: int = 20,
+ network: str = "default",
+ mac: str | None = None,
+ ) -> dict[str, Any]:
+ """Create a new VM inside containerized libvirt.
+
+ Args:
+ name: VM name (e.g., stackbox-node-0)
+ memory_mb: Memory in MB (default 4096)
+ vcpus: Number of vCPUs (default 2)
+ disk_gb: Disk size in GB (default 20)
+ network: Libvirt network name (default "default")
+ mac: MAC address (auto-generated if None)
+
+ Returns:
+ VM info dict with UUID, MAC, disk path, etc.
+
+ Raises:
+ RuntimeError: If VM creation fails
+ """
+ vm_uuid = str(uuid.uuid4())
+
+ # Generate MAC if not provided
+ if mac is None:
+ mac = self._generate_mac()
+
+ # Disk path inside container
+ container_disk_path = f"/var/lib/libvirt/images/{name}.qcow2"
+
+ # Create disk inside container
+ self._create_disk(name, disk_gb)
+
+ # Render XML template
+ template = self.env.get_template("vm/node.xml.j2")
+ vm_xml = template.render(
+ vm_name=name,
+ vm_uuid=vm_uuid,
+ memory_mb=memory_mb,
+ vcpus=vcpus,
+ disk_path=container_disk_path,
+ network_name=network,
+ mac_address=mac,
+ )
+
+ # Save XML locally as reference
+ xml_path = self.vm_dir / f"{name}.xml"
+ xml_path.write_text(vm_xml)
+
+ # Copy XML into container and define VM
+ self._copy_to_container(xml_path, f"/tmp/{name}.xml")
+ self._virsh(["define", f"/tmp/{name}.xml"])
+
+ click.echo(f"ā
Created VM '{name}' in container")
+ click.echo(f" UUID: {vm_uuid}")
+ click.echo(f" MAC: {mac}")
+ click.echo(f" Disk: {container_disk_path} (inside container)")
+
+ return {
+ "name": name,
+ "uuid": vm_uuid,
+ "mac": mac,
+ "disk_path": container_disk_path,
+ "xml_path": str(xml_path),
+ "memory_mb": memory_mb,
+ "vcpus": vcpus,
+ "disk_gb": disk_gb,
+ }
+
+ def _create_disk(self, vm_name: str, size_gb: int) -> None:
+ """Create qcow2 disk image inside container.
+
+ Args:
+ vm_name: VM name (used for disk filename)
+ size_gb: Size in GB
+
+ Raises:
+ RuntimeError: If disk creation fails
+ """
+ container_disk_path = f"/var/lib/libvirt/images/{vm_name}.qcow2"
+
+ cmd = [
+ "docker",
+ "exec",
+ self.container_name,
+ "qemu-img",
+ "create",
+ "-f",
+ "qcow2",
+ container_disk_path,
+ f"{size_gb}G",
+ ]
+
+ try:
+ subprocess.run(cmd, check=True, capture_output=True, text=True)
+ click.echo(f"Created {size_gb}GB disk: {container_disk_path} (in container)")
+ except subprocess.CalledProcessError as e:
+ raise RuntimeError(f"Failed to create disk: {container_disk_path}\n{e.stderr}") from e
+
+ def _generate_mac(self) -> str:
+ """Generate random MAC address in QEMU/KVM range.
+
+ Returns:
+ MAC address string (52:54:00:xx:xx:xx)
+ """
+ # Use 52:54:00 prefix (QEMU/KVM range)
+ mac = [
+ 0x52,
+ 0x54,
+ 0x00,
+ random.randint(0x00, 0xFF),
+ random.randint(0x00, 0xFF),
+ random.randint(0x00, 0xFF),
+ ]
+ return ":".join(f"{b:02x}" for b in mac)
+
+ def _virsh(self, args: list[str]) -> subprocess.CompletedProcess:
+ """Run virsh command inside libvirt container.
+
+ Args:
+ args: Command arguments (e.g., ["list", "--all"])
+
+ Returns:
+ Completed process
+
+ Raises:
+ RuntimeError: If virsh command fails
+ """
+ cmd = ["docker", "exec", self.container_name, "virsh", *args]
+
+ result = subprocess.run(cmd, capture_output=True, text=True, check=False)
+
+ if result.returncode != 0:
+ raise RuntimeError(f"virsh command failed: {' '.join(args)}\n{result.stderr}")
+
+ return result
+
+ def _copy_to_container(self, local_path: Path, container_path: str) -> None:
+ """Copy file into libvirt container.
+
+ Args:
+ local_path: Local file path
+ container_path: Path inside container
+
+ Raises:
+ RuntimeError: If copy fails
+ """
+ cmd = [
+ "docker",
+ "cp",
+ str(local_path),
+ f"{self.container_name}:{container_path}",
+ ]
+
+ try:
+ subprocess.run(cmd, check=True, capture_output=True, text=True)
+ except subprocess.CalledProcessError as e:
+ raise RuntimeError(f"Failed to copy {local_path} to container\n{e.stderr}") from e
+
+ def start_vm(self, name: str) -> None:
+ """Start a VM.
+
+ Args:
+ name: VM name
+
+ Raises:
+ RuntimeError: If start fails
+ """
+ self._virsh(["start", name])
+ click.echo(f"ā
Started VM '{name}'")
+
+ def stop_vm(self, name: str, force: bool = False) -> None:
+ """Stop a VM.
+
+ Args:
+ name: VM name
+ force: Force stop (destroy) instead of graceful shutdown
+
+ Raises:
+ RuntimeError: If stop fails
+ """
+ if force:
+ self._virsh(["destroy", name])
+ else:
+ self._virsh(["shutdown", name])
+
+ click.echo(f"ā
Stopped VM '{name}'")
+
+ def delete_vm(self, name: str, delete_disk: bool = True) -> None:
+ """Delete a VM and optionally its disk.
+
+ Args:
+ name: VM name
+ delete_disk: Also delete disk file from container
+
+ Raises:
+ RuntimeError: If deletion fails
+ """
+ # Undefine VM
+ self._virsh(["undefine", name])
+
+ # Delete disk inside container if requested
+ if delete_disk:
+ container_disk_path = f"/var/lib/libvirt/images/{name}.qcow2"
+ try:
+ subprocess.run(
+ [
+ "docker",
+ "exec",
+ self.container_name,
+ "rm",
+ "-f",
+ container_disk_path,
+ ],
+ check=True,
+ capture_output=True,
+ text=True,
+ )
+ click.echo(f"Deleted disk: {container_disk_path} (from container)")
+ except subprocess.CalledProcessError:
+ click.echo(f"Warning: Could not delete disk {container_disk_path}", err=True)
+
+ # Delete local XML reference
+ xml_path = self.vm_dir / f"{name}.xml"
+ if xml_path.exists():
+ xml_path.unlink()
+
+ click.echo(f"ā
Deleted VM '{name}'")
+
+ def list_vms(self) -> list[dict[str, str]]:
+ """List all VMs in containerized libvirt.
+
+ Returns:
+ List of VM info dicts with 'name' and 'state'
+
+ Raises:
+ RuntimeError: If list fails
+ """
+ result = self._virsh(["list", "--all", "--name"])
+ vms = []
+
+ for name in result.stdout.strip().split("\n"):
+ if name:
+ state_result = self._virsh(["domstate", name])
+ state = state_result.stdout.strip()
+ vms.append({"name": name, "state": state})
+
+ return vms
+
+ def network_exists(self, network_name: str) -> bool:
+ """Check if a libvirt network exists.
+
+ Args:
+ network_name: Network name to check
+
+ Returns:
+ True if network exists, False otherwise
+ """
+ try:
+ result = self._virsh(["net-info", network_name])
+ return result.returncode == 0
+ except RuntimeError:
+ # net-info fails if network doesn't exist
+ return False
+
+ def ensure_default_network(self) -> None:
+ """Ensure the default libvirt network exists and is active.
+
+ Creates and starts the default network if it doesn't exist.
+ This is needed for VM network connectivity.
+
+ Raises:
+ RuntimeError: If network creation fails
+ """
+ if self.network_exists("default"):
+ # Network exists, check if it's active
+ result = self._virsh(["net-info", "default"])
+ if "Active: yes" not in result.stdout:
+ click.echo("Starting default network...")
+ self._virsh(["net-start", "default"])
+ return
+
+ click.echo("Creating default libvirt network...")
+
+ # Default network XML definition
+ default_network_xml = """
+ default
+
+
+
+
+
+
+
+"""
+
+ # Write XML to temporary file in container
+ try:
+ # Copy XML into container
+ subprocess.run(
+ [
+ "docker",
+ "exec",
+ "-i",
+ self.container_name,
+ "bash",
+ "-c",
+ "cat > /tmp/default-network.xml",
+ ],
+ input=default_network_xml,
+ check=True,
+ capture_output=True,
+ text=True,
+ )
+
+ # Define the network
+ self._virsh(["net-define", "/tmp/default-network.xml"])
+
+ # Set to autostart
+ self._virsh(["net-autostart", "default"])
+
+ # Start the network
+ self._virsh(["net-start", "default"])
+
+ click.echo("ā
Default network created and started")
+
+ except subprocess.CalledProcessError as e:
+ # Docker exec failed - container likely not running
+ error_msg = str(e.stderr) if e.stderr else str(e)
+
+ raise RuntimeError(
+ f"Failed to access libvirt container: {error_msg}\n\n"
+ "Troubleshooting:\n"
+ f" 1. Check if container is running: docker ps --filter name={self.container_name}\n"
+ f" 2. If stopped, start it: docker start {self.container_name}\n"
+ f" 3. Check container logs: docker logs {self.container_name}\n"
+ ) from e
+
+ except RuntimeError as e:
+ # virsh command failed - analyze the error
+ error_msg = str(e)
+
+ # Detect specific error patterns and provide targeted help
+ if "bridge" in error_msg.lower() or "virbr0" in error_msg.lower():
+ raise RuntimeError(
+ f"Network bridge conflict detected:\n{error_msg}\n\n"
+ "The bridge 'virbr0' may already exist on your host.\n\n"
+ "Solutions:\n"
+ " 1. Use a different bridge name (requires template modification)\n"
+ " 2. Stop host libvirt: sudo systemctl stop libvirtd\n"
+ " 3. Remove conflicting bridge: sudo ip link delete virbr0\n\n"
+ "Note: This is a containerized setup - the bridge should be isolated.\n"
+ "Check if host libvirt is conflicting with the container."
+ ) from e
+
+ elif "permission denied" in error_msg.lower():
+ raise RuntimeError(
+ f"Permission error creating network:\n{error_msg}\n\n"
+ "The libvirt container may not have sufficient privileges.\n\n"
+ "Verify in docker-compose.yml:\n"
+ " libvirt:\n"
+ " privileged: true\n"
+ " volumes:\n"
+ " - /dev:/dev\n\n"
+ "Then restart the container:\n"
+ f" docker restart {self.container_name}"
+ ) from e
+
+ elif "already exists" in error_msg.lower():
+ raise RuntimeError(
+ f"Network definition conflict:\n{error_msg}\n\n"
+ "A network with this name already exists but wasn't detected.\n\n"
+ "Fix:\n"
+ f" docker exec {self.container_name} virsh net-undefine default\n"
+ " Then retry VM creation"
+ ) from e
+
+ else:
+ # Generic virsh error
+ raise RuntimeError(
+ f"Failed to create default network:\n{error_msg}\n\n"
+ "This is needed for VM network connectivity.\n\n"
+ "Debug:\n"
+ f" 1. Check virsh inside container: docker exec {self.container_name} virsh net-list --all\n"
+ f" 2. Try manual creation: docker exec {self.container_name} virsh net-define /tmp/default-network.xml\n"
+ f" 3. Check libvirt logs: docker exec {self.container_name} journalctl -u libvirtd"
+ ) from e
+
+ def get_vm_info(self, name: str) -> dict[str, Any]:
+ """Get VM information by parsing XML.
+
+ Args:
+ name: VM name
+
+ Returns:
+ VM info dict with name, uuid, memory_mb, vcpus, mac, disk_path
+
+ Raises:
+ RuntimeError: If VM not found or parsing fails
+ """
+ # Get XML from virsh
+ result = self._virsh(["dumpxml", name])
+ xml = result.stdout
+
+ # Parse XML
+ root = ET.fromstring(xml)
+
+ # Extract basic info (these are required elements)
+ name_elem = root.find("name")
+ uuid_elem = root.find("uuid")
+ memory_elem = root.find("memory")
+ vcpu_elem = root.find("vcpu")
+
+ # These elements must exist in valid domain XML
+ assert name_elem is not None and name_elem.text is not None
+ assert uuid_elem is not None and uuid_elem.text is not None
+ assert memory_elem is not None and memory_elem.text is not None
+ assert vcpu_elem is not None and vcpu_elem.text is not None
+
+ info = {
+ "name": name_elem.text,
+ "uuid": uuid_elem.text,
+ "memory_mb": int(memory_elem.text) // 1024,
+ "vcpus": int(vcpu_elem.text),
+ }
+
+ # Get MAC address
+ interface = root.find(".//interface/mac")
+ if interface is not None:
+ info["mac"] = interface.get("address")
+
+ # Get disk path
+ disk = root.find('.//disk[@device="disk"]/source')
+ if disk is not None:
+ info["disk_path"] = disk.get("file")
+
+ return info
diff --git a/stackbox/templates/vm/node.xml.j2 b/stackbox/templates/vm/node.xml.j2
new file mode 100644
index 0000000..bf05bcc
--- /dev/null
+++ b/stackbox/templates/vm/node.xml.j2
@@ -0,0 +1,64 @@
+
+ {{ vm_name }}
+ {{ vm_uuid }}
+ {{ memory_mb }}
+ {{ vcpus }}
+
+ hvm
+
+
+
+
+
+
+
+
+
+
+
+
+
+ destroy
+ restart
+ destroy
+
+ /usr/bin/qemu-system-x86_64
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/tests/unit/core/test_vm.py b/tests/unit/core/test_vm.py
new file mode 100644
index 0000000..c9e1be7
--- /dev/null
+++ b/tests/unit/core/test_vm.py
@@ -0,0 +1,597 @@
+"""Tests for stackbox.core.vm module."""
+
+from pathlib import Path
+import subprocess
+from unittest.mock import MagicMock, patch
+import xml.etree.ElementTree as ET
+
+import pytest
+
+from stackbox.core.vm import LibvirtManager
+
+
+class TestLibvirtManager:
+ """Tests for LibvirtManager class."""
+
+ def test_init_creates_directories(self, tmp_path: Path) -> None:
+ """Test that __init__ creates vms directory."""
+ config_dir = tmp_path / "config"
+ manager = LibvirtManager(config_dir)
+
+ assert manager.config_dir == config_dir
+ assert manager.vm_dir == config_dir / "vms"
+ assert manager.vm_dir.exists()
+ assert manager.container_name == "stackbox-libvirt"
+
+ def test_init_with_custom_container_name(self, tmp_path: Path) -> None:
+ """Test that custom container name is used."""
+ manager = LibvirtManager(tmp_path, container_name="custom-libvirt")
+
+ assert manager.container_name == "custom-libvirt"
+
+
+class TestCreateVM:
+ """Tests for create_vm method."""
+
+ @patch("stackbox.core.vm.LibvirtManager._virsh")
+ @patch("stackbox.core.vm.LibvirtManager._copy_to_container")
+ @patch("stackbox.core.vm.LibvirtManager._create_disk")
+ @patch("click.echo")
+ def test_create_vm_with_defaults(
+ self,
+ mock_echo: MagicMock,
+ mock_create_disk: MagicMock,
+ mock_copy: MagicMock,
+ mock_virsh: MagicMock,
+ tmp_path: Path,
+ ) -> None:
+ """Test VM creation with default parameters."""
+ manager = LibvirtManager(tmp_path)
+
+ result = manager.create_vm("test-node")
+
+ # Check returned info
+ assert result["name"] == "test-node"
+ assert "uuid" in result
+ assert result["memory_mb"] == 4096
+ assert result["vcpus"] == 2
+ assert result["disk_gb"] == 20
+ assert result["mac"].startswith("52:54:00:")
+ assert result["disk_path"] == "/var/lib/libvirt/images/test-node.qcow2"
+
+ # Verify disk was created
+ mock_create_disk.assert_called_once_with("test-node", 20)
+
+ # Verify XML was copied
+ assert mock_copy.called
+ xml_path = mock_copy.call_args[0][0]
+ assert xml_path.name == "test-node.xml"
+
+ # Verify virsh define was called
+ mock_virsh.assert_called_once_with(["define", "/tmp/test-node.xml"])
+
+ # Verify XML file was saved locally
+ saved_xml = tmp_path / "vms" / "test-node.xml"
+ assert saved_xml.exists()
+
+ # Parse and verify XML content
+ xml_content = saved_xml.read_text()
+ root = ET.fromstring(xml_content)
+ assert root.find("name").text == "test-node"
+ assert root.find("memory").text == "4096"
+ assert root.find("vcpu").text == "2"
+
+ # Verify boot order
+ boot_devices = [b.get("dev") for b in root.findall(".//boot")]
+ assert boot_devices == ["network", "hd"]
+
+ @patch("stackbox.core.vm.LibvirtManager._virsh")
+ @patch("stackbox.core.vm.LibvirtManager._copy_to_container")
+ @patch("stackbox.core.vm.LibvirtManager._create_disk")
+ def test_create_vm_with_custom_params(
+ self,
+ mock_create_disk: MagicMock,
+ mock_copy: MagicMock,
+ mock_virsh: MagicMock,
+ tmp_path: Path,
+ ) -> None:
+ """Test VM creation with custom parameters."""
+ manager = LibvirtManager(tmp_path)
+
+ result = manager.create_vm(
+ name="custom-node",
+ memory_mb=8192,
+ vcpus=4,
+ disk_gb=50,
+ network="custom-net",
+ mac="52:54:00:aa:bb:cc",
+ )
+
+ assert result["memory_mb"] == 8192
+ assert result["vcpus"] == 4
+ assert result["disk_gb"] == 50
+ assert result["mac"] == "52:54:00:aa:bb:cc"
+
+ # Verify custom disk size
+ mock_create_disk.assert_called_once_with("custom-node", 50)
+
+ # Verify XML has custom params
+ saved_xml = tmp_path / "vms" / "custom-node.xml"
+ xml_content = saved_xml.read_text()
+ root = ET.fromstring(xml_content)
+ assert root.find("memory").text == "8192"
+ assert root.find("vcpu").text == "4"
+
+ # Verify custom network
+ interface = root.find(".//interface/source")
+ assert interface.get("network") == "custom-net"
+
+ # Verify custom MAC
+ mac_elem = root.find(".//interface/mac")
+ assert mac_elem.get("address") == "52:54:00:aa:bb:cc"
+
+
+class TestCreateDisk:
+ """Tests for _create_disk method."""
+
+ @patch("subprocess.run")
+ @patch("click.echo")
+ def test_create_disk_success(
+ self, mock_echo: MagicMock, mock_run: MagicMock, tmp_path: Path
+ ) -> None:
+ """Test successful disk creation."""
+ manager = LibvirtManager(tmp_path)
+ mock_run.return_value = MagicMock(returncode=0)
+
+ manager._create_disk("test-vm", 20)
+
+ # Verify qemu-img command
+ mock_run.assert_called_once()
+ cmd = mock_run.call_args[0][0]
+ assert cmd == [
+ "docker",
+ "exec",
+ "stackbox-libvirt",
+ "qemu-img",
+ "create",
+ "-f",
+ "qcow2",
+ "/var/lib/libvirt/images/test-vm.qcow2",
+ "20G",
+ ]
+
+ @patch("subprocess.run")
+ def test_create_disk_failure(self, mock_run: MagicMock, tmp_path: Path) -> None:
+ """Test disk creation failure."""
+ manager = LibvirtManager(tmp_path)
+ mock_run.side_effect = subprocess.CalledProcessError(
+ 1, "qemu-img", stderr="Permission denied"
+ )
+
+ with pytest.raises(RuntimeError, match="Failed to create disk"):
+ manager._create_disk("test-vm", 20)
+
+
+class TestGenerateMAC:
+ """Tests for _generate_mac method."""
+
+ def test_generate_mac_format(self, tmp_path: Path) -> None:
+ """Test that generated MAC has correct format."""
+ manager = LibvirtManager(tmp_path)
+ mac = manager._generate_mac()
+
+ # Check format: xx:xx:xx:xx:xx:xx
+ parts = mac.split(":")
+ assert len(parts) == 6
+ assert all(len(p) == 2 for p in parts)
+
+ # Check QEMU/KVM prefix
+ assert mac.startswith("52:54:00:")
+
+ def test_generate_mac_uniqueness(self, tmp_path: Path) -> None:
+ """Test that generated MACs are different."""
+ manager = LibvirtManager(tmp_path)
+ macs = [manager._generate_mac() for _ in range(10)]
+
+ # All should be unique (very high probability)
+ assert len(set(macs)) == 10
+
+
+class TestVirsh:
+ """Tests for _virsh method."""
+
+ @patch("subprocess.run")
+ def test_virsh_success(self, mock_run: MagicMock, tmp_path: Path) -> None:
+ """Test successful virsh command."""
+ manager = LibvirtManager(tmp_path)
+ mock_run.return_value = MagicMock(returncode=0, stdout="output", stderr="")
+
+ result = manager._virsh(["list", "--all"])
+
+ assert result.returncode == 0
+ assert result.stdout == "output"
+
+ # Verify docker exec command
+ mock_run.assert_called_once()
+ cmd = mock_run.call_args[0][0]
+ assert cmd == ["docker", "exec", "stackbox-libvirt", "virsh", "list", "--all"]
+
+ @patch("subprocess.run")
+ def test_virsh_failure(self, mock_run: MagicMock, tmp_path: Path) -> None:
+ """Test virsh command failure."""
+ manager = LibvirtManager(tmp_path)
+ mock_run.return_value = MagicMock(
+ returncode=1, stdout="", stderr="error: failed to connect"
+ )
+
+ with pytest.raises(RuntimeError, match="virsh command failed"):
+ manager._virsh(["list"])
+
+
+class TestCopyToContainer:
+ """Tests for _copy_to_container method."""
+
+ @patch("subprocess.run")
+ def test_copy_success(self, mock_run: MagicMock, tmp_path: Path) -> None:
+ """Test successful file copy."""
+ manager = LibvirtManager(tmp_path)
+ test_file = tmp_path / "test.xml"
+ test_file.write_text("")
+
+ manager._copy_to_container(test_file, "/tmp/test.xml")
+
+ # Verify docker cp command
+ mock_run.assert_called_once()
+ cmd = mock_run.call_args[0][0]
+ assert cmd == [
+ "docker",
+ "cp",
+ str(test_file),
+ "stackbox-libvirt:/tmp/test.xml",
+ ]
+
+ @patch("subprocess.run")
+ def test_copy_failure(self, mock_run: MagicMock, tmp_path: Path) -> None:
+ """Test copy failure."""
+ manager = LibvirtManager(tmp_path)
+ test_file = tmp_path / "test.xml"
+ test_file.write_text("")
+
+ mock_run.side_effect = subprocess.CalledProcessError(1, "docker cp", stderr="Error")
+
+ with pytest.raises(RuntimeError, match="Failed to copy"):
+ manager._copy_to_container(test_file, "/tmp/test.xml")
+
+
+class TestStartStopVM:
+ """Tests for start_vm and stop_vm methods."""
+
+ @patch("stackbox.core.vm.LibvirtManager._virsh")
+ @patch("click.echo")
+ def test_start_vm(self, mock_echo: MagicMock, mock_virsh: MagicMock, tmp_path: Path) -> None:
+ """Test starting a VM."""
+ manager = LibvirtManager(tmp_path)
+
+ manager.start_vm("test-node")
+
+ mock_virsh.assert_called_once_with(["start", "test-node"])
+
+ @patch("stackbox.core.vm.LibvirtManager._virsh")
+ @patch("click.echo")
+ def test_stop_vm_graceful(
+ self, mock_echo: MagicMock, mock_virsh: MagicMock, tmp_path: Path
+ ) -> None:
+ """Test graceful VM shutdown."""
+ manager = LibvirtManager(tmp_path)
+
+ manager.stop_vm("test-node")
+
+ mock_virsh.assert_called_once_with(["shutdown", "test-node"])
+
+ @patch("stackbox.core.vm.LibvirtManager._virsh")
+ @patch("click.echo")
+ def test_stop_vm_force(
+ self, mock_echo: MagicMock, mock_virsh: MagicMock, tmp_path: Path
+ ) -> None:
+ """Test force VM stop."""
+ manager = LibvirtManager(tmp_path)
+
+ manager.stop_vm("test-node", force=True)
+
+ mock_virsh.assert_called_once_with(["destroy", "test-node"])
+
+
+class TestDeleteVM:
+ """Tests for delete_vm method."""
+
+ @patch("subprocess.run")
+ @patch("stackbox.core.vm.LibvirtManager._virsh")
+ @patch("click.echo")
+ def test_delete_vm_with_disk(
+ self,
+ mock_echo: MagicMock,
+ mock_virsh: MagicMock,
+ mock_run: MagicMock,
+ tmp_path: Path,
+ ) -> None:
+ """Test VM deletion with disk removal."""
+ manager = LibvirtManager(tmp_path)
+
+ # Create XML file
+ xml_path = manager.vm_dir / "test-node.xml"
+ xml_path.write_text("")
+
+ manager.delete_vm("test-node", delete_disk=True)
+
+ # Verify virsh undefine
+ mock_virsh.assert_called_once_with(["undefine", "test-node"])
+
+ # Verify disk deletion command
+ assert mock_run.called
+ cmd = mock_run.call_args[0][0]
+ assert "rm" in cmd
+ assert "/var/lib/libvirt/images/test-node.qcow2" in cmd
+
+ # Verify local XML deleted
+ assert not xml_path.exists()
+
+ @patch("stackbox.core.vm.LibvirtManager._virsh")
+ @patch("click.echo")
+ def test_delete_vm_without_disk(
+ self, mock_echo: MagicMock, mock_virsh: MagicMock, tmp_path: Path
+ ) -> None:
+ """Test VM deletion without disk removal."""
+ manager = LibvirtManager(tmp_path)
+
+ # Create XML file
+ xml_path = manager.vm_dir / "test-node.xml"
+ xml_path.write_text("")
+
+ manager.delete_vm("test-node", delete_disk=False)
+
+ # Verify only undefine was called
+ mock_virsh.assert_called_once_with(["undefine", "test-node"])
+
+ # Verify local XML deleted
+ assert not xml_path.exists()
+
+
+class TestListVMs:
+ """Tests for list_vms method."""
+
+ @patch("stackbox.core.vm.LibvirtManager._virsh")
+ def test_list_vms(self, mock_virsh: MagicMock, tmp_path: Path) -> None:
+ """Test listing VMs."""
+ manager = LibvirtManager(tmp_path)
+
+ # Mock virsh responses
+ mock_virsh.side_effect = [
+ MagicMock(stdout="vm1\nvm2\n"),
+ MagicMock(stdout="running"),
+ MagicMock(stdout="shut off"),
+ ]
+
+ vms = manager.list_vms()
+
+ assert len(vms) == 2
+ assert vms[0] == {"name": "vm1", "state": "running"}
+ assert vms[1] == {"name": "vm2", "state": "shut off"}
+
+ @patch("stackbox.core.vm.LibvirtManager._virsh")
+ def test_list_vms_empty(self, mock_virsh: MagicMock, tmp_path: Path) -> None:
+ """Test listing VMs when none exist."""
+ manager = LibvirtManager(tmp_path)
+ mock_virsh.return_value = MagicMock(stdout="")
+
+ vms = manager.list_vms()
+
+ assert vms == []
+
+
+class TestGetVMInfo:
+ """Tests for get_vm_info method."""
+
+ @patch("stackbox.core.vm.LibvirtManager._virsh")
+ def test_get_vm_info(self, mock_virsh: MagicMock, tmp_path: Path) -> None:
+ """Test getting VM information."""
+ manager = LibvirtManager(tmp_path)
+
+ # Mock virsh dumpxml output
+ vm_xml = """
+
+ test-vm
+ abc-123
+ 4194304
+ 2
+
+
+
+
+
+
+
+
+
+ """
+ mock_virsh.return_value = MagicMock(stdout=vm_xml)
+
+ info = manager.get_vm_info("test-vm")
+
+ assert info["name"] == "test-vm"
+ assert info["uuid"] == "abc-123"
+ assert info["memory_mb"] == 4096
+ assert info["vcpus"] == 2
+ assert info["mac"] == "52:54:00:aa:bb:cc"
+ assert info["disk_path"] == "/var/lib/libvirt/images/test-vm.qcow2"
+
+
+class TestNetworkExists:
+ """Tests for network_exists method."""
+
+ @patch("stackbox.core.vm.LibvirtManager._virsh")
+ def test_network_exists_returns_true(self, mock_virsh: MagicMock, tmp_path: Path) -> None:
+ """Test network_exists returns True when network exists."""
+ manager = LibvirtManager(tmp_path)
+ mock_virsh.return_value = MagicMock(returncode=0)
+
+ result = manager.network_exists("default")
+
+ assert result is True
+ mock_virsh.assert_called_once_with(["net-info", "default"])
+
+ @patch("stackbox.core.vm.LibvirtManager._virsh")
+ def test_network_exists_returns_false(self, mock_virsh: MagicMock, tmp_path: Path) -> None:
+ """Test network_exists returns False when network doesn't exist."""
+ manager = LibvirtManager(tmp_path)
+ mock_virsh.side_effect = RuntimeError("Network not found")
+
+ result = manager.network_exists("default")
+
+ assert result is False
+
+
+class TestEnsureDefaultNetwork:
+ """Tests for ensure_default_network method."""
+
+ @patch("stackbox.core.vm.LibvirtManager._virsh")
+ @patch("click.echo")
+ def test_ensure_network_when_active(
+ self, mock_echo: MagicMock, mock_virsh: MagicMock, tmp_path: Path
+ ) -> None:
+ """Test that ensure_default_network does nothing if network is active."""
+ manager = LibvirtManager(tmp_path)
+ mock_virsh.side_effect = [
+ MagicMock(returncode=0), # net-info for exists check
+ MagicMock(stdout="Active: yes"), # net-info for active check
+ ]
+
+ manager.ensure_default_network()
+
+ # Should only check network status
+ assert mock_virsh.call_count == 2
+
+ @patch("stackbox.core.vm.LibvirtManager._virsh")
+ @patch("click.echo")
+ def test_ensure_network_starts_inactive(
+ self, mock_echo: MagicMock, mock_virsh: MagicMock, tmp_path: Path
+ ) -> None:
+ """Test that ensure_default_network starts inactive network."""
+ manager = LibvirtManager(tmp_path)
+ mock_virsh.side_effect = [
+ MagicMock(returncode=0), # net-info for exists check
+ MagicMock(stdout="Active: no"), # net-info for active check
+ MagicMock(returncode=0), # net-start
+ ]
+
+ manager.ensure_default_network()
+
+ # Should check and start network
+ assert mock_virsh.call_count == 3
+ mock_virsh.assert_any_call(["net-start", "default"])
+
+ @patch("subprocess.run")
+ @patch("stackbox.core.vm.LibvirtManager._virsh")
+ @patch("click.echo")
+ def test_ensure_network_creates_when_missing(
+ self,
+ mock_echo: MagicMock,
+ mock_virsh: MagicMock,
+ mock_run: MagicMock,
+ tmp_path: Path,
+ ) -> None:
+ """Test that ensure_default_network creates network if missing."""
+ manager = LibvirtManager(tmp_path)
+
+ # First call to net-info fails (network doesn't exist)
+ # Then successful calls for define, autostart, start
+ mock_virsh.side_effect = [
+ RuntimeError("Network not found"), # net-info fails
+ MagicMock(returncode=0), # net-define
+ MagicMock(returncode=0), # net-autostart
+ MagicMock(returncode=0), # net-start
+ ]
+
+ manager.ensure_default_network()
+
+ # Should have created network XML via docker exec
+ assert mock_run.called
+ cmd = mock_run.call_args[0][0]
+ assert "docker" in cmd
+ assert "exec" in cmd
+ assert "bash" in cmd
+
+ # Should have called define, autostart, start
+ mock_virsh.assert_any_call(["net-define", "/tmp/default-network.xml"])
+ mock_virsh.assert_any_call(["net-autostart", "default"])
+ mock_virsh.assert_any_call(["net-start", "default"])
+
+ @patch("subprocess.run")
+ @patch("stackbox.core.vm.LibvirtManager._virsh")
+ def test_ensure_network_raises_on_container_error(
+ self, mock_virsh: MagicMock, mock_run: MagicMock, tmp_path: Path
+ ) -> None:
+ """Test error when container is not accessible."""
+ manager = LibvirtManager(tmp_path)
+
+ # Network doesn't exist
+ mock_virsh.side_effect = RuntimeError("Network not found")
+ # Docker exec fails (container not running)
+ mock_run.side_effect = subprocess.CalledProcessError(
+ 1, "docker", stderr="Error: No such container"
+ )
+
+ with pytest.raises(RuntimeError, match="Failed to access libvirt container"):
+ manager.ensure_default_network()
+
+ @patch("subprocess.run")
+ @patch("stackbox.core.vm.LibvirtManager._virsh")
+ def test_ensure_network_raises_on_bridge_conflict(
+ self, mock_virsh: MagicMock, mock_run: MagicMock, tmp_path: Path
+ ) -> None:
+ """Test error when bridge already exists."""
+ manager = LibvirtManager(tmp_path)
+
+ # Network doesn't exist
+ # Docker exec succeeds
+ # But virsh net-define fails due to bridge conflict
+ mock_virsh.side_effect = [
+ RuntimeError("Network not found"), # network_exists check
+ RuntimeError("error: bridge 'virbr0' already in use"), # net-define fails
+ ]
+
+ with pytest.raises(RuntimeError, match="Network bridge conflict detected"):
+ manager.ensure_default_network()
+
+ @patch("subprocess.run")
+ @patch("stackbox.core.vm.LibvirtManager._virsh")
+ def test_ensure_network_raises_on_permission_error(
+ self, mock_virsh: MagicMock, mock_run: MagicMock, tmp_path: Path
+ ) -> None:
+ """Test error when permissions are insufficient."""
+ manager = LibvirtManager(tmp_path)
+
+ # Network doesn't exist
+ mock_virsh.side_effect = [
+ RuntimeError("Network not found"), # network_exists check
+ RuntimeError("error: permission denied"), # net-define fails
+ ]
+
+ with pytest.raises(RuntimeError, match="Permission error creating network"):
+ manager.ensure_default_network()
+
+ @patch("subprocess.run")
+ @patch("stackbox.core.vm.LibvirtManager._virsh")
+ def test_ensure_network_raises_on_generic_virsh_error(
+ self, mock_virsh: MagicMock, mock_run: MagicMock, tmp_path: Path
+ ) -> None:
+ """Test generic virsh error handling."""
+ manager = LibvirtManager(tmp_path)
+
+ # Network doesn't exist
+ mock_virsh.side_effect = [
+ RuntimeError("Network not found"), # network_exists check
+ RuntimeError("error: unknown virsh error"), # generic failure
+ ]
+
+ with pytest.raises(RuntimeError, match="Failed to create default network"):
+ manager.ensure_default_network()