Skip to content

Virtual_disk: add two nvme disk related blockcopy scenarios #6655

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Virtual_disk: add two nvme disk related blockcopy scenarios #6655

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 25 additions & 2 deletions libvirt/tests/cfg/virtual_disks/virtual_disks_nvme.cfg
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,8 @@
type = virtual_disks_nvme
take_regular_screendumps = "no"
start_vm = "no"
target_format = "raw"
type_name = "nvme"
driver_type = 'raw'
disk_image_format = "raw"
device_type = "disk"
target_dev = "vdb"
target_bus = "virtio"
Expand All @@ -14,6 +13,30 @@
variants:
- attach_nvme:
source_attrs = "{'type':'pci', 'managed':'yes', 'namespace':'1', 'index':'1'}"
- blockcopy_nvme:
only coldplug
disk_type = "file"
nvme_disk_type = "block"
with_blockcopy = "yes"
func_supported_since_libvirt_ver = (6, 3, 0)
disk_dict = {"type_name":"${disk_type}", "target":{"dev": "${target_dev}", "bus": "virtio"}, "driver": {"name": "qemu", "type": "${disk_image_format}"}}
source_attrs = "{'type':'pci', 'managed':'yes', 'namespace':'1', 'index':'1'}"
blockcopy_common_options = "--transient-job --verbose --wait"
variants:
- finish:
blockcopy_option = "--finish"
- pivot:
blockcopy_option = "--pivot"
variants:
- without_luks:
nvme_disk_dict = {"type_name":"${nvme_disk_type}", 'device': 'disk', "target":{"dev": "${target_dev}", "bus": "virtio"}, "driver": {"name": "qemu", "type": "${disk_image_format}"}}
- with_luks:
with_luks = "yes"
sec_volume = "luks"
luks_secret_passwd = "redhat"
disk_image_format = "qcow2"
luks_extra_parameter = "--object secret,data=${luks_secret_passwd},id=sec0 -o encrypt.format=luks,encrypt.key-secret=sec0"
nvme_disk_dict = {"source": {"encryption": {"attrs": {"format": "luks"}, "encryption": "luks", "secret": {"type": "passphrase", "usage": "${sec_volume}"}}}, "type_name":"${nvme_disk_type}", 'device': 'disk', "target":{"dev": "${target_dev}", "bus": "virtio"}, "driver": {"name": "qemu", "type": "${disk_image_format}"}}
variants:
- coldplug:
virt_device_hotplug = "no"
Expand Down
221 changes: 193 additions & 28 deletions libvirt/tests/src/virtual_disks/virtual_disks_nvme.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import ast
import logging
import platform
import shutil
Expand All @@ -9,9 +10,13 @@
from virttest import utils_misc
from virttest import virsh
from virttest import virt_vm
from virttest import libvirt_version

from virttest.libvirt_xml import vm_xml, xcepts
from virttest.utils_libvirt import libvirt_disk
from virttest.utils_test import libvirt

from provider.virtual_disk import disk_base

LOG = logging.getLogger('avocado.' + __name__)

Expand All @@ -20,15 +25,20 @@ def run(test, params, env):
"""
Test nvme virtual.

For attach test case:
1.Prepare a vm with nvme type disk
2.Attach the virtual disk to the vm
3.Start vm
4.Check the disk in vm
5.Detach nvme device
"""
vm_name = params.get("main_vm")
vm = env.get_vm(vm_name)

For blockcopy test case:
1.Start a VM with a regular disk
2.Prepare nvme destination disk XML
3.Execute blockcopy with --xml option (--finish or --pivot)
4.Check VM XML after operation
5.Verify disk source in VM
"""
def get_new_disks(old_partitions):
"""
Get new virtual disks in VM after disk plug.
Expand Down Expand Up @@ -81,40 +91,92 @@ def create_customized_disk(params, pci_address):
disk_device = params.get("device_type")
device_target = params.get("target_dev")
device_bus = params.get("target_bus")
device_format = params.get("target_format")
device_format = params.get("disk_image_format")

