Virtual_disk: add two nvme disk related blockcopy scenarios #6655
Conversation
WalkthroughAdds blockcopy test coverage for NVMe virtual disks and refactors disk parsing. The cfg replaces Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes
Areas to review closely:
Pre-merge checks and finishing touches✅ Passed checks (3 passed)
✨ Finishing touches
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (1)
libvirt/tests/src/virtual_disks/virtual_disks_nvme.py (1)
109-169: create_block_nvme_disk: overall logic OK; consider tightening error handlingThe NVMe provisioning logic is coherent:
lsblk+grep nvmeto enumerate candidates, thenlsblk <dev> | grep -q 'SWAP\|/'to skip mounted/in‑use devices.partedcreates a GPT and a 1G primary partition;nvme_partition_path = nvme_device + "p1"matches nvme naming.- Optional LUKS flow creates a libvirt secret and sets its value, then
prepare_disk_objis invoked withdisk_type=nvme_disk_typeand the partition path.- Return tuple
(disk_xml, nvme_device, nvme_partition_path, luks_sec_uuid)nicely exposes both XML and paths.A couple of small points you may want to refine:
- The
except Exceptionblock is very broad; in this test context it’s not dangerous, but if you care about debugging, catching a narrower set (e.g.process.CmdError,xcepts.LibvirtXMLError) or re‑raising after logging would make failures easier to interpret.- All shell commands here operate on fixed strings constructed in code (no user input), so
shell=Trueis acceptable for this harness; just keep it in mind if these ever start taking dynamic input.Functionally, the flow looks sound for the NVMe/LUKS setup.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
libvirt/tests/cfg/virtual_disks/virtual_disks_nvme.cfg(2 hunks)libvirt/tests/src/virtual_disks/virtual_disks_nvme.py(6 hunks)
🧰 Additional context used
🧠 Learnings (1)
📚 Learning: 2025-09-24T08:01:27.899Z
Learnt from: hholoubk
Repo: autotest/tp-libvirt PR: 6579
File: libvirt/tests/src/sriov/vIOMMU/iommu_device_lifecycle.py:95-97
Timestamp: 2025-09-24T08:01:27.899Z
Learning: In the libvirt test framework used in tp-libvirt, VM cleanup including destroying running VMs is handled by the teardown method (test_obj.teardown_iommu_test()) called in the finally block, so explicit VM destroy calls on timeout are not necessary according to the maintainers.
Applied to files:
libvirt/tests/src/virtual_disks/virtual_disks_nvme.py
🧬 Code graph analysis (1)
libvirt/tests/src/virtual_disks/virtual_disks_nvme.py (1)
provider/virtual_disk/disk_base.py (5)
prepare_disk_obj(114-250)DiskBase(25-633)add_vm_disk(95-112)cleanup_disk_preparation(268-301)get_source_list(56-93)
🪛 Ruff (0.14.4)
libvirt/tests/src/virtual_disks/virtual_disks_nvme.py
128-128: Function call with shell=True parameter identified, security issue
(S604)
136-136: Function call with shell=True parameter identified, security issue
(S604)
145-145: Function call with shell=True parameter identified, security issue
(S604)
150-150: Function call with shell=True parameter identified, security issue
(S604)
165-165: Consider moving this statement to an else block
(TRY300)
167-167: Do not catch blind exception: Exception
(BLE001)
246-246: Function call with shell=True parameter identified, security issue
(S604)
248-248: Local variable result is assigned to but never used
Remove assignment to unused variable result
(F841)
256-256: Function call with shell=True parameter identified, security issue
(S604)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (4)
- GitHub Check: Python 3.12
- GitHub Check: Python 3.11
- GitHub Check: Python 3.9
- GitHub Check: Python 3.8
🔇 Additional comments (4)
libvirt/tests/src/virtual_disks/virtual_disks_nvme.py (4)
1-1: Imports and extended docstring look consistent with new functionalityThe added imports (
ast,libvirt_version,libvirtfromvirttest.utils_test, anddisk_base) are all used in the new blockcopy flow and replace unsafeevalusage withast.literal_eval. The updatedrun()docstring clearly explains both the attach/detach and new blockcopy scenarios and matches the high‑level steps in the code.Also applies to: 13-21, 28-41
170-208: run_attach_detach_test refactor preserves original behaviorThe extracted
run_attach_detach_test()keeps the original attach/detach semantics:
- Ensures the VM is started to compute
old_partitions, then destroys it before reconfiguring XML.- Uses
get_usable_nvme_pci_address()andcreate_customized_disk()to build the NVMe device.- Handles both coldplug (
vmxml.add_device+ define) and hotplug (virsh.attach_device --live).- Waits for exactly one new disk and validates I/O with
libvirt_disk.check_virtual_disk_io.- Detaches the device via
virsh.detach_device.The refactor improves readability without changing logic; no issues seen.
295-323: Top-level run() flow and cleanup look goodThe main control flow:
- Backs up inactive VM XML (
vmxml_backup) and works on a copy.- Verifies required host tooling (
lspci) before proceeding.- Chooses between
run_blockcopy_test()andrun_attach_detach_test()based onwith_blockcopy.- Catches
VMStartErrorandLibvirtXMLErrorand fails the test with clear messages.- In
finally, ensures the VM is destroyed if still running and restores the original XML viavmxml_backup.sync().Given that
run_blockcopy_test()andrun_attach_detach_test()also handle their own short‑lived VM lifecycles as needed, this outer cleanup is a reasonable safety net and should restore the environment reliably after each test.
262-294: Update docstring to reflect actual parameter type; consider guarding[0]index access onget_source_listresultThe verification helper logic is sound. After checking the codebase:
Docstring parameter type mismatch — Confirmed
The parameter docstring says
nvme_dest_disk: NVME destination disk XML object, but the function is called with a string path (nvme_partition_pathat line 252) and used as a path in the comparison. Update:
:param nvme_dest_disk: NVME destination disk XML object used in blockcopy
:param nvme_dest_disk: NVME destination disk path used in blockcopy
Unguarded
[0]indexing onget_source_listresult — ConfirmedLines 282 and 289 index directly without verifying the list is non-empty. While you check
len(disk_sources) < 2on line 278, that usesget_disk_source(), notget_source_list(). If these methods diverge, anIndexErrorcould occur. Add a guard:source_list = disk_obj.get_source_list(current_vmxml, disk_type, target_dev) if not source_list: test.fail("No disk source found for target device...") current_disk_source = source_list[0]
get_disk_source(vm_name)on instance — Requires verificationLine 276 calls
current_vmxml.get_disk_source(vm_name)(instance method withvm_nameparameter). While this pattern appears elsewhere in the codebase, please verify the method signature is correct; the external library definition is not in this repository.
| def run_blockcopy_test(): | ||
| """ | ||
| Run blockcopy test with nvme destination | ||
| """ | ||
| libvirt_version.is_libvirt_feature_supported(params) | ||
| target_dev = params.get("target_dev", "vdb") | ||
| disk_type = params.get("disk_type", "file") | ||
| nvme_disk_type = params.get("nvme_disk_type", "block") | ||
| disk_dict = ast.literal_eval(params.get("disk_dict", "{}")) | ||
| nvme_disk_dict = ast.literal_eval(params.get("nvme_disk_dict", "{}")) | ||
| common_options = params.get("blockcopy_common_options") | ||
| specific_option = params.get("blockcopy_option", "") | ||
| with_luks = "yes" == params.get("with_luks", "no") | ||
| luks_extra_parameter = params.get("luks_extra_parameter", "") | ||
| disk_obj = disk_base.DiskBase(test, vm, params) | ||
|
|
||
| LOG.info("TEST_STEP1: Start VM.") | ||
| nvme_disk, nvme_device, nvme_partition_path, luks_sec_uuid = create_block_nvme_disk(params, | ||
| nvme_disk_type, | ||
| disk_obj, | ||
| nvme_disk_dict) | ||
| disk_info_dict = utils_misc.get_image_info(nvme_partition_path) | ||
| disk_size = disk_info_dict.get("vsize") | ||
| disk_size_bytes = str(disk_size) + "b" | ||
| original_disk_source = disk_obj.add_vm_disk(disk_type, disk_dict, size=disk_size_bytes) | ||
| LOG.debug("Original disk source: %s", original_disk_source) | ||
| if vm.is_dead(): | ||
| vm.start() | ||
| LOG.debug("The current guest xml is: %s" % (virsh.dumpxml(vm_name).stdout_text)) | ||
| vm.wait_for_login().close() | ||
|
|
||
| LOG.info("TEST_STEP2: Execute blockcopy command with option '%s'.", specific_option) | ||
| if with_luks: | ||
| cmd = ("qemu-img create -f qcow2 %s %s %s" % (luks_extra_parameter, | ||
| nvme_partition_path, | ||
| disk_size_bytes)) | ||
| process.run(cmd, shell=True, ignore_status=False) | ||
| options = "--xml %s %s %s" % (nvme_disk.xml, specific_option, common_options) | ||
| result = virsh.blockcopy(vm_name, target_dev, options, debug=True, ignore_status=False) | ||
| LOG.debug("Blockcopy completed successfully.") | ||
|
|
||
| LOG.info("TEST_STEP3: Verify blockcopy result in VM XML.") | ||
| verify_blockcopy_result(params, original_disk_source, nvme_partition_path) | ||
|
|
||
| if vm.is_alive(): | ||
| vm.destroy(gracefully=False) | ||
| process.run("sgdisk --zap-all %s" % nvme_device, shell=True, ignore_status=False) | ||
| disk_obj.cleanup_disk_preparation(disk_type) | ||
| if with_luks: | ||
| virsh.secret_undefine(luks_sec_uuid, debug=True, ignore_status=False) | ||
|
|
There was a problem hiding this comment.
🧩 Analysis chain
Align LUKS NVMe destination format with driver type and drop unused variable
The blockcopy flow is generally well-structured (version check, NVMe setup, original disk creation, VM start, blockcopy, verification, cleanup). Two concrete points here:
-
Format mismatch risk (ties to cfg change)
For LUKS:
if with_luks: cmd = ("qemu-img create -f qcow2 %s %s %s" % (luks_extra_parameter, nvme_partition_path, disk_size_bytes)) process.run(cmd, shell=True, ignore_status=False)
Combined with the cfg file,
nvme_disk_dict.driver.typeis currently${disk_image_format}, which defaults to"raw"invirtual_disks_nvme.cfg. That means:- Actual image on
nvme_partition_pathis qcow2 + LUKS. - Libvirt XML will advertise it as
<driver type='raw'>unlessdisk_image_formatis overridden.
This mismatch will cause blockcopy/pivot to fail when the LUKS NVMe destination is used. This is the other half of the issue flagged in the cfg review.
Suggested direction (coordinated with cfg):
- Ensure the driver type in
nvme_disk_dictmatches the created image format, e.g. by:- Setting
disk_image_format = "qcow2"for LUKS variants in the cfg, or - Introducing a dedicated
nvme_disk_image_formatparam used both here (-f) and innvme_disk_dict.driver.type.
- Setting
- Actual image on
-
Unused
resultvariableresult = virsh.blockcopy(vm_name, target_dev, options, debug=True, ignore_status=False) LOG.debug("Blockcopy completed successfully.")
The
resultis never inspected. Sinceignore_status=Falsealready lets the helper raise on failure, you can safely drop the assignment:
-
result = virsh.blockcopy(vm_name, target_dev, options, debug=True, ignore_status=False) -
LOG.debug("Blockcopy completed successfully.")
-
virsh.blockcopy(vm_name, target_dev, options, debug=True, ignore_status=False) -
LOG.debug("Blockcopy completed successfully.")This removes the Ruff `F841` warning without behavior change.
Overall structure is good; just make sure the format wiring is consistent.
I'll verify the review comment by examining the codebase and configuration files.
---
🏁 Script executed:
```shell
#!/bin/bash
# Locate the cfg file and related test files
fd -type f -name "*virtual_disks_nvme*"
fd -type f -name "virtual_disks_nvme.cfg"
Length of output: 341
🏁 Script executed:
#!/bin/bash
# Search for disk_image_format default value in cfg files
rg "disk_image_format" -A 3 -B 1 --type=cfg --type=yamlLength of output: 27
🏁 Script executed:
#!/bin/bash
# Inspect the reviewed file for result variable usage after blockcopy
rg -n "result.*blockcopy|result\s*=" libvirt/tests/src/virtual_disks/virtual_disks_nvme.py -A 5 -B 2Length of output: 1719
🏁 Script executed:
#!/bin/bash
# Search for create_block_nvme_disk function to understand nvme_disk structure
ast-grep --pattern 'function create_block_nvme_disk($_) { $$$ }'Length of output: 0
🏁 Script executed:
#!/bin/bash
# Alternative: search with rg for the function
rg "def create_block_nvme_disk" -A 30 --type=pyLength of output: 3396
---
🏁 Script executed:
```shell
#!/bin/bash
# Broad search for cfg files - remove type filter
find . -name "*.cfg" -type f 2>/dev/null | head -20
Length of output: 897
🏁 Script executed:
#!/bin/bash
# Search for disk_image_format in all files
rg "disk_image_format" -A 2 -B 1 -iLength of output: 8793
🏁 Script executed:
#!/bin/bash
# Find where nvme_disk is created and driver type set
rg "driver.*type|driver_type" libvirt/tests/src/virtual_disks/virtual_disks_nvme.py -B 3 -A 3Length of output: 0
🏁 Script executed:
#!/bin/bash
# Check the full context around the blockcopy call - see if result is used in verify function
rg "verify_blockcopy_result" -A 20 -B 5 libvirt/tests/src/virtual_disks/virtual_disks_nvme.py | head -40Length of output: 1832
🏁 Script executed:
#!/bin/bash
# Search for disk_obj.add_disk or similar methods to understand XML generation
rg "def.*disk.*xml|nvme_disk.*xml" libvirt/tests/src/virtual_disks/virtual_disks_nvme.py -B 2 -A 8Length of output: 592
---
🏁 Script executed:
```shell
#!/bin/bash
# Find nvme_disk object creation - search create_block_nvme_disk for disk XML generation
rg "create_block_nvme_disk" -A 80 libvirt/tests/src/virtual_disks/virtual_disks_nvme.py | grep -A 30 "disk_xml\|customized_disk\|nvme_disk"
Length of output: 7662
🏁 Script executed:
#!/bin/bash
# Specifically search for libvirt_disk.create_primitive_disk_xml call
rg "create_primitive_disk_xml" -B 5 -A 10 libvirt/tests/src/virtual_disks/virtual_disks_nvme.pyLength of output: 722
Fix LUKS NVMe destination format mismatch and remove unused variable
The blockcopy flow is well-structured, but two concrete issues need addressing:
-
Format mismatch causes blockcopy/pivot failure with LUKS
The code creates a qcow2+LUKS image when LUKS is enabled (lines 243–246):
if with_luks: cmd = ("qemu-img create -f qcow2 %s %s %s" % (luks_extra_parameter, nvme_partition_path, disk_size_bytes))
However, the nvme_disk XML is generated with driver type from
nvme_disk_dict, which has"type": "${disk_image_format}". Invirtual_disks_nvme.cfg,disk_image_format = "raw"by default, so the XML declares the destination asrawwhile the actual image isqcow2. This mismatch will cause blockcopy/pivot to fail.Fix: Ensure driver type in
nvme_disk_dictmatches the created image format:- Set
disk_image_format = "qcow2"for LUKS variants in the cfg, or - Introduce a dedicated
nvme_disk_image_formatparam and use it both in the qemu-img command (-f) and innvme_disk_dict.driver.type.
- Set
-
Remove unused
resultvariable (line 248)result = virsh.blockcopy(vm_name, target_dev, options, debug=True, ignore_status=False) LOG.debug("Blockcopy completed successfully.")
The assignment is never read. Since
ignore_status=Falseraises on failure, drop the assignment.
🧰 Tools
🪛 Ruff (0.14.4)
246-246: Function call with shell=True parameter identified, security issue
(S604)
248-248: Local variable result is assigned to but never used
Remove assignment to unused variable result
(F841)
256-256: Function call with shell=True parameter identified, security issue
(S604)
🤖 Prompt for AI Agents
libvirt/tests/src/virtual_disks/virtual_disks_nvme.py lines 210-260: the LUKS
branch creates a qcow2 image but the NVMe disk XML may still declare driver.type
as raw causing blockcopy/pivot failures, and the virsh.blockcopy return is
assigned to an unused variable; fix by ensuring the image format is consistent:
add/consume a parameter (e.g. nvme_disk_image_format) or set
nvme_disk_image_format = "qcow2" when with_luks and use that value both in the
qemu-img -f argument and to populate nvme_disk_dict.driver.type (so XML matches
the actual file format), and remove the unused assignment by calling
virsh.blockcopy(...) without assigning its return value.
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check if the unused result variable has been removed from blockcopy call
rg -n "virsh\.blockcopy.*result|result.*=.*virsh\.blockcopy" libvirt/tests/src/virtual_disks/virtual_disks_nvme.py -B 2 -A 2
Length of output: 0
🏁 Script executed:
#!/bin/bash
# Check the LUKS handling section to see if format consistency is addressed
rg -n "if with_luks:" libvirt/tests/src/virtual_disks/virtual_disks_nvme.py -A 10
Length of output: 1866
🏁 Script executed:
#!/bin/bash
# Check the cfg file for any format-related changes in LUKS variants
rg -n "with_luks|disk_image_format|nvme_disk_image_format" libvirt/tests/cfg/virtual_disks/virtual_disks_nvme.cfg -B 2 -A 3
Length of output: 1679
🏁 Script executed:
#!/bin/bash
# Check create_block_nvme_disk to see if format parameters are used
rg -n "def create_block_nvme_disk" libvirt/tests/src/virtual_disks/virtual_disks_nvme.py -A 50 | grep -E "disk_image_format|nvme_disk_image_format|format"
Length of output: 0
@meinaLi Thank you for the update! I can confirm that the unused result variable has been successfully removed - great work on that fix! ✓
However, I noticed that the LUKS format mismatch issue still needs attention. Here's what I found:
Current situation:
-
In the Python code (line 243), the LUKS branch creates a qcow2 format image:
cmd = ("qemu-img create -f qcow2 %s %s %s" % (luks_extra_parameter, ...))
-
But in the cfg file (line 37-39), the
with_luksvariant sets:disk_image_format = "raw" nvme_disk_dict = {..., "driver": {"name": "qemu", "type": "${disk_image_format}"}} -
This means the XML will declare
<driver type='raw'>while the actual image isqcow2, causing blockcopy/pivot to fail.
Suggested fix:
Change line 37 in virtual_disks_nvme.cfg from:
disk_image_format = "raw"to:
disk_image_format = "qcow2"This will ensure the driver type in the XML matches the actual image format created by qemu-img, allowing blockcopy operations to succeed with LUKS encryption.
🧬 Code Graph Analysis Results
[filename: libvirt/tests/src/virtual_disks/virtual_disks_nvme.py]
[Possible relevant snippets to understand the file and handle the user comment "Finish to update."]
-
File: provider/virtual_disk/disk_base.py
-
Line range: 95-112
-
Content:
def add_vm_disk(self, disk_type, disk_dict, new_image_path='', **kwargs):
"""
Add vm a disk.:param disk_type: disk type :param disk_dict: dict to update disk :param new_image_path: new disk image if needed :return new_image_path: return the updated new image path of new disk """ vmxml = vm_xml.VMXML.new_from_dumpxml(self.vm.name) dest_obj, new_image_path = self.prepare_disk_obj( disk_type, disk_dict, new_image_path, **kwargs) libvirt.add_vm_device(vmxml, dest_obj) return new_image_path
-
-
File: provider/virtual_disk/disk_base.py
-
Line range: 56-93
-
Content:
@staticmethod
def get_source_list(vmxml, disk_type, target_dev):
"""
Get vmxml all source list with specific device, include backingstore
and source path:param vmxml: vm xml :param disk_type: disk type :param target_dev: dev of target disk :return source list, include backingstore and source path """ source_list = [] disks = vmxml.devices.by_device_tag('disk') for disk in disks: if disk.target['dev'] == target_dev: active_level_path = \ disk.xmltreefile.find('source').get('file') \ or disk.xmltreefile.find('source').get('dev') \ or disk.xmltreefile.find('source').get('volume') \ or disk.xmltreefile.find('source').get('name') backing_list = disk.get_backingstore_list() if disk_type != 'rbd_with_auth' and backing_list != []: backing_list.pop() for elem in backing_list: source_ele = elem.find("source") if source_ele is not None: source_list.append(source_ele.get('file') or source_ele.get('name') or source_ele.get('dev') or source_ele.get('volume')) if source_ele.find("dataStore"): source_list.append( source_ele.find("dataStore").find('source').get('file')) source_list.insert(0, active_level_path) break return source_list
-
-
File: provider/virtual_disk/disk_base.py
-
Line range: 114-250
-
Content:
def prepare_disk_obj(self, disk_type, disk_dict, new_image_path='', **kwargs):
"""
Prepare disk object:param disk_type: disk type :param disk_dict: dict to update disk :param new_image_path: new disk image if needed :param kwargs: optional keyword arguments. :return disk_obj: return disk object :return new_image_path: return the updated new image path of new disk """ if disk_type == 'file': if not new_image_path: new_image_path = data_dir.get_data_dir() + '/test.img' size = kwargs.get("size", "50M") if kwargs.get("size"): kwargs.pop("size") libvirt.create_local_disk( "file", path=new_image_path, size=size, disk_format=self.params.get("disk_image_format", 'qcow2'), **kwargs) disk_dict.update({'source': {'attrs': {'file': new_image_path}}}) elif disk_type == 'block': if not new_image_path: new_image_path = self.create_lvm_disk_path( vg_name=self.vg_name, lv_name=self.lv_name, simulated_iscsi=self.params.get("simulated_iscsi"), **kwargs) if self.params.get('convert_format'): process.run("qemu-img create -f %s /dev/%s/%s %s" % ( self.params.get('convert_format'), self.vg_name, self.lv_name, self.disk_size), shell=True) disk_dict.update({'source': {'attrs': {'dev': new_image_path}}}) elif disk_type == 'volume': if not new_image_path: pool_name, new_image_path = self.create_volume_for_disk_path( self.test, self.params, self.pool_name, self.pool_type, self.pool_target, self.emulated_image, **kwargs) disk_dict.update({'source': {'attrs': {'pool': self.pool_name, 'volume': new_image_path}}}) elif disk_type == 'nfs': if not new_image_path: nfs_res = libvirt.setup_or_cleanup_nfs(is_setup=True, is_mount=True) new_image_path = nfs_res['mount_dir'] + '/test.img' libvirt.create_local_disk("file", path=new_image_path, size=self.disk_size, disk_format="qcow2", **kwargs) disk_dict.update({'source': {'attrs': {'file': new_image_path}}}) elif disk_type == 'rbd_with_auth': no_update_dict = kwargs.get("no_update_dict") if no_update_dict: pass else: mon_host, auth_username, new_image_path, _ = \ self.create_rbd_disk_path(self.params) disk_dict.update({'source': { 'attrs': {'protocol': "rbd", "name": new_image_path}, "hosts": [{"name": mon_host}], "auth": {"auth_user": auth_username, "secret_usage": "cephlibvirt", "secret_type": "ceph"}}}) elif disk_type == 'rbd_with_luks_and_auth': mon_host, auth_username, new_image_path, _ = \ self.create_rbd_disk_path(self.params, create_img_by_qemu_cmd=True) disk_dict.update({'source': { 'attrs': {'protocol': "rbd", "name": new_image_path}, "hosts": [{"name": mon_host}], "auth": {"auth_user": auth_username, "secret_usage": self.params.get("ceph_usage_name"), "secret_type": "ceph"}}}) self.add_luks_encryption_to_disk_dict( disk_dict, self.params.get("luks_sec_dict") % new_image_path) elif disk_type == "nbd": nbd_server_host = process.run('hostname', ignore_status=False, shell=True, verbose=True).stdout_text.strip() nbd_server_port = self.params.get("nbd_server_port", "10001") image_path = self.params.get( "image_path", data_dir.get_data_dir() + "/images/nbdtest.img") device_format = self.params.get("device_format", "qcow2") export_name = self.params.get("export_name", "new") tls_enabled = "yes" == self.params.get("enable_tls", "yes") image_size = kwargs.get("size", self.params.get("image_size", "500M")) self.disk_backend = NbdExport( image_path, image_format=device_format, port=nbd_server_port, image_size=image_size, export_name=export_name, tls=tls_enabled) self.disk_backend.start_nbd_server() utils_misc.wait_for( lambda: process.system('netstat -nlp | grep %s ' % nbd_server_port, ignore_status=True, shell=True) == 0, 10) disk_dict.update( {'source': {"attrs": {"protocol": "nbd", "tls": self.params.get("enable_tls", "yes"), "name": export_name}, "hosts": [{"name": nbd_server_host, "port": nbd_server_port}]}, }) elif disk_type == "ssh_with_key": new_image_path, known_hosts_path, keyfile = \ self.create_ssh_disk_path(self.params, **kwargs) ssh_user = os.environ.get("USER") disk_dict.update( {'source': {"attrs": {"protocol": "ssh", "name": new_image_path}, "identity": [{"username": "%s" % ssh_user, "keyfile": keyfile}], "knownhosts": known_hosts_path, "hosts": [{"name": "127.0.0.1", "port": "22"}]}}) elif disk_type == "ssh_with_passwd": new_image_path, known_hosts_path, _ = \ self.create_ssh_disk_path(self.params, **kwargs) disk_dict.update( {'source': {"attrs": {"protocol": "ssh", "name": new_image_path}, "knownhosts": known_hosts_path, "hosts": [{"name": "127.0.0.1", "port": "22"}], "auth": {"auth_user": self.params.get("iscsi_user"), "secret_usage": "libvirtiscsi", "secret_type": "iscsi"}}}) disk_obj = libvirt_vmxml.create_vm_device_by_type("disk", disk_dict) self.test.log.debug('Create disk object is :%s', disk_obj) return disk_obj, new_image_path
-
-
File: provider/virtual_disk/disk_base.py
-
Line range: 268-301
-
Content:
def cleanup_disk_preparation(self, disk_type, **kwargs):
"""
Clean up the preparation of different type disk:param disk_type: disk type """ if disk_type == 'block': self.cleanup_block_disk_preparation( self.vg_name, self.lv_name, simulated_iscsi=self.params.get("simulated_iscsi")) elif disk_type == 'file': if os.path.exists(self.new_image_path): os.remove(self.new_image_path) elif disk_type == 'volume': pvt = libvirt.PoolVolumeTest(self.test, self.params) pvt.cleanup_pool(self.pool_name, self.pool_type, self.pool_target, self.emulated_image) elif disk_type == 'nfs': libvirt.setup_or_cleanup_nfs(is_setup=False) elif disk_type in ["rbd_with_auth", "rbd_with_luks_and_auth"]: self.cleanup_rbd_disk_path(self.params) elif disk_type == "nbd": try: self.disk_backend.cleanup() except Exception as ndbEx: self.test.log.debug("Clean Up nbd failed: %s", str(ndbEx)) elif disk_type == "ssh_with_key" or disk_type == "ssh_with_passwd": self.cleanup_ssh_disk_path(self.new_image_path, **kwargs)
-
-
File: provider/virtual_disk/disk_base.py
-
Line range: 323-? (as per 25-633 block)
-
Content:
@staticmethod
def create_pool_from_xml(pool_dict, source_dict):
"""
Build pool from xml:param pool_dict: pool dict, :param source_dict: pool source dict """ device_name = libvirt.setup_or_cleanup_iscsi(is_setup=True) source_dict['device_path'] = device_name pool_name = pool_dict['name'] pool = pool_xml.PoolXML() pool.setup_attrs(**pool_dict) source_xml = pool_xml.SourceXML() source_xml.device_path = device_name source_xml.vg_name = "vg0" source_xml.format_type = "lvm2" pool.source = source_xml virsh.pool_define(pool.xml, debug=True, ignore_status=False) virsh.pool_build(pool_name, options='--overwrite', debug=True, ignore_status=False) virsh.pool_start(pool_name, debug=True, ignore_status=False) return pool_name
-
-
File: provider/virtual_disk/disk_base.py
- Line range: 25-633
- Content:
(Includes the static method create_pool_from_xml and other related helpers described above, plus related methods such as create_lvm_disk_path, cleanup_ssh_disk_path, create_rbd_disk_path, create_ssh_disk_path, etc.)
[Note: This block encompasses multiple static/class methods related to pool creation, SSH/RBD disk path preparation, and related utilities referenced by the other snippets above.]
-
File: provider/virtual_disk/disk_base.py
- Line range: 56-93 (duplicate listing above) for reference to get_source_list
- See item 2 above.
If there are no other clearly more relevant snippets beyond these disk manipulation helpers, these are the primary constructs that relate to adding/updating disks, NVMe/block devices, and VM disk sources, which would help in finishing an update to the NVMe-related tests.
There was a problem hiding this comment.
Yes, I mistakenly thought I had made changes. This version is fine now.
There was a problem hiding this comment.
Seems like the humans are having a chat. I'll hop back into my burrow for now. If you need me again, just tag @coderabbitai in a new comment, and I'll come hopping out!
Automate case: RHEL-203952 - Blockcopy disk to nvme destination with luks(with usage) RHEL-241215 - Blockcopy disk to nvme destination without luks Signed-off-by: Meina Li <meili@redhat.com>
|
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (1)
libvirt/tests/src/virtual_disks/virtual_disks_nvme.py (1)
42-61: Minor clarity and efficiency tweaks (docstring, helper reuse)A few small refinements to make the test easier to maintain:
verify_blockcopy_result’s docstring currently statesnvme_dest_diskis an XML object; in reality the caller passes the NVMe destination path. Updating the docstring (and possibly renaming the parameter tonvme_dest_path) would better match actual usage.In
run_attach_detach_test,get_new_disks(old_partitions)is evaluated twice in succession:utils_misc.wait_for(lambda: get_new_disks(old_partitions), 20) new_disks = get_new_disks(old_partitions)You could avoid the second guest login by having
wait_forreturn the new partitions (e.g., store the result it returns or wrap it in a small helper), or by caching insideget_new_disksif that fits your style.Overall branching on
with_blockcopyat the end ofrun()is clear; once the LUKS and NVMe selection issues are addressed, this split between attach/detach and blockcopy flows is straightforward.These are non-blocking, but would make the test logic a bit cleaner.
Also applies to: 170-201, 261-268, 294-313
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
libvirt/tests/cfg/virtual_disks/virtual_disks_nvme.cfg(2 hunks)libvirt/tests/src/virtual_disks/virtual_disks_nvme.py(6 hunks)
🧰 Additional context used
🧠 Learnings (2)
📚 Learning: 2025-10-24T06:35:25.679Z
Learnt from: meinaLi
Repo: autotest/tp-libvirt PR: 6628
File: libvirt/tests/src/virtual_disks/virtual_disks_io_error.py:128-129
Timestamp: 2025-10-24T06:35:25.679Z
Learning: In libvirt tests, `virsh.snapshot_delete()` without the `--metadata` flag automatically deletes both snapshot metadata and external disk files. Explicit `os.remove()` is only needed when using `virsh.snapshot_delete()` with the `--metadata` flag, which removes only metadata.
Applied to files:
libvirt/tests/src/virtual_disks/virtual_disks_nvme.py
📚 Learning: 2025-09-24T08:01:27.899Z
Learnt from: hholoubk
Repo: autotest/tp-libvirt PR: 6579
File: libvirt/tests/src/sriov/vIOMMU/iommu_device_lifecycle.py:95-97
Timestamp: 2025-09-24T08:01:27.899Z
Learning: In the libvirt test framework used in tp-libvirt, VM cleanup including destroying running VMs is handled by the teardown method (test_obj.teardown_iommu_test()) called in the finally block, so explicit VM destroy calls on timeout are not necessary according to the maintainers.
Applied to files:
libvirt/tests/src/virtual_disks/virtual_disks_nvme.py
🧬 Code graph analysis (1)
libvirt/tests/src/virtual_disks/virtual_disks_nvme.py (1)
provider/virtual_disk/disk_base.py (4)
prepare_disk_obj(114-250)add_vm_disk(95-112)cleanup_disk_preparation(268-301)get_source_list(56-93)
🪛 Ruff (0.14.4)
libvirt/tests/src/virtual_disks/virtual_disks_nvme.py
128-128: Function call with shell=True parameter identified, security issue
(S604)
136-136: Function call with shell=True parameter identified, security issue
(S604)
145-145: Function call with shell=True parameter identified, security issue
(S604)
150-150: Function call with shell=True parameter identified, security issue
(S604)
165-165: Consider moving this statement to an else block
(TRY300)
167-167: Do not catch blind exception: Exception
(BLE001)
246-246: Function call with shell=True parameter identified, security issue
(S604)
256-256: Function call with shell=True parameter identified, security issue
(S604)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (4)
- GitHub Check: Python 3.9
- GitHub Check: Python 3.8
- GitHub Check: Python 3.12
- GitHub Check: Python 3.11
🔇 Additional comments (2)
libvirt/tests/cfg/virtual_disks/virtual_disks_nvme.cfg (1)
6-39: Blockcopy NVMe/LUKS cfg wiring looks consistent with Python implementationThe
disk_image_formatdefault and thewith_luksoverride to"qcow2"align with the Python path that creates a qcow2+LUKS NVMe destination, and bothdisk_dictandnvme_disk_dictcorrectly pick up${disk_image_format}. The newblockcopy_nvme/finish/pivotvariants and NVMe type separation (disk_type="file",nvme_disk_type="block") look coherent withrun_blockcopy_test()andcreate_block_nvme_disk().libvirt/tests/src/virtual_disks/virtual_disks_nvme.py (1)
94-101: Safe use of disk_image_format and ast.literal_evalUsing
params.get("disk_image_format")andast.literal_eval(params.get("source_attrs", "{}"))removes the earlierliteral_evalTypeError risk and keeps parsing constrained to literal structures. This is a solid, minimal change.
| luks_secret_passwd = params.get("luks_secret_passwd", "redhat") | ||
| with_luks = "yes" == params.get("with_luks", "no") | ||
| luks_sec_uuid = None | ||
| try: | ||
| LOG.info("SETUP_TEST: Getting available NVME device in host") | ||
| lsblk_result = process.run("lsblk -d -n -o NAME | grep nvme", | ||
| timeout=10, ignore_status=True, shell=True) | ||
| if lsblk_result.exit_status != 0: | ||
| test.cancel("No NVME device found in host") | ||
|
|
||
| # Backup vm xml | ||
| vmxml_backup = vm_xml.VMXML.new_from_inactive_dumpxml(vm_name) | ||
| vmxml = vmxml_backup.copy() | ||
| nvme_devices = lsblk_result.stdout_text.strip().split('\n') | ||
| for device in nvme_devices: | ||
| device_path = "/dev/" + device.strip() | ||
| mount_check = process.run(f"lsblk {device_path} | grep -q SWAP\\|/", | ||
| ignore_status=True, shell=True) | ||
| if mount_check.exit_status != 0: | ||
| nvme_device = device_path | ||
| break | ||
| else: | ||
| test.cancel("No available unmounted NVME device found") | ||
|
|
||
| try: | ||
| # install essential package in host | ||
| if not shutil.which('lspci'): | ||
| test.error("package {} not installed, please install them before testing".format(pkgs_host)) | ||
| LOG.info("SETUP_TEST: Creating GPT partition table on NVME device") | ||
| result = process.run("parted -s %s mklabel gpt" % nvme_device, | ||
| timeout=30, ignore_status=False, verbose=True, shell=True) | ||
| LOG.debug("GPT partition table creation result: %s", result) | ||
|
|
||
| LOG.info("SETUP_TEST: Creating 1G primary partition on NVME device") | ||
| result = process.run("parted -s %s mkpart primary 0%% 1G" % nvme_device, | ||
| timeout=30, ignore_status=False, verbose=True, shell=True) | ||
| LOG.debug("Partition creation result: %s", result) | ||
| time.sleep(2) | ||
|
|
||
| # Use the created partition as disk image | ||
| nvme_partition_path = nvme_device + "p1" | ||
| if with_luks: | ||
| luks_sec_uuid = libvirt.create_secret(params) | ||
| ret = virsh.secret_set_value(luks_sec_uuid, luks_secret_passwd, | ||
| encode=True, debug=True) | ||
| libvirt.check_exit_status(ret) | ||
|
|
||
| disk_xml, _ = disk_obj.prepare_disk_obj(disk_type, disk_dict, nvme_partition_path) | ||
| LOG.debug("Created disk XML: %s", disk_xml) | ||
| return disk_xml, nvme_device, nvme_partition_path, luks_sec_uuid | ||
|
|
There was a problem hiding this comment.
🧩 Analysis chain
Ensure LUKS NVMe destination XML preserves encryption and secret usage
In the LUKS variant, the cfg defines nvme_disk_dict with a source.encryption structure and disk_image_format="qcow2", while the Python path does:
with_luks = "yes" == params.get("with_luks", "no")
...
disk_xml, _ = disk_obj.prepare_disk_obj(disk_type, disk_dict, nvme_partition_path)
...
if with_luks:
cmd = ("qemu-img create -f qcow2 %s %s %s"
% (luks_extra_parameter, nvme_partition_path, disk_size_bytes))
process.run(cmd, shell=True, ignore_status=False)
...
verify_blockcopy_result(..., nvme_partition_path)Given disk_type for the NVMe destination is "block" and disk_base.prepare_disk_obj’s block branch does:
disk_dict.update({'source': {'attrs': {'dev': new_image_path}}})any pre-populated disk_dict["source"]["encryption"] from nvme_disk_dict will be overwritten. That means:
- The disk XML used in
--xmlwill likely not carry<encryption>and<secret>elements, even though a qcow2+LUKS image and secret are created. - QEMU/libvirt won’t have a key source for this encrypted image during blockcopy, leading to failure or unexpected behavior rather than a correct LUKS NVMe blockcopy test.
Consider adjusting this path so that encryption metadata survives:
- Either modify
create_block_nvme_diskto merge thedevattribute into any existingsourcestructure instead of replacing it (e.g., updatesource.attrs.devon the existing dict, keepingsource.encryptionintact before buildingdisk_xml), or - Extend
DiskBase.prepare_disk_obj/ its call-site with a flag similar to therbd_with_luks_and_authpath (which usesadd_luks_encryption_to_disk_dict) so that block NVMe + LUKS can reuse the same pattern without losing encryption info.
Also, in verify_blockcopy_result, the parameter nvme_dest_disk is documented as an “NVME destination disk XML object” but is actually passed a path string (nvme_partition_path). Updating the docstring (and variable name if desired) to reflect that it is a path, not an XML object, would avoid confusion when reasoning about the pivot check:
elif "--pivot" in specific_option:
current_disk_source = disk_obj.get_source_list(current_vmxml, nvme_disk_type, target_dev)[0]
if current_disk_source != nvme_dest_disk:
...Also applies to: 223-224, 241-249, 261-293
---
🏁 Script executed:
```shell
#!/bin/bash
# Find the DiskBase class and prepare_disk_obj method
find . -name "*.py" -type f | head -20 | xargs grep -l "class.*DiskBase"
Length of output: 0
🏁 Script executed:
#!/bin/bash
# Search for prepare_disk_obj method implementation
rg -n "def prepare_disk_obj" --type=py -A 30Length of output: 5229
🏁 Script executed:
#!/bin/bash
# Search for the specific file under review
find . -name "virtual_disks_nvme.py" -type fLength of output: 55
🏁 Script executed:
#!/bin/bash
# Search for verify_blockcopy_result function
rg -n "def verify_blockcopy_result" --type=py -A 20Length of output: 2345
🏁 Script executed:
#!/bin/bash
# Search for add_luks_encryption_to_disk_dict pattern mentioned in review
rg -n "add_luks_encryption_to_disk_dict" --type=py -B 2 -A 5Length of output: 1311
🏁 Script executed:
#!/bin/bash
# Check the specific lines mentioned in the review (223-224, 241-249, 261-293)
find . -name "virtual_disks_nvme.py" -type f -exec sed -n '220,300p' {} \;Length of output: 4568
🏁 Script executed:
#!/bin/bash
# Find create_block_nvme_disk function mentioned in review
rg -n "def create_block_nvme_disk" --type=py -A 20Length of output: 2412
---
🏁 Script executed:
```shell
#!/bin/bash
# Get the complete block branch implementation from disk_base.py
sed -n '137,160p' ./provider/virtual_disk/disk_base.py
Length of output: 1374
🏁 Script executed:
#!/bin/bash
# Get the complete prepare_disk_obj method to see how different disk types handle source dict
sed -n '114,250p' ./provider/virtual_disk/disk_base.pyLength of output: 7080
🏁 Script executed:
#!/bin/bash
# Check the add_luks_encryption_to_disk_dict implementation
sed -n '252,270p' ./provider/virtual_disk/disk_base.pyLength of output: 835
🏁 Script executed:
#!/bin/bash
# Check the file branch to compare with block branch
sed -n '125,136p' ./provider/virtual_disk/disk_base.pyLength of output: 555
🏁 Script executed:
#!/bin/bash
# Search for rbd_with_luks_and_auth pattern mentioned in review
rg -n "rbd_with_luks_and_auth\|rbd" ./provider/virtual_disk/disk_base.py -B 3 -A 10 | head -60Length of output: 0
🏁 Script executed:
#!/bin/bash
# Check how disk_dict is structured in the NVME test - look for nvme_disk_dict definition
rg -n "nvme_disk_dict" ./libvirt/tests/src/virtual_disks/virtual_disks_nvme.py -B 2 -A 5Length of output: 1234
---
🏁 Script executed:
```shell
#!/bin/bash
# Find test configuration files that might define nvme_disk_dict with encryption
find . -name "*.cfg" -o -name "*.conf" | xargs grep -l "nvme_disk_dict\|source.*encryption" 2>/dev/null | head -10
Length of output: 270
🏁 Script executed:
#!/bin/bash
# Search for test configs in the virtual_disks directory
find ./libvirt/tests/cfg -name "*nvme*" -o -name "*virtual_disk*" | head -20Length of output: 1250
🏁 Script executed:
#!/bin/bash
# Look specifically at lines 223-224 and surrounding context
sed -n '220,230p' ./libvirt/tests/src/virtual_disks/virtual_disks_nvme.pyLength of output: 770
🏁 Script executed:
#!/bin/bash
# Look at lines 241-249
sed -n '238,255p' ./libvirt/tests/src/virtual_disks/virtual_disks_nvme.pyLength of output: 984
🏁 Script executed:
#!/bin/bash
# Search for where nvme_disk_dict gets encryption configuration
rg -n "source.*encryption\|encryption.*source" ./libvirt/tests/src/virtual_disks/ --type=py -B 3 -A 3Length of output: 0
🏁 Script executed:
#!/bin/bash
# Check if qcow2+LUKS creation happens but encryption is not in XML
sed -n '241,249p' ./libvirt/tests/src/virtual_disks/virtual_disks_nvme.pyLength of output: 639
---
🏁 Script executed:
```shell
#!/bin/bash
# Check the virtual_disks_nvme.cfg for nvme_disk_dict definition with encryption
cat ./libvirt/tests/cfg/virtual_disks/virtual_disks_nvme.cfg | head -100
Length of output: 2226
🏁 Script executed:
#!/bin/bash
# Look for nvme_disk_dict that might contain encryption in the config
grep -n "nvme_disk_dict\|encryption\|source" ./libvirt/tests/cfg/virtual_disks/virtual_disks_nvme.cfg -B 2 -A 2 | head -60Length of output: 1544
🏁 Script executed:
#!/bin/bash
# Get full context around create_block_nvme_disk to understand the flow
sed -n '109,175p' ./libvirt/tests/src/virtual_disks/virtual_disks_nvme.pyLength of output: 3400
🏁 Script executed:
#!/bin/bash
# Verify exact line numbers for the docstring issue in verify_blockcopy_result
sed -n '261,293p' ./libvirt/tests/src/virtual_disks/virtual_disks_nvme.pyLength of output: 1886
Preserve LUKS encryption metadata in block NVME disk XML
The test config for with_luks variant (line 39 of virtual_disks_nvme.cfg) defines nvme_disk_dict with a pre-populated source.encryption structure:
"source": {"encryption": {"attrs": {"format": "luks"}, "encryption": "luks", "secret": {"type": "passphrase", "usage": "${sec_volume}"}}}
However, when create_block_nvme_disk (line 168) calls disk_obj.prepare_disk_obj(disk_type="block", nvme_disk_dict, nvme_partition_path), the block branch in DiskBase.prepare_disk_obj (disk_base.py line 147) replaces the entire source dict:
disk_dict.update({'source': {'attrs': {'dev': new_image_path}}})This overwrites any pre-populated encryption metadata. Later, when the qcow2+LUKS image is created (lines 241–248) and blockcopy is executed with nvme_disk.xml (line 249), QEMU/libvirt lacks encryption metadata, causing the blockcopy to fail or behave unexpectedly.
The rbd_with_luks_and_auth pattern in disk_base.py (lines 173–190) demonstrates the correct approach: it calls add_luks_encryption_to_disk_dict after setting source, which preserves encryption using an .update() call rather than replacement (disk_base.py lines 263–266).
Fixes:
- Modify
create_block_nvme_diskto preserve encryption: either updatesource.attrs.devwithout replacing the dict, or useadd_luks_encryption_to_disk_dictafterprepare_disk_obj. - Update the docstring for
verify_blockcopy_result(line 267): parameternvme_dest_diskis documented as "NVME destination disk XML object" but is actually a path string (passed asnvme_partition_pathat line 252 and used as a string at line 289).
🧰 Tools
🪛 Ruff (0.14.4)
128-128: Function call with shell=True parameter identified, security issue
(S604)
136-136: Function call with shell=True parameter identified, security issue
(S604)
145-145: Function call with shell=True parameter identified, security issue
(S604)
150-150: Function call with shell=True parameter identified, security issue
(S604)
165-165: Consider moving this statement to an else block
(TRY300)
🤖 Prompt for AI Agents
In libvirt/tests/src/virtual_disks/virtual_disks_nvme.py around lines 123-166,
the block NVME disk creation overwrites any pre-populated source.encryption
metadata when calling DiskBase.prepare_disk_obj, causing LUKS info to be lost
and blockcopy to fail; change create_block_nvme_disk so it preserves encryption
by either (A) setting only source.attrs.dev (i.e., update
disk_dict['source']['attrs']['dev'] = nvme_partition_path) instead of replacing
the whole source dict before calling prepare_disk_obj, or (B) call
disk_obj.add_luks_encryption_to_disk_dict(disk_dict, luks_secret_passwd,
luks_sec_uuid) immediately after prepare_disk_obj so encryption is merged via
.update() as done for rbd_with_luks_and_auth; also fix the
verify_blockcopy_result docstring to state nvme_dest_disk is a path string
(nvme_partition_path) not an XML object.
| lsblk_result = process.run("lsblk -d -n -o NAME | grep nvme", | ||
| timeout=10, ignore_status=True, shell=True) | ||
| if lsblk_result.exit_status != 0: | ||
| test.cancel("No NVME device found in host") | ||
|
|
||
| # Backup vm xml | ||
| vmxml_backup = vm_xml.VMXML.new_from_inactive_dumpxml(vm_name) | ||
| vmxml = vmxml_backup.copy() | ||
| nvme_devices = lsblk_result.stdout_text.strip().split('\n') | ||
| for device in nvme_devices: | ||
| device_path = "/dev/" + device.strip() | ||
| mount_check = process.run(f"lsblk {device_path} | grep -q SWAP\\|/", | ||
| ignore_status=True, shell=True) | ||
| if mount_check.exit_status != 0: | ||
| nvme_device = device_path |
There was a problem hiding this comment.
🧩 Analysis chain
Fix NVMe “unused” detection to avoid wiping in-use host disks
The logic:
lsblk_result = process.run("lsblk -d -n -o NAME | grep nvme", ...)
...
mount_check = process.run(
f"lsblk {device_path} | grep -q SWAP\\|/", ignore_status=True, shell=True)
if mount_check.exit_status != 0:
nvme_device = device_path
break
...
process.run("sgdisk --zap-all %s" % nvme_device, shell=True, ignore_status=False)has two coupled problems:
grep -q SWAP\|/under basicgrepdoes not implement alternation; it looks for the literal stringSWAP|/, which will never match typicallsblkoutput. As a result,mount_check.exit_statusis effectively always non‑zero, so the first NVMe device is treated as “unused” even if it holds mounted filesystems or swap.- The subsequent
sgdisk --zap-allon that device can therefore operate on an in-use NVMe disk (e.g., root or data disk), which is extremely destructive.
You should reliably detect any mounted/swap usage and only pick genuinely unused devices (or make the device explicit via params). For example:
- mount_check = process.run(f"lsblk {device_path} | grep -q SWAP\\|/",
- ignore_status=True, shell=True)
- if mount_check.exit_status != 0:
+ # Treat any non-empty mountpoint (including SWAP) as "in use"
+ mount_check = process.run(
+ f"lsblk -n -o MOUNTPOINT {device_path} | grep -qE '\\S'",
+ ignore_status=True, shell=True)
+ if mount_check.exit_status != 0:
nvme_device = device_path
breakand keep sgdisk --zap-all restricted to devices that have passed this stricter check (or additionally guard via a whitelist/parameter).
Also applies to: 256-257
I'll help you verify this review comment. Let me examine the actual code in the repository to confirm the issues described.
Now let me examine the actual code in the repository:
---
🏁 Script executed:
```shell
#!/bin/bash
# View the actual code at the specified lines
if [ -d "libvirt" ]; then
cat libvirt/tests/src/virtual_disks/virtual_disks_nvme.py | sed -n '120,160p'
else
echo "Repository structure check:"
find . -name "virtual_disks_nvme.py" -type f 2>/dev/null | head -5
fi
Length of output: 2238
🏁 Script executed:
#!/bin/bash
# Also check lines around 256-257 for similar issues
if [ -f "libvirt/tests/src/virtual_disks/virtual_disks_nvme.py" ]; then
echo "=== Lines 250-265 ==="
sed -n '250,265p' libvirt/tests/src/virtual_disks/virtual_disks_nvme.py
fiLength of output: 687
🏁 Script executed:
#!/bin/bash
# Search for sgdisk calls in the file
rg -n "sgdisk" libvirt/tests/src/virtual_disks/virtual_disks_nvme.pyLength of output: 93
🏁 Script executed:
#!/bin/bash
# Get context around grep patterns used in the file
rg -n "grep.*SWAP" -A3 -B3 libvirt/tests/src/virtual_disks/virtual_disks_nvme.pyLength of output: 448
🏁 Script executed:
#!/bin/bash
# Check lsblk output to understand what the grep is actually matching against
# Show what lsblk -d -n -o NAME outputs and what lsblk on a device outputs
cat <<'EOF'
=== Understanding lsblk output formats ===
lsblk -d -n -o NAME outputs:
nvme0n1
sda
...
lsblk /dev/nvme0n1 outputs columns like:
NAME MAJ:MIN RM SIZE RO TYPE MOUNTPOINT ...
nvme0n1 259:0 0 476.97G 0 disk
With mounted filesystem:
nvme0n1 259:0 0 476.97G 0 disk
nvme0n1p1 259:1 0 512B 0 part /boot
nvme0n1p2 259:2 0 100G 0 part /
With swap:
nvme0n1 259:0 0 476.97G 0 disk
nvme0n1p1 259:1 0 20G 0 part [SWAP]
EOFLength of output: 408
🏁 Script executed:
#!/bin/bash
# Search for all process.run commands with grep patterns
rg -n 'process\.run.*grep' -A2 -B2 libvirt/tests/src/virtual_disks/virtual_disks_nvme.py | head -60Length of output: 635
🏁 Script executed:
#!/bin/bash
# Look for the full context of nvme_device assignment and sgdisk usage
ast-grep --pattern 'nvme_device = $_' libvirt/tests/src/virtual_disks/virtual_disks_nvme.pyLength of output: 103
🏁 Script executed:
#!/bin/bash
# Search for sgdisk execution pattern
rg -n "sgdisk.*nvme_device" -B5 -A5 libvirt/tests/src/virtual_disks/virtual_disks_nvme.pyLength of output: 583
🏁 Script executed:
#!/bin/bash
# Get full function context to understand the flow
rg -n "def.*test.*nvme" -A200 libvirt/tests/src/virtual_disks/virtual_disks_nvme.py | head -150Length of output: 0
🏁 Script executed:
#!/bin/bash
# Search for the loop and device selection logic
rg -n "for device in" -A15 libvirt/tests/src/virtual_disks/virtual_disks_nvme.pyLength of output: 931
🏁 Script executed:
#!/bin/bash
# Final check - see if break statement exists after nvme_device assignment
rg -n "nvme_device = device_path" -A5 libvirt/tests/src/virtual_disks/virtual_disks_nvme.pyLength of output: 266
I need to verify the exact behavior of the grep command with alternation syntax:
Now let me verify the actual behavior of lsblk to understand what the grep is matching against:
Based on my verification, I've discovered a critical distinction: GNU grep allows you to escape | in BRE (basic regular expressions) and get alternation, so grep -q SWAP\|/ will actually work on Linux systems with GNU grep.
However, your review's actual concern about the device selection logic is valid and critical. The real problem isn't the grep syntax—it's the pattern itself:
The mount point will be empty if it is unmounted, and unmounted partitions will have an empty MOUNTPOINT. When running lsblk /dev/nvme0nX, the output shows all partitions nested under that device with their mount points. The current grep pattern SWAP\|/ only detects devices mounted at root or swap—but misses devices mounted at /data, /mnt, /home, or other paths.
This means a device with partitions mounted anywhere except root or swap would be incorrectly classified as "unused" and subsequently destroyed by sgdisk --zap-all.
Fix NVMe device selection to detect all mount points, not just root/swap
Devices can be mounted at arbitrary paths (e.g., /data, /mnt). The current grep pattern only catches / and [SWAP]:
- mount_check = process.run(f"lsblk {device_path} | grep -q SWAP\\|/",
- ignore_status=True, shell=True)
- if mount_check.exit_status != 0:
+ # Check for ANY non-empty mountpoint (catches all mount paths, not just / and SWAP)
+ mount_check = process.run(
+ f"lsblk -n -o MOUNTPOINT {device_path} | grep -qE '\\S'",
+ ignore_status=True, shell=True)
+ if mount_check.exit_status != 0:
nvme_device = device_path
breakThis reliably detects any partition with a non-empty mountpoint before invoking sgdisk --zap-all.
🧰 Tools
🪛 Ruff (0.14.4)
128-128: Function call with shell=True parameter identified, security issue
(S604)
136-136: Function call with shell=True parameter identified, security issue
(S604)
🤖 Prompt for AI Agents
libvirt/tests/src/virtual_disks/virtual_disks_nvme.py around lines 128-139: the
current mount check uses grep -q SWAP\|/ which only detects root or swap and
will miss partitions mounted at other paths, causing mounted devices to be
treated as unused; replace the grep-based check with a test that detects any
non-empty MOUNTPOINT for the device and its partitions (e.g., query lsblk for
MOUNTPOINTs for the device and consider the device mounted if any line is
non-empty), keep ignore_status/shell handling, and only set nvme_device when no
mounted partitions are found.
| if with_blockcopy: | ||
| run_blockcopy_test() | ||
| else: | ||
| run_attach_detach_test() |
There was a problem hiding this comment.
Better we can have condition when call run_attach_detach_test in order to further expand:
if with_blcok:
xx
elif attach_detach_test:
xx
else:
logging.info()
| except Exception as err: | ||
| test.fail("Failed to create block NVME disk: %s" % str(err)) | ||
|
|
||
| def run_attach_detach_test(): |
There was a problem hiding this comment.
maybe move this out of scope :runtest(), and put them ahead of runtest(). If you implement in runtest(), those will become method inline code
| virsh.detach_device(vm_name, device_obj.xml, flagstr="--live", | ||
| debug=True, ignore_status=False) | ||
|
|
||
| def run_blockcopy_test(): |
There was a problem hiding this comment.
suggest move it out of runtest() scope
| if with_luks: | ||
| virsh.secret_undefine(luks_sec_uuid, debug=True, ignore_status=False) | ||
|
|
||
| def verify_blockcopy_result(params, original_source, nvme_dest_disk): |
There was a problem hiding this comment.
suggest move out of runtest scope
2 participants
Automate case:
RHEL-203952 - Blockcopy disk to nvme destination with luks(with usage)
RHEL-241215 - Blockcopy disk to nvme destination without luks
Summary by CodeRabbit
New Features
Tests