mirror of
https://github.com/GNS3/gns3-server.git
synced 2026-10-21 02:12:21 +03:00
feat(qemu): support two CD/DVD images and consistent ISO handling
This commit is contained in:
Cristi
parent
d047ca0a5f
commit
84b4c05b9f
@ -158,6 +158,7 @@ class QemuVM(KernelDatapathMixin, BaseNode):
|
||||
self._hdc_disk_interface = "none"
|
||||
self._hdd_disk_interface = "none"
|
||||
self._cdrom_image = ""
|
||||
self._cdrom2_image = ""
|
||||
self._bios_image = ""
|
||||
self._boot_priority = "c"
|
||||
self._mac_address = ""
|
||||
@ -524,33 +525,74 @@ class QemuVM(KernelDatapathMixin, BaseNode):
|
||||
else:
|
||||
self._cdrom_image = ""
|
||||
|
||||
@property
|
||||
def cdrom2_image(self):
|
||||
return self._cdrom2_image
|
||||
|
||||
@cdrom2_image.setter
|
||||
def cdrom2_image(self, image):
|
||||
self._cdrom2_image = self.manager.get_abs_image_path(image, self.working_dir) if image else ""
|
||||
|
||||
async def update_property(self, name, value):
|
||||
"""
|
||||
Update Qemu VM properties.
|
||||
"""
|
||||
|
||||
previous = getattr(self, name)
|
||||
setattr(self, name, value)
|
||||
if name == "cdrom_image":
|
||||
# let the guest know about the new cdrom image
|
||||
await self._update_cdrom_image()
|
||||
try:
|
||||
if name in {"cdrom_image", "cdrom2_image"}:
|
||||
await self._update_cdrom_image(name)
|
||||
except (Exception, asyncio.CancelledError):
|
||||
# Old media may have been deleted while QEMU still has it open.
|
||||
# Restore the stored path without resolving it again.
|
||||
setattr(self, f"_{name}", previous)
|
||||
raise
|
||||
self._command_line_changed = True
|
||||
|
||||
async def _update_cdrom_image(self):
|
||||
"""
|
||||
Update the cdrom image path for the Qemu guest OS
|
||||
"""
|
||||
async def _update_cdrom_image(self, name="cdrom_image"):
|
||||
"""Change media only on an existing drive, validating the affected slot."""
|
||||
if not self.is_running():
|
||||
return
|
||||
image = getattr(self, name)
|
||||
self._cdrom_option(name)
|
||||
device = "ide1-cd0" if name == "cdrom_image" else "ide1-cd1"
|
||||
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")
|
||||
|
||||
if self.is_running():
|
||||
if self._cdrom_image:
|
||||
self._cdrom_option() # this will check the cdrom image is accessible
|
||||
await self._control_vm("eject -f ide1-cd0")
|
||||
await self._control_vm(f"change ide1-cd0 {self._cdrom_image}")
|
||||
log.debug(
|
||||
f'QEMU VM "{self._name}" [{self._id}] has changed the cdrom image path to {self._cdrom_image}'
|
||||
)
|
||||
async def command(text):
|
||||
writer.write((text + "\n").encode("utf-8"))
|
||||
await writer.drain()
|
||||
# Consume the echo first: a filename may itself contain "(qemu) ".
|
||||
await asyncio.wait_for(reader.readline(), timeout=3)
|
||||
output = await asyncio.wait_for(reader.readuntil(b"(qemu) "), timeout=3)
|
||||
response = output.decode("utf-8", errors="replace")
|
||||
response = response.removesuffix("(qemu) ").strip()
|
||||
if "Error:" in response or (text != "info block" and response):
|
||||
raise QemuError(response)
|
||||
return response
|
||||
|
||||
try:
|
||||
await asyncio.wait_for(reader.readuntil(b"(qemu) "), timeout=3)
|
||||
blocks = await command("info block")
|
||||
if not re.search(rf"(?m)^{device}[: ]", blocks):
|
||||
raise QemuError("Stop the VM before adding this CD/DVD drive")
|
||||
if image:
|
||||
filename = image
|
||||
if not image.isascii() or any(ord(char) < 32 for char in image):
|
||||
# HMP's line editor cannot reliably handle non-ASCII bytes.
|
||||
# A JSON block filename carries them as ASCII escapes, which
|
||||
# the block layer decodes after the HMP parser has run.
|
||||
filename = "json:" + json.dumps({"driver": "raw", "file": {"driver": "file", "filename": image}})
|
||||
await command(f"change {device} -f {json.dumps(filename)}")
|
||||
else:
|
||||
await self._control_vm("eject -f ide1-cd0")
|
||||
log.debug(f'QEMU VM "{self._name}" [{self._id}] has ejected the cdrom image')
|
||||
await command(f"eject -f {device}")
|
||||
log.debug('QEMU VM "%s" has changed %s to %s', self._name, name, image)
|
||||
except (asyncio.TimeoutError, asyncio.IncompleteReadError, OSError) as e:
|
||||
raise QemuError(f"Could not change CD/DVD media: {e}") from e
|
||||
finally:
|
||||
writer.close()
|
||||
|
||||
@property
|
||||
def bios_image(self):
|
||||
@ -2387,10 +2429,10 @@ class QemuVM(KernelDatapathMixin, BaseNode):
|
||||
|
||||
return options
|
||||
|
||||
def _cdrom_option(self):
|
||||
def _cdrom_option(self, name=None):
|
||||
|
||||
options = []
|
||||
if self._cdrom_image:
|
||||
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(
|
||||
@ -2401,6 +2443,17 @@ class QemuVM(KernelDatapathMixin, BaseNode):
|
||||
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"]
|
||||
)
|
||||
return options
|
||||
|
||||
def _bios_option(self):
|
||||
@ -2868,6 +2921,8 @@ class QemuVM(KernelDatapathMixin, BaseNode):
|
||||
|
||||
answer["cdrom_image"] = self.manager.get_relative_image_path(self._cdrom_image, self.working_dir)
|
||||
answer["cdrom_image_md5sum"] = md5sum(self._cdrom_image, self.working_dir)
|
||||
answer["cdrom2_image"] = self.manager.get_relative_image_path(self._cdrom2_image, self.working_dir)
|
||||
answer["cdrom2_image_md5sum"] = md5sum(self._cdrom2_image, self.working_dir)
|
||||
answer["bios_image"] = self.manager.get_relative_image_path(self._bios_image, self.working_dir)
|
||||
answer["bios_image_md5sum"] = md5sum(self._bios_image, self.working_dir)
|
||||
answer["initrd"] = self.manager.get_relative_image_path(self._initrd, self.working_dir)
|
||||
|
||||
@ -208,6 +208,7 @@ async def _patch_project_file(
|
||||
|
||||
# image files that we need to include in the exported archive
|
||||
images = []
|
||||
portable_references = []
|
||||
|
||||
try:
|
||||
with open(path) as f:
|
||||
@ -249,32 +250,54 @@ async def _patch_project_file(
|
||||
if value is None or value.strip() == "":
|
||||
continue
|
||||
|
||||
if not keep_compute_ids: # If we keep the original compute we can keep the image path
|
||||
if not keep_compute_ids:
|
||||
node["properties"][prop] = os.path.basename(value)
|
||||
|
||||
if include_images is True:
|
||||
images.append({"compute_id": compute_id, "image": value, "image_type": node["node_type"]})
|
||||
if include_images:
|
||||
image = {"compute_id": compute_id, "image": value, "image_type": node["node_type"]}
|
||||
images.append(image)
|
||||
if not keep_compute_ids:
|
||||
portable_references.append((node["properties"], prop, image))
|
||||
|
||||
if not keep_compute_ids:
|
||||
topology["topology"][
|
||||
"computes"
|
||||
] = [] # Strip compute information because could contain secret info like password
|
||||
|
||||
local_images = {i["image"] for i in images if i["compute_id"] == "local"}
|
||||
destinations = _image_export_names(images)
|
||||
for properties, prop, image in portable_references:
|
||||
properties[prop] = destinations[(image["compute_id"], image["image_type"], image["image"])]
|
||||
|
||||
for image in local_images:
|
||||
_export_local_image(image, zstream)
|
||||
|
||||
remote_images = {(i["compute_id"], i["image_type"], i["image"]) for i in images if i["compute_id"] != "local"}
|
||||
|
||||
for compute_id, image_type, image in remote_images:
|
||||
await _export_remote_images(project, compute_id, image_type, image, zstream, temporary_dir)
|
||||
for (compute_id, image_type, image), export_name in destinations.items():
|
||||
if compute_id == "local":
|
||||
_export_local_image(image, zstream, export_name, image_type)
|
||||
else:
|
||||
await _export_remote_images(project, compute_id, image_type, image, zstream, temporary_dir, export_name)
|
||||
|
||||
zstream.writestr("project.gns3", json.dumps(topology, indent=4, sort_keys=True).encode())
|
||||
return images
|
||||
|
||||
|
||||
def _export_local_image(image, zstream):
|
||||
def _image_export_names(images):
|
||||
"""Allocate distinct portable names without changing ordinary image names."""
|
||||
destinations = {}
|
||||
used = set()
|
||||
for image in images:
|
||||
key = (image["compute_id"], image["image_type"], image["image"])
|
||||
if key in destinations:
|
||||
continue
|
||||
filename = os.path.basename(image["image"])
|
||||
stem, extension = os.path.splitext(filename)
|
||||
index = 1
|
||||
while (image["image_type"], filename) in used:
|
||||
index += 1
|
||||
filename = f"{stem}-{index}{extension}"
|
||||
used.add((image["image_type"], filename))
|
||||
destinations[key] = filename
|
||||
return destinations
|
||||
|
||||
|
||||
def _export_local_image(image, zstream, export_name=None, image_type=None):
|
||||
"""
|
||||
Exports a local image to the zip file.
|
||||
|
||||
@ -285,6 +308,8 @@ def _export_local_image(image, zstream):
|
||||
from ..compute import MODULES
|
||||
|
||||
for module in MODULES:
|
||||
if image_type and getattr(module, "_NODE_TYPE", None) != image_type:
|
||||
continue
|
||||
try:
|
||||
images_directory = module.instance().get_images_directory()
|
||||
except NotImplementedError:
|
||||
@ -298,13 +323,15 @@ def _export_local_image(image, zstream):
|
||||
path = os.path.join(images_directory, image)
|
||||
|
||||
if os.path.exists(path):
|
||||
arcname = os.path.join("images", directory, os.path.basename(image))
|
||||
arcname = os.path.join("images", directory, export_name or os.path.basename(image))
|
||||
_patch_mtime(path)
|
||||
zstream.write(path, arcname)
|
||||
return
|
||||
|
||||
|
||||
async def _export_remote_images(project, compute_id, image_type, image, project_zipfile, temporary_dir):
|
||||
async def _export_remote_images(
|
||||
project, compute_id, image_type, image, project_zipfile, temporary_dir, export_name=None
|
||||
):
|
||||
"""
|
||||
Export specific image from remote compute.
|
||||
"""
|
||||
@ -334,5 +361,6 @@ async def _export_remote_images(project, compute_id, image_type, image, project_
|
||||
break
|
||||
await f.write(data)
|
||||
response.close()
|
||||
arcname = os.path.join("images", image_type, image)
|
||||
directory = "QEMU" if image_type == "qemu" else image_type
|
||||
arcname = os.path.join("images", directory, export_name or image)
|
||||
project_zipfile.write(temp_path, arcname=arcname, compress_type=zipfile.ZIP_DEFLATED)
|
||||
|
||||
@ -58,6 +58,7 @@ IMAGE_PROPERTIES_BY_NODE_TYPE = {
|
||||
"hdc_disk_image": "qemu",
|
||||
"hdd_disk_image": "qemu",
|
||||
"cdrom_image": "qemu",
|
||||
"cdrom2_image": "qemu",
|
||||
"initrd": "qemu",
|
||||
"kernel_image": "qemu",
|
||||
"bios_image": "qemu",
|
||||
@ -900,6 +901,7 @@ class Node:
|
||||
"hdc_disk_image",
|
||||
"hdd_disk_image",
|
||||
"cdrom_image",
|
||||
"cdrom2_image",
|
||||
"bios_image",
|
||||
"initrd",
|
||||
"kernel_image",
|
||||
|
||||
@ -45,6 +45,23 @@ class Image(BaseTable):
|
||||
last_error = Column(String)
|
||||
templates = relationship("Template", secondary=image_template_map, back_populates="images")
|
||||
|
||||
@property
|
||||
def reference(self):
|
||||
"""An emulator-relative reference, or an absolute additional-root path."""
|
||||
import os
|
||||
|
||||
from gns3server.utils import force_unix_path
|
||||
from gns3server.utils.images import default_images_directory
|
||||
|
||||
root = os.path.abspath(default_images_directory(self.image_type))
|
||||
path = os.path.abspath(self.path)
|
||||
try:
|
||||
if os.path.commonpath([root, path]) == root:
|
||||
return force_unix_path(os.path.relpath(path, root))
|
||||
except ValueError: # Different drives on Windows.
|
||||
pass
|
||||
return force_unix_path(path)
|
||||
|
||||
|
||||
class ImageSyncJob(BaseTable):
|
||||
__tablename__ = "image_sync_jobs"
|
||||
|
||||
@ -201,6 +201,7 @@ class QemuTemplate(Template):
|
||||
hdd_disk_image = Column(String)
|
||||
hdd_disk_interface = Column(String)
|
||||
cdrom_image = Column(String)
|
||||
cdrom2_image = Column(String)
|
||||
initrd = Column(String)
|
||||
kernel_image = Column(String)
|
||||
bios_image = Column(String)
|
||||
|
||||
@ -140,12 +140,16 @@ async def connect_to_db(app: FastAPI) -> None:
|
||||
def upgrade_inventory(connection):
|
||||
from alembic.operations import Operations
|
||||
|
||||
from gns3server.db_migrations.versions.a6c8e1b2d490_add_second_qemu_cdrom import (
|
||||
upgrade as upgrade_cdrom,
|
||||
)
|
||||
from gns3server.db_migrations.versions.d9e8a2b7c401_image_inventory_reconciliation import (
|
||||
upgrade,
|
||||
)
|
||||
|
||||
with Operations.context(MigrationContext.configure(connection)):
|
||||
upgrade()
|
||||
upgrade_cdrom()
|
||||
|
||||
await conn.run_sync(upgrade_inventory)
|
||||
await conn.run_sync(run_stamp, alembic_cfg)
|
||||
|
||||
@ -0,0 +1,23 @@
|
||||
"""Add a second QEMU CD-ROM image.
|
||||
|
||||
Revision ID: a6c8e1b2d490
|
||||
Revises: f7d3a9c2e5b1
|
||||
"""
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
revision = "a6c8e1b2d490"
|
||||
down_revision = "f7d3a9c2e5b1"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade():
|
||||
existing = {column["name"] for column in sa.inspect(op.get_bind()).get_columns("qemu_templates")}
|
||||
if "cdrom2_image" not in existing:
|
||||
op.add_column("qemu_templates", sa.Column("cdrom2_image", sa.String()))
|
||||
|
||||
|
||||
def downgrade():
|
||||
op.drop_column("qemu_templates", "cdrom2_image")
|
||||
@ -183,8 +183,10 @@ class QemuBase(BaseModel):
|
||||
hdd_disk_image_backing_file: Optional[str] = Field(None, description="QEMU hdd backing file disk image path")
|
||||
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 cdrom image path")
|
||||
cdrom_image_md5sum: Optional[str] = Field(None, description="QEMU cdrom image checksum")
|
||||
cdrom_image: Optional[str] = Field(None, description="QEMU first cdrom image path")
|
||||
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")
|
||||
bios_image: Optional[str] = Field(None, description="QEMU bios image path")
|
||||
bios_image_md5sum: Optional[str] = Field(None, description="QEMU bios image checksum")
|
||||
initrd: Optional[str] = Field(None, description="QEMU initrd path")
|
||||
|
||||
@ -425,7 +425,8 @@ class ApplianceVersionImages(BaseModel):
|
||||
hdb_disk_image: Optional[str] = Field(None, title="Hdc disk image")
|
||||
hdc_disk_image: Optional[str] = Field(None, title="Hdd disk image")
|
||||
hdd_disk_image: Optional[str] = Field(None, title="Hdd diskimage")
|
||||
cdrom_image: Optional[str] = Field(None, title="cdrom image")
|
||||
cdrom_image: Optional[str] = Field(None, title="First cdrom image")
|
||||
cdrom2_image: Optional[str] = Field(None, title="Second cdrom image")
|
||||
|
||||
|
||||
class ApplianceVersion(BaseModel):
|
||||
|
||||
@ -37,6 +37,9 @@ class ImageBase(BaseModel):
|
||||
|
||||
filename: str = Field(..., description="Image filename")
|
||||
path: str = Field(..., description="Image path")
|
||||
reference: Optional[str] = Field(
|
||||
None, description="Image reference relative to its emulator directory, or an absolute additional-root path"
|
||||
)
|
||||
image_type: ImageType = Field(..., description="Image type")
|
||||
image_size: int = Field(..., description="Image size in bytes")
|
||||
checksum: str = Field(..., description="Checksum value")
|
||||
|
||||
@ -79,7 +79,8 @@ class QemuTemplate(TemplateBase):
|
||||
hdd_disk_interface: Optional[QemuDiskInterfaceType] = Field(
|
||||
QemuDiskInterfaceType.none, description="QEMU hdd interface"
|
||||
)
|
||||
cdrom_image: Optional[str] = Field("", description="QEMU cdrom image path")
|
||||
cdrom_image: Optional[str] = Field("", description="QEMU first cdrom image path")
|
||||
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")
|
||||
bios_image: Optional[str] = Field("", description="QEMU bios image path")
|
||||
|
||||
@ -49,6 +49,9 @@ from . import force_unix_path
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
# Include the first ISO 9660 volume descriptor in image type detection.
|
||||
IMAGE_HEADER_SIZE = 16 * 2048 + 7
|
||||
|
||||
|
||||
async def list_images(image_type):
|
||||
"""
|
||||
@ -145,7 +148,7 @@ def inspect_image_file(path, expected_image_type=None, allow_raw_image=False, st
|
||||
info = os.fstat(f.fileno())
|
||||
if not stat.S_ISREG(info.st_mode) or stat_fingerprint(info) != before:
|
||||
raise ImageChangedError(f"Image changed while opening: {path}")
|
||||
header = f.read(7)
|
||||
header = f.read(IMAGE_HEADER_SIZE)
|
||||
if len(header) < 7:
|
||||
raise InvalidImageError(f"Image '{path}' is too small to be valid")
|
||||
image_type = check_valid_image_header(path, header, allow_raw_image)
|
||||
@ -359,6 +362,8 @@ def check_valid_image_header(path: str, data: bytes, allow_raw_image: bool = Fal
|
||||
elif data[:4] == b"QFI\xfb" or data[:4] == b"KDMV":
|
||||
# for Qemy images: file must be QCOW2 or VMDK
|
||||
return "qemu"
|
||||
elif data[16 * 2048 + 1 : 16 * 2048 + 6] == b"CD001":
|
||||
return "qemu"
|
||||
else:
|
||||
if allow_raw_image is True:
|
||||
return "qemu"
|
||||
@ -380,12 +385,14 @@ async def write_image(
|
||||
if subdirectory is not None:
|
||||
validate_image_upload_name(image_filename)
|
||||
subfolders = validate_image_subdirectory(subdirectory)
|
||||
# HTTP chunk boundaries need not align with the seven-byte image header.
|
||||
# HTTP chunk boundaries need not align with the image header or ISO descriptor.
|
||||
iterator = stream.__aiter__()
|
||||
prefix = bytearray()
|
||||
async for chunk in iterator:
|
||||
prefix.extend(chunk)
|
||||
if len(prefix) >= 7:
|
||||
# Known disk formats keep their original early detection behavior.
|
||||
header_size = 7 if prefix.startswith((b"\x7fELF", b"QFI\xfb", b"KDMV")) else IMAGE_HEADER_SIZE
|
||||
if len(prefix) >= header_size:
|
||||
break
|
||||
if len(prefix) < 7:
|
||||
raise InvalidImageError("The image content is empty or too small to be valid")
|
||||
|
||||
@ -476,3 +476,40 @@ class TestImageRoutes:
|
||||
assert len(templates) == 1
|
||||
assert templates[0].name == "Empty VM"
|
||||
await templates_repo.delete_template(templates[0].template_id)
|
||||
|
||||
|
||||
class TestIsoImages:
|
||||
async def test_iso_uploads_can_be_used_together_in_a_template(self, app, client, config, images_dir):
|
||||
config.settings.Server.allow_raw_images = False
|
||||
references = []
|
||||
for index, name in enumerate(("installer.ISO", "drivers.iso")):
|
||||
content = bytes(16 * 2048) + b"\x01CD001\x01" + bytes([index]) * 2041
|
||||
response = await client.post(
|
||||
app.url_path_for("upload_image", image_path=name),
|
||||
params={"subdirectory": "installers"} if index == 0 else {},
|
||||
content=content,
|
||||
)
|
||||
assert response.status_code == status.HTTP_201_CREATED, response.text
|
||||
assert response.json()["image_type"] == "qemu"
|
||||
relative = "installers/" + name if index == 0 else name
|
||||
assert response.json()["path"] == os.path.join(images_dir, "QEMU", relative)
|
||||
assert response.json()["reference"] == relative
|
||||
assert Qemu.instance().get_abs_image_path(relative) == response.json()["path"]
|
||||
references.append(relative)
|
||||
|
||||
response = await client.post(
|
||||
app.url_path_for("create_template"),
|
||||
json={
|
||||
"name": "Windows installer with drivers",
|
||||
"template_type": "qemu",
|
||||
"compute_id": "local",
|
||||
"cdrom_image": references[0],
|
||||
"cdrom2_image": references[1],
|
||||
},
|
||||
)
|
||||
assert response.status_code == status.HTTP_201_CREATED, response.text
|
||||
template_id = response.json()["template_id"]
|
||||
response = await client.get(app.url_path_for("get_template", template_id=template_id))
|
||||
assert response.status_code == status.HTTP_200_OK
|
||||
assert response.json()["cdrom_image"] == references[0]
|
||||
assert response.json()["cdrom2_image"] == references[1]
|
||||
|
||||
@ -15,9 +15,11 @@
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
@ -1039,3 +1041,126 @@ async def test_run_with_kvm_linux_dev_kvm_missing(linux_platform, vm):
|
||||
vm.manager.config.settings.Qemu.require_hardware_acceleration = True
|
||||
with pytest.raises(QemuError):
|
||||
await vm._run_with_hardware_acceleration("qemu-system-x86_64", "")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_two_cdrom_images(vm, images_dir):
|
||||
installer = Path(images_dir) / "installer.iso"
|
||||
drivers = Path(images_dir) / "virtio,drivers.iso"
|
||||
installer.touch()
|
||||
drivers.touch()
|
||||
vm.cdrom_image = str(installer)
|
||||
vm.cdrom2_image = str(drivers)
|
||||
assert vm._cdrom_option() == [
|
||||
"-cdrom",
|
||||
str(installer),
|
||||
"-drive",
|
||||
f"file={str(drivers).replace(',', ',,')},if=ide,index=3,media=cdrom,readonly=on",
|
||||
]
|
||||
assert vm.asdict()["cdrom2_image"].endswith("virtio,drivers.iso")
|
||||
vm.cdrom2_image = ""
|
||||
assert vm._cdrom_option() == ["-cdrom", str(installer)]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("config_disk", [False, True])
|
||||
async def test_second_cdrom_ide_conflict(vm, images_dir, config_disk):
|
||||
iso = Path(images_dir) / "drivers.iso"
|
||||
iso.touch()
|
||||
vm.cdrom2_image = str(iso)
|
||||
vm._hdd_disk_interface = "ide"
|
||||
vm._create_config_disk = config_disk
|
||||
if not config_disk:
|
||||
vm._hdd_disk_image = str(iso)
|
||||
with pytest.raises(QemuError, match="second CDROM"):
|
||||
vm._cdrom_option()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cdrom_monitor(vm):
|
||||
reader = asyncio.StreamReader()
|
||||
reader.feed_data(b"QEMU monitor\r\n(qemu) ")
|
||||
writer = MagicMock()
|
||||
writer.drain = AsyncioMagicMock()
|
||||
state = {"blocks": "ide1-cd0: installer.iso\r\nide1-cd1: drivers.iso\r\n", "error": ""}
|
||||
|
||||
def respond(data):
|
||||
command = data.decode("utf-8").strip()
|
||||
output = state["blocks"] if command == "info block" else state["error"]
|
||||
reader.feed_data(data.rstrip(b"\n") + b"\r\n" + output.encode("utf-8") + b"(qemu) ")
|
||||
|
||||
writer.write.side_effect = respond
|
||||
with (
|
||||
patch.object(vm, "is_running", return_value=True),
|
||||
patch.object(
|
||||
vm, "_open_qemu_monitor_connection_vm", new_callable=AsyncioMagicMock, return_value=(reader, writer)
|
||||
),
|
||||
):
|
||||
yield state, writer
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("filename", ["drivers with spaces.iso", "drivers-雪.iso", "drivers (qemu) .iso"])
|
||||
async def test_second_cdrom_update(vm, images_dir, cdrom_monitor, filename):
|
||||
iso = Path(images_dir) / filename
|
||||
iso.touch()
|
||||
state, writer = cdrom_monitor
|
||||
await vm.update_property("cdrom2_image", str(iso))
|
||||
assert writer.write.call_args_list[0] == mock.call(b"info block\n")
|
||||
command = writer.write.call_args_list[1].args[0].decode("ascii")
|
||||
encoded_filename = json.loads(command.removeprefix("change ide1-cd1 -f ").strip())
|
||||
if encoded_filename.startswith("json:"):
|
||||
assert json.loads(encoded_filename[5:])["file"]["filename"] == str(iso)
|
||||
else:
|
||||
assert encoded_filename == str(iso)
|
||||
assert vm.cdrom2_image == str(iso)
|
||||
writer.close.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_second_cdrom_eject(vm, cdrom_monitor):
|
||||
state, writer = cdrom_monitor
|
||||
await vm.update_property("cdrom2_image", "")
|
||||
writer.write.assert_called_with(b"eject -f ide1-cd1\n")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_second_cdrom_hot_add_requires_restart(vm, images_dir, cdrom_monitor):
|
||||
iso = Path(images_dir) / "drivers.iso"
|
||||
iso.touch()
|
||||
state, writer = cdrom_monitor
|
||||
state["blocks"] = "ide1-cd0: installer.iso\r\n"
|
||||
vm._command_line_changed = False
|
||||
with pytest.raises(QemuError, match="Stop the VM"):
|
||||
await vm.update_property("cdrom2_image", str(iso))
|
||||
assert vm.cdrom2_image == ""
|
||||
assert not vm._command_line_changed
|
||||
assert writer.write.call_count == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_missing_other_iso_does_not_block_media_swap(vm, images_dir, cdrom_monitor):
|
||||
iso = Path(images_dir) / "installer.iso"
|
||||
iso.touch()
|
||||
vm._cdrom2_image = str(Path(images_dir) / "deleted.iso")
|
||||
await vm.update_property("cdrom_image", str(iso))
|
||||
assert vm.cdrom_image == str(iso)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("error", ["Error: media change failed", "change: filename expected"])
|
||||
@pytest.mark.parametrize("original_deleted", [False, True])
|
||||
async def test_cdrom_monitor_error_preserves_previous_image(vm, images_dir, cdrom_monitor, error, original_deleted):
|
||||
original = Path(images_dir) / "original.iso"
|
||||
original.touch()
|
||||
replacement = Path(images_dir) / "replacement.iso"
|
||||
replacement.touch()
|
||||
vm.cdrom2_image = str(original)
|
||||
if original_deleted:
|
||||
original.unlink()
|
||||
state, writer = cdrom_monitor
|
||||
state["error"] = error + "\r\n"
|
||||
with pytest.raises(QemuError, match=error):
|
||||
await vm.update_property("cdrom2_image", str(replacement))
|
||||
assert vm.cdrom2_image == str(original)
|
||||
writer.close.assert_called_once()
|
||||
|
||||
@ -628,3 +628,22 @@ def test_no_appliance_metadata_when_appliance_has_none():
|
||||
template = ApplianceToTemplate().new_template(appliance, None, "local")
|
||||
|
||||
assert "appliance_metadata" not in template
|
||||
|
||||
|
||||
@pytest.mark.parametrize("version", [6, 8])
|
||||
def test_second_cdrom_survives_appliance_conversion(version):
|
||||
from copy import deepcopy
|
||||
|
||||
from gns3server.schemas.controller.appliances import ApplianceVersionImages
|
||||
|
||||
slots = {"cdrom_image": "installer.iso", "cdrom2_image": "drivers.iso"}
|
||||
validated = ApplianceVersionImages.model_validate(slots).model_dump(exclude_none=True)
|
||||
assert validated == slots
|
||||
appliance = deepcopy(VYOS_V8)
|
||||
appliance["versions"][0]["images"] = validated
|
||||
if version == 6:
|
||||
appliance["registry_version"] = 6
|
||||
appliance["qemu"] = {"arch": "x86_64", "ram": 512, "adapters": 1, "kvm": "allow"}
|
||||
converted = ApplianceToTemplate().new_template(appliance, appliance["versions"][0], "local")
|
||||
assert converted["cdrom_image"] == "installer.iso"
|
||||
assert converted["cdrom2_image"] == "drivers.iso"
|
||||
|
||||
@ -413,3 +413,49 @@ async def test_export_with_ignoring_snapshots(tmpdir, project):
|
||||
|
||||
with zipfile.ZipFile(str(tmpdir / "zipfile.zip")) as myzip:
|
||||
assert os.path.join("snapshots", "snap.gns3project") not in [f.filename for f in myzip.filelist]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_export_two_isos_with_identical_filenames(tmpdir, project):
|
||||
from gns3server.compute.qemu import Qemu
|
||||
from gns3server.controller.import_project import _import_images
|
||||
|
||||
root = Path(str(tmpdir)) / "QEMU"
|
||||
installer = root / "windows/install.iso"
|
||||
drivers = root / "drivers/install.iso"
|
||||
for image, data in ((installer, b"installer"), (drivers, b"drivers")):
|
||||
image.parent.mkdir(parents=True)
|
||||
image.write_bytes(data)
|
||||
topology = {
|
||||
"topology": {
|
||||
"nodes": [
|
||||
{
|
||||
"node_type": "qemu",
|
||||
"properties": {
|
||||
"cdrom_image": str(installer),
|
||||
"cdrom2_image": str(drivers),
|
||||
},
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
Path(project.path, "test.gns3").write_text(json.dumps(topology))
|
||||
archive = str(tmpdir / "iso-project.zip")
|
||||
with aiozipstream.ZipFile() as stream:
|
||||
with patch.object(Qemu, "get_images_directory", return_value=str(root)):
|
||||
await export_project(stream, project, str(tmpdir), include_images=True)
|
||||
await write_file(archive, stream)
|
||||
with zipfile.ZipFile(archive) as stream:
|
||||
properties = json.loads(stream.read("project.gns3"))["topology"]["nodes"][0]["properties"]
|
||||
assert properties["cdrom_image"] != properties["cdrom2_image"]
|
||||
assert stream.read("images/QEMU/" + properties["cdrom_image"]) == b"installer"
|
||||
assert stream.read("images/QEMU/" + properties["cdrom2_image"]) == b"drivers"
|
||||
assert len(stream.namelist()) == len(set(stream.namelist()))
|
||||
destination = Path(str(tmpdir)) / "unpacked"
|
||||
stream.extractall(destination)
|
||||
controller = MagicMock()
|
||||
imported_images = Path(str(tmpdir)) / "imported-images"
|
||||
controller.images_path.return_value = str(imported_images)
|
||||
await _import_images(controller, str(destination / "images"))
|
||||
assert (imported_images / "QEMU" / properties["cdrom_image"]).read_bytes() == b"installer"
|
||||
assert (imported_images / "QEMU" / properties["cdrom2_image"]).read_bytes() == b"drivers"
|
||||
|
||||
@ -187,3 +187,53 @@ async def test_list_images(tmpdir, config):
|
||||
"path": "qemu_image.qcow2",
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("subdirectory", [None, "installers/linux"])
|
||||
async def test_iso_upload_and_discovery(tmp_path, config, subdirectory):
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
from gns3server.utils.images import discover_images, inspect_image_file, write_image
|
||||
|
||||
config.settings.Server.images_path = str(tmp_path)
|
||||
config.settings.Server.additional_images_paths = []
|
||||
# ISO 9660 primary volume descriptor; no raw-image fallback is needed.
|
||||
content = b"\0" * (16 * 2048) + b"\x01CD001\x01" + b"\0" * 2041
|
||||
|
||||
async def stream():
|
||||
# Exercise a descriptor split across upload chunks.
|
||||
for offset in range(0, len(content), 13):
|
||||
yield content[offset : offset + 13]
|
||||
|
||||
repo = AsyncMock()
|
||||
repo.get_image_by_checksum.return_value = None
|
||||
repo.save_verified_image.side_effect = lambda image: image
|
||||
result = await write_image(
|
||||
"installer.ISO", str(tmp_path / "installer.ISO"), stream(), repo, subdirectory=subdirectory
|
||||
)
|
||||
expected = tmp_path / "QEMU" / (subdirectory or "") / "installer.ISO"
|
||||
assert result["path"] == str(expected)
|
||||
assert result["image_type"] == "qemu"
|
||||
assert expected.read_bytes() == content
|
||||
assert inspect_image_file(expected)["checksum"] == result["checksum"]
|
||||
assert [image["path"] for image in await discover_images("qemu")] == [str(expected)]
|
||||
|
||||
|
||||
def test_iso_extension_alone_is_not_a_valid_image():
|
||||
from gns3server.utils.images import InvalidImageError, check_valid_image_header
|
||||
|
||||
with pytest.raises(InvalidImageError):
|
||||
check_valid_image_header("fake.iso", b"not an ISO image")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("images_root", ["images", "QEMU/images"])
|
||||
def test_image_reference_uses_configured_root(tmp_path, config, images_root):
|
||||
from gns3server.db.models.images import Image
|
||||
|
||||
root = tmp_path / images_root
|
||||
config.settings.Server.images_path = str(root)
|
||||
image = Image(image_type="qemu", path=str(root / "QEMU/installers/linux.iso"))
|
||||
assert image.reference == "installers/linux.iso"
|
||||
image.path = str(tmp_path / "extra/QEMU/linux.iso")
|
||||
assert image.reference == image.path
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user