diff --git a/gns3server/api/routes/compute/qemu_nodes.py b/gns3server/api/routes/compute/qemu_nodes.py index ca858d245..bee61b88a 100644 --- a/gns3server/api/routes/compute/qemu_nodes.py +++ b/gns3server/api/routes/compute/qemu_nodes.py @@ -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(): diff --git a/gns3server/compute/qemu/qemu_vm.py b/gns3server/compute/qemu/qemu_vm.py index 908ce15cb..e8ee33140 100644 --- a/gns3server/compute/qemu/qemu_vm.py +++ b/gns3server/compute/qemu/qemu_vm.py @@ -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): diff --git a/gns3server/db/models/templates.py b/gns3server/db/models/templates.py index ffcaeccef..5d9866c02 100644 --- a/gns3server/db/models/templates.py +++ b/gns3server/db/models/templates.py @@ -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) diff --git a/gns3server/db/tasks.py b/gns3server/db/tasks.py index 1ef935a86..62de57508 100644 --- a/gns3server/db/tasks.py +++ b/gns3server/db/tasks.py @@ -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) diff --git a/gns3server/db_migrations/versions/e2a7c9d4b601_add_cdrom_interfaces.py b/gns3server/db_migrations/versions/e2a7c9d4b601_add_cdrom_interfaces.py new file mode 100644 index 000000000..1f4897378 --- /dev/null +++ b/gns3server/db_migrations/versions/e2a7c9d4b601_add_cdrom_interfaces.py @@ -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") diff --git a/gns3server/schemas/compute/qemu_nodes.py b/gns3server/schemas/compute/qemu_nodes.py index 339f4fdae..30c3b0f3d 100644 --- a/gns3server/schemas/compute/qemu_nodes.py +++ b/gns3server/schemas/compute/qemu_nodes.py @@ -15,7 +15,7 @@ # along with this program. If not, see . 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") diff --git a/gns3server/schemas/controller/appliances.py b/gns3server/schemas/controller/appliances.py index d3300a675..a62617398 100644 --- a/gns3server/schemas/controller/appliances.py +++ b/gns3server/schemas/controller/appliances.py @@ -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" ) diff --git a/gns3server/schemas/controller/templates/qemu_templates.py b/gns3server/schemas/controller/templates/qemu_templates.py index 399918f55..4976dd5fd 100644 --- a/gns3server/schemas/controller/templates/qemu_templates.py +++ b/gns3server/schemas/controller/templates/qemu_templates.py @@ -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") diff --git a/tests/api/routes/compute/test_qemu_nodes.py b/tests/api/routes/compute/test_qemu_nodes.py index 1281f338f..37463fd36 100644 --- a/tests/api/routes/compute/test_qemu_nodes.py +++ b/tests/api/routes/compute/test_qemu_nodes.py @@ -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"} diff --git a/tests/api/routes/controller/test_templates.py b/tests/api/routes/controller/test_templates.py index 6c127a5c0..0e2f9ce9a 100644 --- a/tests/api/routes/controller/test_templates.py +++ b/tests/api/routes/controller/test_templates.py @@ -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 = { diff --git a/tests/compute/qemu/test_qemu_vm.py b/tests/compute/qemu/test_qemu_vm.py index 047a32ad9..66210e92f 100644 --- a/tests/compute/qemu/test_qemu_vm.py +++ b/tests/compute/qemu/test_qemu_vm.py @@ -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()) diff --git a/tests/controller/test_appliance_to_template.py b/tests/controller/test_appliance_to_template.py index 843e52653..a488c2fa0 100644 --- a/tests/controller/test_appliance_to_template.py +++ b/tests/controller/test_appliance_to_template.py @@ -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" diff --git a/tests/controller/test_qemu_cdrom_migration.py b/tests/controller/test_qemu_cdrom_migration.py new file mode 100644 index 000000000..87aa74a94 --- /dev/null +++ b/tests/controller/test_qemu_cdrom_migration.py @@ -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()