Project Utilities
This page provides reference documentation for the core domain-specific utilities and custom classes used in openshift-virtualization-tests.
See Resource Lifecycle & Validation for guidelines on using these utilities, and External Ecosystem Wrappers for interactions relying on ocp-resources or pyhelper-utils.
Utility Overview
| Utility / Class | Purpose | Location |
|---|---|---|
VirtualMachineForTests |
Primary context manager and object representing a test VM. | utilities/virt.py |
BaseVirtualMachine |
Foundational class for loading existing VM resources from the API. | libs/vm/vm.py |
running_vm |
Bootstraps a VM and guarantees it is completely functional. | utilities/virt.py |
migrate_vm_and_verify |
Orchestrates live migration and verifies success. | utilities/virt.py |
virtctl_upload_dv |
Uploads local images to DataVolumes or PVCs via virtctl. | utilities/storage.py |
create_vm_from_dv |
Wraps VirtualMachineForTests to quickly spawn VMs from a DataVolume. |
utilities/storage.py |
ping |
Verifies ICMP connectivity from a source VM to a destination IP. | utilities/network.py |
get_valid_ip_address |
Validates IPv4 or IPv6 formats and dependencies. | utilities/network.py |
Virtualization Utilities
VirtualMachineForTests
Description: Main wrapper class extending ocp_resources.virtual_machine.VirtualMachine. Used to construct, configure, and manage VMs during testing.
Return Value / Effect: Yields an unstarted or started VM object depending on usage context. Can generate and update cloud-init user data, resource constraints, and SSH configuration.
| Parameter | Type | Default | Description |
|---|---|---|---|
name |
str |
N/A | Name of the Virtual Machine. |
namespace |
str |
N/A | Namespace where the VM will reside. |
client |
DynamicClient |
None |
OpenShift API client. |
memory_guest |
str |
None |
Requested memory allocation (e.g., 2Gi). |
cloud_init_data |
dict |
None |
Data to inject via cloud-init. |
ssh |
bool |
True |
Inject public SSH keys automatically. |
from utilities.virt import VirtualMachineForTests
from ocp_resources.resource import Resource
with VirtualMachineForTests(
name="my-test-vm",
namespace="test-namespace",
client=admin_client,
memory_guest="1Gi",
ssh=True,
) as vm:
assert vm.instance.status.phase == Resource.Status.STOPPED
BaseVirtualMachine
Description: A foundational VM class that includes alternative constructors to bind Python instances to existing cluster VMs instead of creating new ones.
Return Value / Effect: Returns a bound BaseVirtualMachine class capable of modifying existing VM resources.
| Parameter | Type | Default | Description |
|---|---|---|---|
name |
str |
N/A | Name of the existing Virtual Machine. |
namespace |
str |
N/A | Namespace where the VM already exists. |
os_distribution |
dict |
None |
Pre-existing OS mapping configuration. |
from libs.vm.vm import BaseVirtualMachine
# Instantiate object purely from an existing cluster resource
existing_vm = BaseVirtualMachine.from_existing(
name="persistent-vm",
namespace="default",
client=admin_client,
)
running_vm
Description: Triggers the VM start process and polls the resource until it is fully available, checking network assignments and cloud-init execution.
Return Value / Effect: Returns the started VirtualMachine instance. Validates interfaces, IP assignments, and prevents silent boot failures.
| Parameter | Type | Default | Description |
|---|---|---|---|
vm |
VirtualMachineForTests |
N/A | Instantiated VM object. |
wait_for_interfaces |
bool |
True |
Poll until all network interfaces have an IP assigned. |
check_ssh_connectivity |
bool |
True |
Verify SSH service is running and accessible. |
ssh_timeout |
int |
120 |
Seconds to wait for SSH connectivity. |
wait_for_cloud_init |
bool |
False |
Wait until cloud-init logs indicate success. |
from utilities.virt import VirtualMachineForTests, running_vm
with VirtualMachineForTests(name="test-vm", namespace="ns", client=client) as vm:
active_vm = running_vm(
vm=vm,
wait_for_interfaces=True,
wait_for_cloud_init=True,
)
migrate_vm_and_verify
Description: Orchestrates live migration by creating a VirtualMachineInstanceMigration object and tracks status until successful completion.
Return Value / Effect: Returns the completed VirtualMachineInstanceMigration object if successful. Modifies cluster state.
| Parameter | Type | Default | Description |
|---|---|---|---|
vm |
VirtualMachine |
N/A | VM target to migrate. |
client |
DynamicClient |
N/A | Cluster Admin client required for migrations. |
timeout |
int |
720 |
Maximum time to wait in seconds. |
wait_for_interfaces |
bool |
True |
Re-validate network stack post-migration. |
from utilities.virt import migrate_vm_and_verify
migration_obj = migrate_vm_and_verify(
vm=my_running_vm,
client=admin_client,
timeout=600,
check_ssh_connectivity=True
)
Storage Utilities
virtctl_upload_dv
Description: Submits local disk images to the OpenShift cluster by wrapping the virtctl image-upload command.
Return Value / Effect: Configures a DataVolume or PVC and uploads image bits. Logs the stdout/stderr.
| Parameter | Type | Default | Description |
|---|---|---|---|
namespace |
str |
N/A | Target namespace. |
name |
str |
N/A | Target DataVolume or PVC name. |
image_path |
str |
N/A | Local file path to the QCOW2 or raw image. |
size |
str |
N/A | Size string (e.g., 10Gi). |
client |
DynamicClient |
N/A | OpenShift API client. |
pvc |
bool |
False |
Create a PVC directly instead of a DataVolume. |
storage_class |
str |
None |
Override the default StorageClass. |
from utilities.storage import virtctl_upload_dv
virtctl_upload_dv(
namespace="storage-tests-ns",
name="my-custom-dv",
image_path="/tmp/fedora-cloud.qcow2",
size="5Gi",
client=admin_client,
)
create_vm_from_dv
Description: Generates a full Virtual Machine specification bounded to an existing DataVolume.
Return Value / Effect: Yields a VirtualMachineForTests context manager initialized from the block volume.
| Parameter | Type | Default | Description |
|---|---|---|---|
dv |
DataVolume |
N/A | Existing, bound DataVolume object. |
client |
DynamicClient |
N/A | Cluster client. |
vm_name |
str |
"cirros-vm" |
Identifier for the VM. |
start |
bool |
True |
Automatically call running_vm. |
from utilities.storage import create_vm_from_dv
with create_vm_from_dv(
dv=my_uploaded_dv,
client=admin_client,
vm_name="disk-boot-vm",
start=True
) as vm:
# VM is fully booted from the DV at this point
assert vm.instance.status.printableStatus == "Running"
Network Utilities
ping
Description: Executes an ICMP ping from a source VM's guest OS to a destination IP. Adjusts parameters for IPv4 vs IPv6 automatically.
Return Value / Effect: Returns the packet loss percentage (float between 0 and 100) or None if output parsing fails.
| Parameter | Type | Default | Description |
|---|---|---|---|
src_vm |
VirtualMachine |
N/A | Source VM to execute command via ssh_exec. |
dst_ip |
str |
N/A | Destination IP address to ping. |
packet_size |
str |
None |
Payload data size in bytes. |
count |
int |
3 |
Number of ICMP requests to send. |
windows |
bool |
False |
Use Windows-specific ping syntax. |
from utilities.network import ping
loss_percentage = ping(
src_vm=client_vm,
dst_ip="192.168.1.10",
count=5
)
assert loss_percentage == 0.0, f"Ping failed with {loss_percentage}% loss"
get_valid_ip_address
Description: Validates strings to guarantee valid IPv4 or IPv6 notation, utilizing the standard ipaddress library.
Return Value / Effect: Returns True if valid, otherwise False.
| Parameter | Type | Default | Description |
|---|---|---|---|
dst_ip |
str |
N/A | IP string to validate. |
family |
str |
N/A | The expected family: "ipv4" or "ipv6". |
from utilities.network import get_valid_ip_address
from utilities.constants.network import IPV6_STR
if get_valid_ip_address(dst_ip="fd10:244::8c4c", family=IPV6_STR):
# Proceed with IPv6 configuration checks
pass
Test Integration Patterns
The utility suite interacts closely with test lifecycle mechanics. See Pytest Fixture Strategy and Running and Filtering Tests for detailed configurations.
Fixture Scoping & Utilities
Utility functions and context managers (like VirtualMachineForTests) should match the scope of the fixture wrapping them.
scope="function": Creates isolated utilities per test. (e.g., temporary VMs).scope="module": Used for expensive components (e.g., executingvirtctl_upload_dvin atests/storage/conftest.pysetup module).scope="session": Binds cluster-wide administrative changes.
Warning: Never use
__test__ = Falsedirectly on tests implementing these utilities unless it is purely a placeholder for STD (Software Test Description).
Pytest Markers
Several utilities implicitly assume marker presence:
migrate_vm_and_verifyrelies on live-migration topology. Tests using this utility typically belong intests/virt/or are flagged with appropriate markers (e.g.,@pytest.mark.tier2).- Networking utilities like
pingrequire advanced cluster configurations if testing SRIOV or DPDK. Mark these with@pytest.mark.special_infraand ensure nodes are properly allocated.