customized_disk = libvirt_disk.create_primitive_disk_xml(
type_name, disk_device,
device_target, device_bus,
device_format, None, None)

source_dict = eval(params.get("source_attrs"))
source_dict = ast.literal_eval(params.get("source_attrs", "{}"))
disk_src_dict = {"attrs": source_dict}
disk_source = customized_disk.new_disk_source(**disk_src_dict)
disk_source.address = pci_address
customized_disk.source = disk_source
LOG.debug("create customized xml: %s", customized_disk)
return customized_disk

hotplug = "yes" == params.get("virt_device_hotplug")
pkgs_host = params.get("pkgs_host", "")
def create_block_nvme_disk(params, disk_type, disk_obj, disk_dict):
"""
Create a block NVME disk by partitioning an NVME device and creating disk XML

# Get disk partitions info before hot/cold plug virtual disk
if vm.is_dead():
vm.start()
session = vm.wait_for_login()
old_partitions = utils_disk.get_parts_list(session)
session.close()
vm.destroy(gracefully=False)
:param params: Dictionary containing test parameters
:param disk_type: the disk type
:param disk_obj: DiskBase object used for preparing disk XML configuration
:param disk_dict: Dictionary containing disk configuration parameters
:return: Tuple containing:
disk_xml: Disk XML object for the created NVME partition
nvme_device: String path of the base NVME device (e.g., "/dev/nvme0n1")
nvme_partition_path: String path of the created partition (e.g., "/dev/nvme0n1p1")
luks_sec_uuid: UUID of the LUKS secret (None if LUKS not used)
"""
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
Comment on lines +128 to +139
Copy link

@coderabbitai coderabbitai bot Nov 17, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

🧩 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 basic grep does not implement alternation; it looks for the literal string SWAP|/, which will never match typical lsblk output. As a result, mount_check.exit_status is 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-all on 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
                     break

and 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
fi

Length 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.py

Length 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.py

Length 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]
EOF

Length 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 -60

Length 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.py

Length 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.py

Length 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 -150

Length 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.py

Length 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.py

Length 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
                     break

This 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.

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

Comment on lines +123 to +166
Copy link

@coderabbitai coderabbitai bot Nov 17, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

