mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-10-20 01:42:21 +03:00
feat(qemu): support SATA CD-ROM interfaces
- Add per-drive IDE/SATA selection with backward-compatible IDE defaults - Persist interfaces in nodes, templates, and appliance imports - Add database migration support for versioned and unversioned databases - Prevent partial updates when rejecting live interface changes - Validate interfaces and support SATA media changes through the monitor
This commit is contained in:
Cristi
parent
84b4c05b9f
commit
58e5591935
@ -135,6 +135,10 @@ async def update_qemu_node(node_data: schemas.QemuUpdate, node: QemuVM = Depends
|
||||
"""
|
||||
|
||||
data = jsonable_encoder(node_data, exclude_unset=True)
|
||||
# Reject interface changes before applying any other property or changing live media.
|
||||
for name in ("cdrom_interface", "cdrom2_interface"):
|
||||
if name in data:
|
||||
node.validate_cdrom_interface(name, data[name])
|
||||
# update the console first to avoid issue if updating console type
|
||||
node.console = data.pop("console", node.console)
|
||||
for name, value in data.items():
|
||||
|
||||
@ -159,6 +159,8 @@ class QemuVM(KernelDatapathMixin, BaseNode):
|
||||
self._hdd_disk_interface = "none"
|
||||
self._cdrom_image = ""
|
||||
self._cdrom2_image = ""
|
||||
self._cdrom_interface = "ide"
|
||||
self._cdrom2_interface = "ide"
|
||||
self._bios_image = ""
|
||||
self._boot_priority = "c"
|
||||
self._mac_address = ""
|
||||
@ -533,6 +535,31 @@ class QemuVM(KernelDatapathMixin, BaseNode):
|
||||
def cdrom2_image(self, image):
|
||||
self._cdrom2_image = self.manager.get_abs_image_path(image, self.working_dir) if image else ""
|
||||
|
||||
@property
|
||||
def cdrom_interface(self):
|
||||
return self._cdrom_interface
|
||||
|
||||
@cdrom_interface.setter
|
||||
def cdrom_interface(self, interface):
|
||||
self.validate_cdrom_interface("cdrom_interface", interface)
|
||||
self._cdrom_interface = interface
|
||||
|
||||
@property
|
||||
def cdrom2_interface(self):
|
||||
return self._cdrom2_interface
|
||||
|
||||
@cdrom2_interface.setter
|
||||
def cdrom2_interface(self, interface):
|
||||
self.validate_cdrom_interface("cdrom2_interface", interface)
|
||||
self._cdrom2_interface = interface
|
||||
|
||||
def validate_cdrom_interface(self, name, interface):
|
||||
"""Validate without mutating the VM, including during API preflight."""
|
||||
if interface not in ("ide", "sata"):
|
||||
raise QemuError("CD/DVD interface must be ide or sata")
|
||||
if interface != getattr(self, name) and self.is_running():
|
||||
raise QemuError("Stop the VM before changing a CD/DVD interface")
|
||||
|
||||
async def update_property(self, name, value):
|
||||
"""
|
||||
Update Qemu VM properties.
|
||||
@ -556,7 +583,9 @@ class QemuVM(KernelDatapathMixin, BaseNode):
|
||||
return
|
||||
image = getattr(self, name)
|
||||
self._cdrom_option(name)
|
||||
device = "ide1-cd0" if name == "cdrom_image" else "ide1-cd1"
|
||||
slot = 0 if name == "cdrom_image" else 1
|
||||
interface = self.cdrom_interface if slot == 0 else self.cdrom2_interface
|
||||
device = f"sata-cdrom{slot}" if interface == "sata" else f"ide1-cd{slot}"
|
||||
reader, writer = await self._open_qemu_monitor_connection_vm()
|
||||
if reader is None or writer is None:
|
||||
raise QemuError("Could not connect to the QEMU monitor to change CD/DVD media")
|
||||
@ -576,7 +605,7 @@ class QemuVM(KernelDatapathMixin, BaseNode):
|
||||
try:
|
||||
await asyncio.wait_for(reader.readuntil(b"(qemu) "), timeout=3)
|
||||
blocks = await command("info block")
|
||||
if not re.search(rf"(?m)^{device}[: ]", blocks):
|
||||
if not re.search(rf"(?m)^{re.escape(device)}(?:\s+\([^\r\n]*\))?:", blocks):
|
||||
raise QemuError("Stop the VM before adding this CD/DVD drive")
|
||||
if image:
|
||||
filename = image
|
||||
@ -2430,30 +2459,44 @@ class QemuVM(KernelDatapathMixin, BaseNode):
|
||||
return options
|
||||
|
||||
def _cdrom_option(self, name=None):
|
||||
|
||||
options = []
|
||||
if name in (None, "cdrom_image") and self._cdrom_image:
|
||||
if not os.path.isfile(self._cdrom_image) or not os.path.exists(self._cdrom_image):
|
||||
if os.path.islink(self._cdrom_image):
|
||||
raise QemuError(
|
||||
f"cdrom image '{self._cdrom_image}' linked to '{os.path.realpath(self._cdrom_image)}' is not accessible"
|
||||
)
|
||||
else:
|
||||
raise QemuError(f"cdrom image '{self._cdrom_image}' is not accessible")
|
||||
if self._hdc_disk_image:
|
||||
raise QemuError("You cannot use a disk image on hdc disk and a CDROM image at the same time")
|
||||
options.extend(["-cdrom", self._cdrom_image.replace(",", ",,")])
|
||||
if name in (None, "cdrom2_image") and self._cdrom2_image:
|
||||
if not os.path.isfile(self._cdrom2_image):
|
||||
raise QemuError(f"cdrom image '{self._cdrom2_image}' is not accessible")
|
||||
interface = self._hdd_disk_interface
|
||||
if self._create_config_disk and interface == "none":
|
||||
interface = self._hda_disk_interface
|
||||
if (self._hdd_disk_image or self._create_config_disk) and interface == "ide":
|
||||
raise QemuError("You cannot use an IDE hdd disk and a second CDROM image at the same time")
|
||||
options.extend(
|
||||
["-drive", f"file={self._cdrom2_image.replace(',', ',,')},if=ide,index=3,media=cdrom,readonly=on"]
|
||||
)
|
||||
for slot, image_name in enumerate(("cdrom_image", "cdrom2_image")):
|
||||
image = getattr(self, image_name)
|
||||
if name not in (None, image_name) or not image:
|
||||
continue
|
||||
if not os.path.isfile(image):
|
||||
if os.path.islink(image):
|
||||
raise QemuError(f"cdrom image '{image}' linked to '{os.path.realpath(image)}' is not accessible")
|
||||
raise QemuError(f"cdrom image '{image}' is not accessible")
|
||||
interface = self.cdrom_interface if slot == 0 else self.cdrom2_interface
|
||||
if interface == "sata":
|
||||
# Separate IDs avoid collisions with SATA hard disk controllers.
|
||||
options.extend(
|
||||
[
|
||||
"-device",
|
||||
f"ich9-ahci,id=cdrom-ahci{slot}",
|
||||
"-drive",
|
||||
f"file={image.replace(',', ',,')},if=none,id=sata-cdrom{slot},media=cdrom,readonly=on",
|
||||
"-device",
|
||||
f"ide-cd,drive=sata-cdrom{slot},bus=cdrom-ahci{slot}.0",
|
||||
]
|
||||
)
|
||||
elif slot == 0:
|
||||
if self._hdc_disk_image and self._hdc_disk_interface == "ide":
|
||||
raise QemuError("You cannot use an IDE hdc disk and a CDROM image at the same time")
|
||||
options.extend(["-cdrom", image.replace(",", ",,")])
|
||||
else:
|
||||
disk_interface = self._hdd_disk_interface
|
||||
if self._create_config_disk and disk_interface == "none":
|
||||
disk_interface = self._hda_disk_interface
|
||||
if (self._hdd_disk_image or self._create_config_disk) and disk_interface == "ide":
|
||||
raise QemuError("You cannot use an IDE hdd disk and a second CDROM image at the same time")
|
||||
options.extend(
|
||||
[
|
||||
"-drive",
|
||||
f"file={image.replace(',', ',,')},if=ide,index=3,media=cdrom,readonly=on",
|
||||
]
|
||||
)
|
||||
return options
|
||||
|
||||
def _bios_option(self):
|
||||
|
||||
@ -202,6 +202,8 @@ class QemuTemplate(Template):
|
||||
hdd_disk_interface = Column(String)
|
||||
cdrom_image = Column(String)
|
||||
cdrom2_image = Column(String)
|
||||
cdrom_interface = Column(String, nullable=False, default="ide", server_default="ide")
|
||||
cdrom2_interface = Column(String, nullable=False, default="ide", server_default="ide")
|
||||
initrd = Column(String)
|
||||
kernel_image = Column(String)
|
||||
bios_image = Column(String)
|
||||
|
||||
@ -146,10 +146,14 @@ async def connect_to_db(app: FastAPI) -> None:
|
||||
from gns3server.db_migrations.versions.d9e8a2b7c401_image_inventory_reconciliation import (
|
||||
upgrade,
|
||||
)
|
||||
from gns3server.db_migrations.versions.e2a7c9d4b601_add_cdrom_interfaces import (
|
||||
upgrade as upgrade_cdrom_interfaces,
|
||||
)
|
||||
|
||||
with Operations.context(MigrationContext.configure(connection)):
|
||||
upgrade()
|
||||
upgrade_cdrom()
|
||||
upgrade_cdrom_interfaces()
|
||||
|
||||
await conn.run_sync(upgrade_inventory)
|
||||
await conn.run_sync(run_stamp, alembic_cfg)
|
||||
|
||||
@ -0,0 +1,22 @@
|
||||
"""Add QEMU CD/DVD interfaces."""
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
revision = "e2a7c9d4b601"
|
||||
down_revision = "a6c8e1b2d490"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade():
|
||||
existing = {column["name"] for column in sa.inspect(op.get_bind()).get_columns("qemu_templates")}
|
||||
for name in ("cdrom_interface", "cdrom2_interface"):
|
||||
if name in existing:
|
||||
continue
|
||||
op.add_column("qemu_templates", sa.Column(name, sa.String(), nullable=False, server_default="ide"))
|
||||
|
||||
|
||||
def downgrade():
|
||||
op.drop_column("qemu_templates", "cdrom2_interface")
|
||||
op.drop_column("qemu_templates", "cdrom_interface")
|
||||
@ -15,7 +15,7 @@
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
from enum import Enum
|
||||
from typing import List, Optional
|
||||
from typing import List, Literal, Optional
|
||||
from uuid import UUID
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
@ -152,6 +152,9 @@ class QemuDiskInterfaceType(str, Enum):
|
||||
none = "none"
|
||||
|
||||
|
||||
QemuCdromInterface = Literal["ide", "sata"]
|
||||
|
||||
|
||||
class QemuBase(BaseModel):
|
||||
"""
|
||||
Common Qemu node properties.
|
||||
@ -184,6 +187,8 @@ class QemuBase(BaseModel):
|
||||
hdd_disk_image_md5sum: Optional[str] = Field(None, description="QEMU hdd disk image checksum")
|
||||
hdd_disk_interface: Optional[QemuDiskInterfaceType] = Field(None, description="QEMU hdd interface")
|
||||
cdrom_image: Optional[str] = Field(None, description="QEMU first cdrom image path")
|
||||
cdrom_interface: QemuCdromInterface = Field("ide", description="First CD/DVD interface")
|
||||
cdrom2_interface: QemuCdromInterface = Field("ide", description="Second CD/DVD interface")
|
||||
cdrom2_image: Optional[str] = Field(None, description="QEMU second cdrom image path")
|
||||
cdrom_image_md5sum: Optional[str] = Field(None, description="QEMU first cdrom image checksum")
|
||||
cdrom2_image_md5sum: Optional[str] = Field(None, description="QEMU second cdrom image checksum")
|
||||
|
||||
@ -23,6 +23,7 @@ from uuid import UUID
|
||||
from pydantic import AnyUrl, BaseModel, BeforeValidator, Discriminator, EmailStr, Field, Tag
|
||||
|
||||
from ..common import ExtraConfig
|
||||
from ..compute.qemu_nodes import QemuCdromInterface
|
||||
|
||||
# ============================================================================
|
||||
# Shared Enums
|
||||
@ -398,6 +399,8 @@ class Qemu(BaseModel):
|
||||
hdc_disk_interface: Optional[QemuDiskInterface] = Field(
|
||||
None, title="Disk interface for the installed hdc_disk_image"
|
||||
)
|
||||
cdrom_interface: Optional[QemuCdromInterface] = Field(None, title="First CD/DVD interface")
|
||||
cdrom2_interface: Optional[QemuCdromInterface] = Field(None, title="Second CD/DVD interface")
|
||||
hdd_disk_interface: Optional[QemuDiskInterface] = Field(
|
||||
None, title="Disk interface for the installed hdd_disk_image"
|
||||
)
|
||||
@ -579,6 +582,8 @@ class QemuPropertiesV8(BaseModel):
|
||||
hdc_disk_interface: Optional[QemuDiskInterface] = Field(
|
||||
None, title="Disk interface for the installed hdc_disk_image"
|
||||
)
|
||||
cdrom_interface: Optional[QemuCdromInterface] = Field(None, title="First CD/DVD interface")
|
||||
cdrom2_interface: Optional[QemuCdromInterface] = Field(None, title="Second CD/DVD interface")
|
||||
hdd_disk_interface: Optional[QemuDiskInterface] = Field(
|
||||
None, title="Disk interface for the installed hdd_disk_image"
|
||||
)
|
||||
|
||||
@ -23,6 +23,7 @@ from gns3server.schemas.compute.qemu_nodes import (
|
||||
CustomAdapter,
|
||||
QemuAdapterType,
|
||||
QemuBootPriority,
|
||||
QemuCdromInterface,
|
||||
QemuConsoleType,
|
||||
QemuDiskInterfaceType,
|
||||
QemuOnCloseAction,
|
||||
@ -80,6 +81,8 @@ class QemuTemplate(TemplateBase):
|
||||
QemuDiskInterfaceType.none, description="QEMU hdd interface"
|
||||
)
|
||||
cdrom_image: Optional[str] = Field("", description="QEMU first cdrom image path")
|
||||
cdrom_interface: QemuCdromInterface = Field("ide", description="First CD/DVD interface")
|
||||
cdrom2_interface: QemuCdromInterface = Field("ide", description="Second CD/DVD interface")
|
||||
cdrom2_image: Optional[str] = Field("", description="QEMU second cdrom image path")
|
||||
initrd: Optional[str] = Field("", description="QEMU initrd path")
|
||||
kernel_image: Optional[str] = Field("", description="QEMU kernel image path")
|
||||
|
||||
@ -370,6 +370,36 @@ class TestQemuNodesRoutes:
|
||||
assert response.json()["hdb_disk_image"] == "linux载.img"
|
||||
assert response.json()["ram"] == 1024
|
||||
|
||||
async def test_cdrom_interfaces_update_and_read_back(
|
||||
self, app: FastAPI, compute_client: AsyncClient, qemu_vm: dict
|
||||
) -> None:
|
||||
url = app.url_path_for("compute:update_qemu_node", project_id=qemu_vm["project_id"], node_id=qemu_vm["node_id"])
|
||||
response = await compute_client.put(url, json={"cdrom_interface": "sata", "cdrom2_interface": "sata"})
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["cdrom_interface"] == "sata"
|
||||
assert response.json()["cdrom2_interface"] == "sata"
|
||||
response = await compute_client.get(url)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["cdrom_interface"] == "sata"
|
||||
assert response.json()["cdrom2_interface"] == "sata"
|
||||
|
||||
@pytest.mark.parametrize("name", ["cdrom_interface", "cdrom2_interface"])
|
||||
async def test_cdrom_interface_rejection_does_not_partially_update(self, app, compute_client, qemu_vm, name):
|
||||
url = app.url_path_for("compute:update_qemu_node", project_id=qemu_vm["project_id"], node_id=qemu_vm["node_id"])
|
||||
original = (await compute_client.get(url)).json()
|
||||
with patch("gns3server.compute.qemu.qemu_vm.QemuVM.is_running", return_value=True):
|
||||
response = await compute_client.put(url, json={"ram": original["ram"] + 128, name: "sata"})
|
||||
assert response.status_code == status.HTTP_409_CONFLICT
|
||||
saved = (await compute_client.get(url)).json()
|
||||
assert saved["ram"] == original["ram"]
|
||||
assert saved[name] == original[name]
|
||||
|
||||
@pytest.mark.parametrize("name", ["cdrom_interface", "cdrom2_interface"])
|
||||
async def test_cdrom_interface_rejects_null(self, app, compute_client, qemu_vm, name):
|
||||
url = app.url_path_for("compute:update_qemu_node", project_id=qemu_vm["project_id"], node_id=qemu_vm["node_id"])
|
||||
response = await compute_client.put(url, json={name: None})
|
||||
assert response.status_code == status.HTTP_422_UNPROCESSABLE_CONTENT
|
||||
|
||||
async def test_qemu_nio_create_udp(self, app: FastAPI, compute_client: AsyncClient, qemu_vm: dict) -> None:
|
||||
|
||||
params = {"type": "nio_udp", "lport": 4242, "rport": 4343, "rhost": "127.0.0.1"}
|
||||
|
||||
@ -1112,6 +1112,28 @@ class TestDockerTemplate:
|
||||
|
||||
|
||||
class TestQemuTemplate:
|
||||
async def test_cdrom_interfaces_persist_on_template_update(self, app, client):
|
||||
params = {
|
||||
"name": "SATA installer",
|
||||
"compute_id": "local",
|
||||
"template_type": "qemu",
|
||||
"cdrom_interface": "sata",
|
||||
"cdrom2_interface": "sata",
|
||||
}
|
||||
response = await client.post(app.url_path_for("create_template"), json=params)
|
||||
assert response.status_code == status.HTTP_201_CREATED
|
||||
template_id = response.json()["template_id"]
|
||||
url = app.url_path_for("get_template", template_id=template_id)
|
||||
saved = (await client.get(url)).json()
|
||||
assert saved["cdrom_interface"] == saved["cdrom2_interface"] == "sata"
|
||||
response = await client.put(
|
||||
app.url_path_for("update_template", template_id=template_id), json={"cdrom_interface": "ide"}
|
||||
)
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
saved = (await client.get(url)).json()
|
||||
assert saved["cdrom_interface"] == "ide"
|
||||
assert saved["cdrom2_interface"] == "sata"
|
||||
|
||||
async def test_qemu_template_create(self, app: FastAPI, client: AsyncClient) -> None:
|
||||
|
||||
params = {
|
||||
|
||||
@ -1076,6 +1076,40 @@ async def test_second_cdrom_ide_conflict(vm, images_dir, config_disk):
|
||||
vm._cdrom_option()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_first_cdrom_ide_conflict(vm, images_dir):
|
||||
iso = Path(images_dir) / "installer.iso"
|
||||
iso.touch()
|
||||
vm.cdrom_image = str(iso)
|
||||
vm._hdc_disk_image = str(iso)
|
||||
vm._hdc_disk_interface = "ide"
|
||||
with pytest.raises(QemuError, match="IDE hdc disk"):
|
||||
vm._cdrom_option()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("interface", ["virtio", "scsi", "sata", "nvme"])
|
||||
@pytest.mark.parametrize("cdrom_count", [1, 2])
|
||||
async def test_four_non_ide_disks_with_cdroms(vm, images_dir, interface, cdrom_count):
|
||||
iso = Path(images_dir) / "installer.iso"
|
||||
iso.touch()
|
||||
vm.cdrom_image = str(iso)
|
||||
for drive in "abcd":
|
||||
disk = Path(images_dir) / f"hd{drive}.qcow2"
|
||||
disk.touch()
|
||||
setattr(vm, f"_hd{drive}_disk_image", str(disk))
|
||||
setattr(vm, f"_hd{drive}_disk_interface", interface)
|
||||
expected = ["-cdrom", str(iso)]
|
||||
if cdrom_count == 2:
|
||||
vm.cdrom2_image = str(iso)
|
||||
expected.extend(["-drive", f"file={iso},if=ide,index=3,media=cdrom,readonly=on"])
|
||||
assert vm._cdrom_option() == expected
|
||||
vm.linked_clone = False
|
||||
with patch.object(vm, "_qemu_img_exec", new_callable=AsyncioMagicMock, return_value=0):
|
||||
disk_options = await vm._disk_options()
|
||||
assert sum(option.startswith("file=") for option in disk_options) == 4
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cdrom_monitor(vm):
|
||||
reader = asyncio.StreamReader()
|
||||
@ -1164,3 +1198,68 @@ async def test_cdrom_monitor_error_preserves_previous_image(vm, images_dir, cdro
|
||||
await vm.update_property("cdrom2_image", str(replacement))
|
||||
assert vm.cdrom2_image == str(original)
|
||||
writer.close.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("slot", [0, 1])
|
||||
async def test_sata_cdrom_with_ide_disks(vm, images_dir, slot):
|
||||
iso = Path(images_dir) / "installer,drivers.iso"
|
||||
iso.touch()
|
||||
image_name = "cdrom_image" if slot == 0 else "cdrom2_image"
|
||||
interface_name = "cdrom_interface" if slot == 0 else "cdrom2_interface"
|
||||
setattr(vm, image_name, str(iso))
|
||||
setattr(vm, interface_name, "sata")
|
||||
vm._hdc_disk_image = vm._hdd_disk_image = str(iso)
|
||||
vm._hdc_disk_interface = vm._hdd_disk_interface = "ide"
|
||||
assert vm._cdrom_option() == [
|
||||
"-device",
|
||||
f"ich9-ahci,id=cdrom-ahci{slot}",
|
||||
"-drive",
|
||||
f"file={str(iso).replace(',', ',,')},if=none,id=sata-cdrom{slot},media=cdrom,readonly=on",
|
||||
"-device",
|
||||
f"ide-cd,drive=sata-cdrom{slot},bus=cdrom-ahci{slot}.0",
|
||||
]
|
||||
assert vm.asdict()[interface_name] == "sata"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("name", ["cdrom_interface", "cdrom2_interface"])
|
||||
async def test_cdrom_interface_change_requires_stopped_vm(vm, name):
|
||||
with patch.object(vm, "is_running", return_value=True):
|
||||
with pytest.raises(QemuError, match="Stop the VM"):
|
||||
await vm.update_property(name, "sata")
|
||||
assert getattr(vm, name) == "ide"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("slot", [0, 1])
|
||||
async def test_sata_cdrom_media_update(vm, images_dir, cdrom_monitor, slot):
|
||||
state, writer = cdrom_monitor
|
||||
setattr(vm, "_cdrom_interface" if slot == 0 else "_cdrom2_interface", "sata")
|
||||
state["blocks"] = f"sata-cdrom{slot} (#block123): old.iso\r\n"
|
||||
iso = Path(images_dir) / "new.iso"
|
||||
iso.touch()
|
||||
await vm.update_property("cdrom_image" if slot == 0 else "cdrom2_image", str(iso))
|
||||
writer.write.assert_called_with(f'change sata-cdrom{slot} -f "{iso}"\n'.encode())
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("name", ["cdrom_image", "cdrom2_image"])
|
||||
async def test_cdrom_broken_symlink_error_includes_target(vm, images_dir, name):
|
||||
target = Path(images_dir) / "missing.iso"
|
||||
link = Path(images_dir) / "linked.iso"
|
||||
link.symlink_to(target)
|
||||
setattr(vm, f"_{name}", str(link))
|
||||
with pytest.raises(QemuError, match="linked to") as error:
|
||||
vm._cdrom_option()
|
||||
assert str(target) in str(error.value)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("slot", [0, 1])
|
||||
async def test_sata_cdrom_eject(vm, cdrom_monitor, slot):
|
||||
state, writer = cdrom_monitor
|
||||
setattr(vm, "_cdrom_interface" if slot == 0 else "_cdrom2_interface", "sata")
|
||||
state["blocks"] = f"sata-cdrom{slot}: empty\r\n"
|
||||
await vm.update_property("cdrom_image" if slot == 0 else "cdrom2_image", "")
|
||||
writer.write.assert_called_with(f"eject -f sata-cdrom{slot}\n".encode())
|
||||
|
||||
@ -647,3 +647,33 @@ def test_second_cdrom_survives_appliance_conversion(version):
|
||||
converted = ApplianceToTemplate().new_template(appliance, appliance["versions"][0], "local")
|
||||
assert converted["cdrom_image"] == "installer.iso"
|
||||
assert converted["cdrom2_image"] == "drivers.iso"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", ["cdrom_interface", "cdrom2_interface"])
|
||||
def test_v8_cdrom_interface_survives_validation_and_conversion(name):
|
||||
import copy
|
||||
|
||||
appliance = copy.deepcopy(VYOS_V8)
|
||||
appliance["settings"][0]["template_properties"][name] = "sata"
|
||||
validated = ApplianceModel.model_validate(appliance).model_dump(mode="json", exclude_none=True)
|
||||
template = ApplianceToTemplate().new_template(validated, validated["versions"][1], "local")
|
||||
assert template[name] == "sata"
|
||||
|
||||
|
||||
def test_legacy_cdrom_interfaces_are_qemu_settings():
|
||||
from gns3server.schemas.controller.appliances import Qemu
|
||||
|
||||
settings = Qemu.model_validate(
|
||||
{
|
||||
"adapter_type": "e1000",
|
||||
"adapters": 1,
|
||||
"ram": 512,
|
||||
"arch": "x86_64",
|
||||
"console_type": "telnet",
|
||||
"kvm": "allow",
|
||||
"cdrom_interface": "sata",
|
||||
"cdrom2_interface": "sata",
|
||||
}
|
||||
)
|
||||
assert settings.model_dump()["cdrom_interface"] == "sata"
|
||||
assert settings.model_dump()["cdrom2_interface"] == "sata"
|
||||
|
||||
56
tests/controller/test_qemu_cdrom_migration.py
Normal file
56
tests/controller/test_qemu_cdrom_migration.py
Normal file
@ -0,0 +1,56 @@
|
||||
import pytest
|
||||
import sqlalchemy as sa
|
||||
from alembic import command, config
|
||||
from alembic.operations import Operations
|
||||
from alembic.runtime.migration import MigrationContext
|
||||
from fastapi import FastAPI
|
||||
|
||||
from gns3server.db.models import Base
|
||||
from gns3server.db.tasks import connect_to_db
|
||||
from gns3server.db_migrations.versions import e2a7c9d4b601_add_cdrom_interfaces as migration
|
||||
|
||||
|
||||
def test_cdrom_migration_defaults_idempotence_and_downgrade():
|
||||
engine = sa.create_engine("sqlite://")
|
||||
with engine.begin() as connection:
|
||||
connection.exec_driver_sql("CREATE TABLE qemu_templates (id INTEGER PRIMARY KEY)")
|
||||
connection.exec_driver_sql("INSERT INTO qemu_templates (id) VALUES (1)")
|
||||
with Operations.context(MigrationContext.configure(connection)):
|
||||
migration.upgrade()
|
||||
migration.upgrade()
|
||||
assert connection.exec_driver_sql("SELECT cdrom_interface, cdrom2_interface FROM qemu_templates").one() == (
|
||||
"ide",
|
||||
"ide",
|
||||
)
|
||||
migration.downgrade()
|
||||
assert [column["name"] for column in sa.inspect(connection).get_columns("qemu_templates")] == ["id"]
|
||||
engine.dispose()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("versioned", [False, True])
|
||||
async def test_database_startup_adds_cdrom_interfaces(tmp_path, monkeypatch, versioned):
|
||||
path = tmp_path / "controller.db"
|
||||
engine = sa.create_engine(f"sqlite:///{path}")
|
||||
with engine.connect() as connection:
|
||||
Base.metadata.create_all(connection)
|
||||
connection.exec_driver_sql("ALTER TABLE qemu_templates DROP COLUMN cdrom_interface")
|
||||
connection.exec_driver_sql("ALTER TABLE qemu_templates DROP COLUMN cdrom2_interface")
|
||||
if versioned:
|
||||
cfg = config.Config()
|
||||
cfg.set_main_option("script_location", "gns3server:db_migrations")
|
||||
cfg.attributes["connection"] = connection
|
||||
command.stamp(cfg, migration.down_revision)
|
||||
connection.commit()
|
||||
engine.dispose()
|
||||
monkeypatch.setenv("GNS3_DATABASE_URI", f"sqlite+aiosqlite:///{path}")
|
||||
app = FastAPI()
|
||||
await connect_to_db(app)
|
||||
try:
|
||||
async with app.state._db_engine.connect() as connection:
|
||||
columns = await connection.run_sync(lambda conn: sa.inspect(conn).get_columns("qemu_templates"))
|
||||
assert {"cdrom_interface", "cdrom2_interface"} <= {column["name"] for column in columns}
|
||||
revision = await connection.scalar(sa.text("SELECT version_num FROM alembic_version"))
|
||||
assert revision == migration.revision
|
||||
finally:
|
||||
await app.state._db_engine.dispose()
|
||||
Loading…
x
Reference in New Issue
Block a user