🧩 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 --xml will 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_disk to merge the dev attribute into any existing source structure instead of replacing it (e.g., update source.attrs.dev on the existing dict, keeping source.encryption intact before building disk_xml), or
  • Extend DiskBase.prepare_disk_obj / its call-site with a flag similar to the rbd_with_luks_and_auth path (which uses add_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 30

Length of output: 5229

🏁 Script executed:

#!/bin/bash
# Search for the specific file under review
find . -name "virtual_disks_nvme.py" -type f

Length of output: 55

🏁 Script executed:

#!/bin/bash
# Search for verify_blockcopy_result function
rg -n "def verify_blockcopy_result" --type=py -A 20

Length 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 5

Length 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 20

Length 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.py

Length 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.py

Length 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.py

Length 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 -60

Length 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 5

Length 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 -20

Length 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.py

Length of output: 770

🏁 Script executed:

#!/bin/bash
# Look at lines 241-249
sed -n '238,255p' ./libvirt/tests/src/virtual_disks/virtual_disks_nvme.py

Length 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 3

Length 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.py

Length 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 -60

Length 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.py

Length 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.py

Length 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_disk to preserve encryption: either update source.attrs.dev without replacing the dict, or use add_luks_encryption_to_disk_dict after prepare_disk_obj.
  • Update the docstring for verify_blockcopy_result (line 267): parameter nvme_dest_disk is documented as "NVME destination disk XML object" but is actually a path string (passed as nvme_partition_path at 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.

except Exception as err:
test.fail("Failed to create block NVME disk: %s" % str(err))

def run_attach_detach_test():
Copy link
Contributor

@chunfuwen chunfuwen Nov 24, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

maybe move this out of scope :runtest(), and put them ahead of runtest(). If you implement in runtest(), those will become method inline code

"""
Run original attach/detach test with nvme device
"""
if vm.is_dead():
vm.start()
session = vm.wait_for_login()
old_partitions = utils_disk.get_parts_list(session)
session.close()
vm.destroy(gracefully=False)

pci_addr_in_dict = get_usable_nvme_pci_address()

Expand All @@ -131,13 +193,7 @@ def create_customized_disk(params, pci_address):
if hotplug:
virsh.attach_device(vm_name, device_obj.xml, flagstr="--live",
ignore_status=False, debug=True)
except virt_vm.VMStartError as details:
test.fail("VM failed to start."
"Error: %s" % str(details))
except xcepts.LibvirtXMLError as xml_error:
test.fail("VM failed to define"
"Error: %s" % str(xml_error))
else:

utils_misc.wait_for(lambda: get_new_disks(old_partitions), 20)
new_disks = get_new_disks(old_partitions)
if len(new_disks) != 1:
Expand All @@ -147,8 +203,117 @@ def create_customized_disk(params, pci_address):
time.sleep(10)
if not libvirt_disk.check_virtual_disk_io(vm, new_disk):
test.fail("Cannot operate the newly added disk in vm.")

virsh.detach_device(vm_name, device_obj.xml, flagstr="--live",
debug=True, ignore_status=False)

def run_blockcopy_test():
Copy link
Contributor

@chunfuwen chunfuwen Nov 24, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggest move it out of runtest() scope

"""
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)
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)

def verify_blockcopy_result(params, original_source, nvme_dest_disk):
Copy link
Contributor

@chunfuwen chunfuwen Nov 24, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggest move out of runtest scope

"""
Verify blockcopy result by checking VM XML

:param params: Dictionary containing test parameters
:param original_source: Original disk source path before blockcopy operation
:param nvme_dest_disk: NVME destination disk XML object used in blockcopy
"""
specific_option = params.get("blockcopy_option", "")
target_dev = params.get("target_dev", "vdb")
disk_type = params.get("disk_type", "file")
nvme_disk_type = params.get("nvme_disk_type", "block")
current_vmxml = vm_xml.VMXML.new_from_dumpxml(vm_name)
LOG.debug("The current guest xml after blockcopy is: %s" % current_vmxml)
disk_sources = current_vmxml.get_disk_source(vm_name)
disk_obj = disk_base.DiskBase(test, vm, params)
if len(disk_sources) < 2:
test.fail("VM doesn't have enough disks after blockcopy operation.")

if "--finish" in specific_option:
current_disk_source = disk_obj.get_source_list(current_vmxml, disk_type, target_dev)[0]
if current_disk_source != original_source:
test.fail("Expected VM to keep old disk source '%s', but found '%s'" %
(original_source, current_disk_source))
LOG.info("PASS: VM disk kept the old disk source as expected with --finish")

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:
test.fail("Expected VM to pivot to nvme destination, but disk source remains unchanged: '%s'" %
current_disk_source)
LOG.info("PASS: VM disk pivoted to the new nvme destination as expected with --pivot")

vm_name = params.get("main_vm")
vm = env.get_vm(vm_name)
with_blockcopy = "yes" == params.get("with_blockcopy", "no")
hotplug = "yes" == params.get("virt_device_hotplug")
pkgs_host = params.get("pkgs_host", "")

vmxml_backup = vm_xml.VMXML.new_from_inactive_dumpxml(vm_name)
vmxml = vmxml_backup.copy()

try:
# install essential package in host
if not shutil.which('lspci'):
test.error("package {} not installed, please install them before testing".format(pkgs_host))

if with_blockcopy:
run_blockcopy_test()
else:
run_attach_detach_test()
Copy link
Contributor

@chunfuwen chunfuwen Nov 24, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 virt_vm.VMStartError as details:
test.fail("VM failed to start. Error: %s" % str(details))
except xcepts.LibvirtXMLError as xml_error:
test.fail("VM failed to define. Error: %s" % str(xml_error))
finally:
if vm.is_alive():
vm.destroy(gracefully=False)
Expand Down