# openshift-virtualization-tests

> Validate OpenShift Virtualization deployments comprehensively with a standardized, strictly-typed test framework.

---

Source: quickstart.md

# Quickstart & Setup

Get your Python environment configured using `uv`, validate your code quality tools, and execute your first OpenShift Virtualization test suite. This guide gets you up and running against a live cluster in minutes.

## Prerequisites

* A running OpenShift cluster with OpenShift Virtualization installed.
* Cluster admin access with the `KUBECONFIG` environment variable exported.
* Python 3.14 or newer.
* The [`uv` package manager](https://github.com/astral-sh/uv) installed on your workstation.
* Git.

## Quick Example

If you already have `uv` and a valid `KUBECONFIG`, you can run your first tests immediately:

```bash
git clone <repository-url> openshift-virtualization-tests
cd openshift-virtualization-tests

# Ensure pre-commit and linter checks pass
uv run pre-commit run --all-files

# Run basic virtualization tests
uv run pytest --tc-file=tests/global_config.py -m tier2 tests/virt/
```

## Step-by-step Setup

### 1. Install Tooling

The project strictly enforces `uv` for dependency management and test execution. Do not use `pip` or virtualenv directly.

```bash
# Install uv (if not already installed)
curl -LsSf https://astral.sh/uv/install.sh | sh
```

### 2. Prepare the Repository

Clone the repository and verify your environment by running the pre-commit checks. This automatically provisions the required dependencies in an isolated environment.

```bash
git clone <repository-url> openshift-virtualization-tests
cd openshift-virtualization-tests

# Install dependencies and run linters
uv run pre-commit run --all-files
```

> **Note:** The `uv run` command automatically handles virtual environments and dependency resolution behind the scenes. See [Code Quality & Pre-commits](code-quality.html) for more details.

### 3. Configure Cluster Access

Tests interact with your OpenShift cluster using standard Kubernetes authentication.

```bash
export KUBECONFIG=/path/to/your/kubeconfig
```

### 4. Execute Tests

Always supply the global test configuration file when invoking pytest. This file dynamically loads parameters and defaults based on your cluster's architecture.

```bash
uv run pytest --tc-file=tests/global_config.py tests/network/
```

> **Tip:** You do not need to run the entire suite. Target specific domains like `tests/network/` or `tests/storage/`. For more domain details, see [Networking Tests](network-tests.html) and [Storage Tests](storage-tests.html).

## Advanced Usage

### CI Verification with Tox

Before submitting a pull request, you must ensure all Continuous Integration (CI) checks pass locally. The project uses Tox to orchestrate these checks.

```bash
# Run the full CI validation suite
uv run tox

# Run only the utility unit tests
uv run tox -e utilities-unittests
```

See [Pull Request Discipline](pr-discipline.html) for all required pre-merge checks.

### Multi-Architecture Environments

If testing against ARM64 or S390X clusters, configure the architecture environment variable before running tests so the framework pulls the correct container images.

```bash
export OPENSHIFT_VIRTUALIZATION_TEST_IMAGES_ARCH=arm64
uv run pytest --tc-file=tests/global_config.py -m arm64 tests/virt/
```

See [Multi-Architecture Support](multi-architecture-testing.html) for topology constraints and advanced routing configurations.

### Filtering Test Execution

You can narrow down executions using tier markers and specific domains.

```bash
# Run complex/time-consuming storage tests
uv run pytest --tc-file=tests/global_config.py -m tier3 tests/storage/
```

See [Running and Filtering Tests](running-tests.html) for a complete list of available markers and execution methods.

## Troubleshooting

* **Missing test configuration:** If pytest fails with configuration errors or missing global variables, ensure you are passing `--tc-file=tests/global_config.py`. See [Configuration & Global Contexts](configuration-constants.html) for details.
* **Authentication failures:** Tests failing with `Unauthorized` or `Forbidden` usually mean an expired or unset `KUBECONFIG`. Validate your session by running `oc whoami` outside the test suite.
* **Linter suppression errors:** The project strictly prohibits `# noqa` or `# type: ignore`. If `uv run pre-commit run --all-files` fails, you must fix the code itself rather than ignoring the rule.

## Related Pages

- [Running and Filtering Tests](running-tests.html)
- [Implementing New Tests](implementing-tests.html)
- [Code Quality & Pre-commits](code-quality.html)

---

Source: running-tests.md

# Running and Filtering Tests

Execute test suites against your cluster, leverage pytest markers to filter by test domain or complexity, and run the test harness inside an isolated container.

## Prerequisites

- A fully configured Python environment using `uv`. See [Quickstart & Setup](quickstart.html) for details.
- A valid OpenShift cluster connection (via the `KUBECONFIG` environment variable or `~/.kube/config`).
- For containerized execution: `podman` or `docker` installed on your machine.

## Quick Example

To run a specific test file using the project's dependency manager:

```bash
uv run pytest tests/network/connectivity/test_pod_network.py
```

To run all basic network tests while excluding complex configurations:

```bash
uv run pytest tests/network -m "tier2 and not tier3"
```

## Step-by-Step Guide

Follow these steps to filter and run tests effectively:

1. **Use the `uv` wrapper:** Always prefix your commands with `uv run`. This ensures the test suite executes within the correctly resolved virtual environment.
   > **Warning:** Never execute `pytest`, `tox`, or `python` directly. Always use the `uv run` prefix.

2. **Select the target path:** Pass the directory or specific test file you want to validate. Pointing pytest directly to the domain folder saves collection time.
   ```bash
   uv run pytest tests/storage/
   ```

3. **Filter using markers:** Use the `-m` flag to specify which markers to include or exclude. You can combine markers using `and`, `or`, and `not`.
   ```bash
   # Run only virtualization tests marked for gating
   uv run pytest tests/virt -m "gating"
   ```

4. **Run full local validations:** Before pushing code, validate everything using `tox`. This runs linting, formatting, and utilities unit tests.
   ```bash
   uv run tox
   ```

## Advanced Usage

### Marker Categories

The test framework categorizes markers into two types. Understanding this distinction is critical for targeting the right tests.

| Marker Category | Examples | Application Method | Purpose |
|-----------------|----------|--------------------|---------|
| **Implicit** | `tier2`, `network`, `storage`, `virt`, `chaos` | Automatically applied by the framework based on the test's directory location or default rules. | Defines standard customer use cases and categorizes tests into feature domains without cluttering the code. |
| **Explicit** | `tier3`, `gating`, `special_infra`, `gpu`, `dpdk` | Manually added to the test code using `@pytest.mark.<marker>`. | Identifies tests with specific hardware requirements, advanced configurations, or those that strictly block release promotion. |

### Handling Special Infrastructure

When writing or running tests that require specific cluster capabilities (like GPU, SR-IOV, or hugepages), they must be explicitly targeted. These tests use the `special_infra` marker alongside their specific requirement marker.

To run tests that require DPDK:

```bash
uv run pytest -m "special_infra and dpdk"
```

See [Implementing New Tests](implementing-tests.html) for more information on applying infrastructure markers correctly in your code.

### Running Tests in a Container

For environments with strict network boundaries (like disconnected clusters) or CI/CD pipelines, you can build and execute the entire test suite from within a container image.

1. **Build the container image:**
   ```bash
   make build-container
   ```

2. **Execute the tests via Podman:** Mount your local kubeconfig into the container so it can authenticate and communicate with the cluster.
   ```bash
   podman run --rm \
     -v ~/.kube/config:/root/.kube/config:z \
     quay.io/openshift-cnv/openshift-virtualization-tests-github:latest \
     pytest tests/virt -m "tier2"
   ```

## Troubleshooting

- **No tests were collected:** Ensure you are passing the correct directory path and that your marker syntax is valid. Tests lacking explicit exclusion markers are implicitly tagged as `tier2`. If you filter for an explicit marker without considering the implicit ones, you might get zero collected items.
- **Dependency errors or ModuleNotFoundError:** You bypassed `uv`. Ensure you are running `uv run pytest` rather than invoking `pytest` directly.
- **Authentication failures in container:** When using `podman run`, double-check the `:z` flag on your volume mount. It handles SELinux context labeling so the container is permitted to read your `kubeconfig` file.

## Related Pages

- [Quickstart & Setup](quickstart.html)
- [Virtualization Tests](virt-tests.html)
- [Pytest Fixture Strategy](fixture-strategy.html)

---

Source: virt-tests.md

# Virtualization Tests

Test OpenShift Virtualization's core capabilities by orchestrating virtual machine lifecycles, manipulating compute resources via hotplug, triggering live migrations, and validating advanced hardware topologies. This guide covers how to write robust virtualization tests under the `virt/` domain using standardized utilities.

- Access to an OpenShift cluster with the OpenShift Virtualization operator installed.
- Appropriate underlying hardware nodes if writing tests for advanced features (e.g., GPU, SR-IOV, high-resource VMs).

Here is the simplest way to define a virtual machine fixture, start it, and verify its lifecycle states.

```python
import pytest
from utilities.virt import (
    VirtualMachineForTests,
    fedora_vm_body,
    running_vm,
    wait_for_vm_interfaces,
)

@pytest.fixture()
def sample_vm(unprivileged_client, namespace):
    name = "sample-vm"
    with VirtualMachineForTests(
        client=unprivileged_client,
        name=name,
        namespace=namespace.name,
        body=fedora_vm_body(name=name),
    ) as vm:
        running_vm(vm=vm)
        yield vm

def test_vm_restart(sample_vm):
    """Test restarting an active virtual machine."""
    sample_vm.restart(wait=True)
    sample_vm.vmi.wait_until_running()
    wait_for_vm_interfaces(vmi=sample_vm.vmi)
    assert sample_vm.ssh_exec.executor().is_connective(), "VM failed to reconnect after restart"
```

1. **Define the VM in a Fixture**: Use the `with VirtualMachineForTests(...)` context manager in your fixture. This ensures the VirtualMachine custom resource is automatically cleaned up when the test finishes.
2. **Provide OS Configurations**: Use template generators like `fedora_vm_body` or utilize dynamic golden image data sources when specific OS configurations (like RHEL or Windows) are required.
3. **Control State Transitions**: The `running_vm(vm=vm)` helper guarantees the VM reaches the `Running` state before yielding to the test. Within the test, manipulate the lifecycle by calling `vm.start()`, `vm.stop()`, or `vm.restart()` with the `wait=True` parameter.
4. **Validate Interfaces**: Always call `wait_for_vm_interfaces()` before attempting SSH connections. OpenShift Virtualization requires time to provision and attach the underlying network interfaces to the guest OS.

## Advanced Usage

### CPU and Memory Hotplug

Testing resource hotplugging involves patching the VM specification while the virtual machine instance is running, and confirming the guest OS successfully registers the new limits.

```python
from tests.utils import (
    hotplug_spec_vm,
    wait_for_guest_os_cpu_count,
    assert_guest_os_memory_amount
)
from utilities.constants.virt import SIX_CPU_SOCKETS, SIX_GI_MEMORY

def test_hotplug_resources(sample_vm):
    # Hotplug CPU and wait for guest recognition
    hotplug_spec_vm(vm=sample_vm, sockets=SIX_CPU_SOCKETS)
    wait_for_guest_os_cpu_count(vm=sample_vm, spec_cpu_amount=SIX_CPU_SOCKETS)

    # Hotplug Memory and verify
    hotplug_spec_vm(vm=sample_vm, memory_guest=SIX_GI_MEMORY)
    assert_guest_os_memory_amount(vm=sample_vm, spec_memory_amount=SIX_GI_MEMORY)
```

> **Warning:** Negative tests simulating reductions in CPU sockets or memory should capture `UnprocessibleEntityError` exceptions or explicitly assert that a restart is required, as the virtualization API prevents active downscaling.

### Live Migration Validation

To verify a VM can successfully migrate to another node without losing state or network connectivity, use the built-in migration utility. It handles creating the `VirtualMachineInstanceMigration` object and verifying connectivity post-migration.

```python
from utilities.virt import migrate_vm_and_verify

def test_vm_migration(admin_client, sample_vm):
    migrate_vm_and_verify(vm=sample_vm, client=admin_client, check_ssh_connectivity=True)
```

### Hardware Verification via virt_special_infra_sanity

The `tests/virt/conftest.py` file includes a `virt_special_infra_sanity` session-scoped fixture that validates cluster hardware capabilities before test collection finishes. When writing tests that require specialized nodes, you must apply the correct markers so the suite can assert node health early.

| Pytest Marker | Action in `virt_special_infra_sanity` |
| --- | --- |
| `@pytest.mark.gpu` | Verifies cluster has nodes with supported NVIDIA GPUs and lacks incompatible DPDK profiles. |
| `@pytest.mark.sriov` | Validates that at least one worker node has SR-IOV network cards configured. |
| `@pytest.mark.high_resource_vm` | Checks the platform (e.g., bare-metal) and ensures nodes advertise required hardware virtualization extensions (`vmx` or `svm`). |
| `@pytest.mark.descheduler` | Confirms the `kube-descheduler` operator is functional and nodes possess the `psi=1` kernel argument. |

> **Tip:** If hardware requirements are missing, `virt_special_infra_sanity` exits the execution safely rather than allowing hundreds of tests to fail with obscure resource errors. Always ensure your new advanced tests are tagged with the right infra markers.

## Troubleshooting

- **`UnsupportedGPUDeviceError` during test collection**: Ensure you haven't assigned `@pytest.mark.gpu` to a test running on an architecture or cluster profile missing the required `GPU_CARDS_MAP` hardware.
- **Tests hanging on `running_vm`**: This usually indicates the node is overcommitted and the VirtLauncher pod is stuck in a `Pending` state. Check your namespace quotas or use the `node_with_most_available_memory` fixture to pin large VMs.

For more information, explore:
- See [Resource Lifecycle & Validation](resource-lifecycle.html) for timeout handling and resource interactions.
- See [Pytest Fixture Strategy](fixture-strategy.html) to understand how to share golden image variables across classes.
- See [Multi-Architecture Support](multi-architecture-testing.html) for applying node architecture conditional blocks in hardware virtualization tests.

## Related Pages

- [Networking Tests](network-tests.html)
- [Storage Tests](storage-tests.html)
- [Running and Filtering Tests](running-tests.html)

---

Source: network-tests.md

# Networking Tests

Validate and verify virtualization networking topologies like SR-IOV, L2 Bridging, and IPv6 by writing robust network tests. This guide shows how to ensure VirtualMachines correctly route traffic, apply network policies, and handle complex network configurations reliably.

## Prerequisites

*   An OpenShift cluster configured with the desired networking operators (e.g., NMState, SR-IOV Network Operator).
*   Test fixtures providing running VirtualMachines (VMs) connected to the target networks.
*   Familiarity with test markers and execution (see [Running and Filtering Tests](running-tests.html)).

## Quick Example

The most common network test validates basic connectivity (e.g., pinging from one VM to another over a specific interface). Here is a simple example using built-in network utilities:

```python
import pytest
from libs.net.vmspec import lookup_iface_status_ip
from utilities.network import assert_ping_successful

@pytest.mark.ipv4
def test_vm_basic_connectivity(vm_a, vm_b, target_network):
    # Lookup the destination IP on VM B's secondary interface
    dst_ip = lookup_iface_status_ip(
        vm=vm_b,
        iface_name=target_network.name,
        ip_family=4
    )

    # Assert successful ping from VM A to VM B's IP
    assert_ping_successful(src_vm=vm_a, dst_ip=dst_ip)
```

## Step-by-Step: Testing Network Configurations

When developing tests for complex networking features like L2 bridges or SR-IOV, follow these steps to ensure reliable results.

### 1. Extract the Destination IP Address

Do not hardcode or assume IP addresses. Always dynamically look up the IP address of the target VM on the specific interface being tested.

```python
from libs.net.vmspec import lookup_iface_status_ip

# Fetch IPv4 address for a specific interface
ip_address = lookup_iface_status_ip(
    vm=my_target_vm,
    iface_name="bridge-network",
    ip_family=4
)
```

### 2. Verify Connectivity or Packet Loss

Instead of using raw sub-processes or direct SSH libraries, use the project's utility functions to verify connectivity behaviors (like pings or packet loss) with built-in retry logic and robust logging.

*   **Successful Ping:** `assert_ping_successful(src_vm, dst_ip)`
*   **Failed Ping (Negative Test):** `assert_no_ping(src_vm, dst_ip)`

> **Tip:** If you need to test connectivity *during* an event (e.g., during live migration), use polling functions like `wait_for_no_packet_loss_after_connection` from the L2 bridge utilities.

### 3. Tag with Special Infrastructure Markers

Networking features often require specific hardware (like GPUs or physical network cards) or cluster configurations. You MUST tag tests requiring non-standard capabilities using the appropriate infrastructure markers.

```python
import pytest

pytestmark = [
    pytest.mark.special_infra,
    pytest.mark.sriov,  # Requires SR-IOV hardware
]
```

See [Implementing New Tests](implementing-tests.html) for more details on properly tagging infrastructure requirements.

## Advanced Usage

### Testing Network Policies

When testing network policies, you often need to verify that specific ports are blocked or allowed. Use the `run_ssh_commands` utility to execute commands like `curl` directly from the source VM.

```python
import shlex
import pytest
from pyhelper_utils.exceptions import CommandExecFailed
from pyhelper_utils.shell import run_ssh_commands

def test_network_policy_deny_http(vm_a, vm_b, deny_all_policy):
    dst_ip = vm_b.vmi.interfaces[0].ipAddress
    command = shlex.split(f"curl --connect-timeout 5 -I http://{dst_ip}:80")

    # Expect the command to fail because the policy denies the traffic
    with pytest.raises(CommandExecFailed):
        run_ssh_commands(
            host=vm_a.ssh_exec,
            commands=[command]
        )
```

### IPv6 and Link-Local Addresses

If your test operates in an IPv6 environment, standard IP lookups might return a list of addresses including link-local addresses. You can filter these specifically if your scenario requires routing via `fe80::` addresses:

```python
from libs.net.ip import filter_link_local_addresses
from libs.net.vmspec import lookup_iface_status

def test_ipv6_link_local_ping(vm_a, vm_b, ipv6_network):
    # Fetch all IPs for the interface
    all_ips = lookup_iface_status(vm=vm_b, iface_name=ipv6_network.name)["ipAddresses"]

    # Filter for link-local addresses only
    link_local_ips = filter_link_local_addresses(ip_addresses=all_ips)

    for dst_ip in link_local_ips:
         assert_ping_successful(src_vm=vm_a, dst_ip=dst_ip)
```

### Testing L2 Protocols (DHCP, Custom Ethernet Types)

When testing lower-level protocols over bridging configurations (like OVS or Linux Bridges), you might need to broadcast or send custom Ethernet types.

| Traffic Type | Implementation Strategy |
|---|---|
| **DHCP Broadcast** | Use `TimeoutSampler` to poll `lookup_iface_status_ip` until an IP in the DHCP range is assigned. |
| **Custom EtherTypes** | Execute tools like `nping -e eth1 --ether-type 0x88B6` via `run_ssh_commands` and parse the output. |
| **Multicast** | Use `assert_ping_successful` targeting multicast IPs (e.g., `224.0.0.1`). |

See [Resource Lifecycle & Validation](resource-lifecycle.html) for detailed usage of polling timeouts during network traffic evaluation.

## Troubleshooting

*   **Interface IP Not Found:** If `lookup_iface_status_ip` fails or returns `None`, the VM might still be booting or the secondary network might be incorrectly configured. Wrap the lookup in a timeout loop to wait for the IP to appear.
*   **SSH Command Timeouts:** Commands executed over SSH might hang indefinitely if a network policy silently drops traffic. Always use `--connect-timeout` (or equivalent flags) in tools like `curl` to ensure they fail fast.
*   **Fixture Collection Errors:** If tests complain about missing networking components or undefined fixtures, ensure your test environment supports the requested network type and that you haven't forgotten required decorators (e.g., `@pytest.mark.ipv4`).

## Related Pages

- [Virtualization Tests](virt-tests.html)
- [Storage Tests](storage-tests.html)
- [Infrastructure & Observability](infrastructure-observability.html)

---

Source: storage-tests.md

# Storage Tests

Run tests to validate persistent storage, Containerized Data Importer (CDI) behavior, volume modes, and dynamic provisioning. Testing storage capabilities ensures that virtual machines can successfully attach disks, clone images, handle live migrations, and manage data protection under various conditions.

## Prerequisites

* An active OpenShift cluster with virtualization enabled.
* At least one configured StorageClass (CSI drivers preferred).
* For multi-node storage access, a default StorageClass that supports `ReadWriteMany` (RWX) access modes.

## Quick Example

To execute all storage-related tests across the cluster, use the standard storage marker:

```bash
uv run pytest -m storage
```

If you only want to validate environments with a default RWX StorageClass, filter using the specific marker:

```bash
uv run pytest -m rwx_default_storage
```

## Step-by-Step

Testing storage typically revolves around creating DataVolumes, verifying their phases, and ensuring VMs can consume them.

### 1. Import DataVolumes

When verifying CDI capabilities, ensure that DataVolumes can pull from external sources (HTTP, registry) or clone from existing PVCs. Always use OpenShift resource classes for these operations.

```python
from ocp_resources.datavolume import DataVolume

def test_successful_import_secure_image(namespace, storage_class):
    dv = DataVolume(
        name="alpine-import",
        namespace=namespace.name,
        source={"http": {"url": "https://example.com/alpine.qcow2"}},
        size="200Mi",
        storage_class=storage_class.name,
    )
    dv.deploy()
    dv.wait_for_condition(
        condition=DataVolume.Condition.Type.READY,
        status=DataVolume.Condition.Status.TRUE,
        timeout=300,
    )
```

> **Tip:** When creating tests that wait on storage provisioning, always use explicit polling instead of static delays. See [Resource Lifecycle & Validation](resource-lifecycle.html) for timeout handling strategies.

### 2. Validate RWX Workloads

Many Virtualization features, such as live migration and eviction strategies, depend on multiple nodes having read/write access to the same volume. Tag any test that explicitly requires RWX storage with the appropriate marker.

```python
import pytest

@pytest.mark.rwx_default_storage
def test_vm_live_migration_with_shared_disk():
    # Test implementation relies on the presence of an RWX default storage class
    pass
```

## Advanced Usage

### Working with Volume Binding Modes

Storage classes configured with the `WaitForFirstConsumer` binding mode require a pod to request the volume before the underlying storage is provisioned. Tests validating this must explicitly create a consumer pod or start the VM to trigger provisioning.

| Binding Mode | Test Requirement | Verification Strategy |
|---|---|---|
| `Immediate` | None | Check PVC `Bound` status immediately after creation. |
| `WaitForFirstConsumer` | Launch a dummy pod or VM first | Wait for CDI worker pods to start, then verify `Bound` status. |

### Hotplugging Disks

You can validate dynamic disk attachment by hotplugging DataVolumes into running VMs. Ensure you test both the attachment and the removal processes, verifying that the guest OS recognizes the hardware changes.

```python
def test_hotplug_volume_to_vm(running_vm, new_data_volume):
    running_vm.add_volume(name="data-disk", volume_source={"dataVolume": {"name": new_data_volume.name}})

    # Verify disk is visible in the guest

    running_vm.remove_volume(name="data-disk")
```

See [Virtualization Tests](virt-tests.html) for more on VM lifecycle and hardware interactions.

## Troubleshooting

* **Pending DataVolumes:** If a DataVolume remains in a `Pending` state, check if the associated StorageClass uses `WaitForFirstConsumer` binding mode. If so, you must attach the DataVolume to a workload to trigger provisioning.
* **Skipped RWX Tests:** Tests decorated with `@pytest.mark.rwx_default_storage` will fail or be automatically skipped if the cluster does not have a default StorageClass capable of RWX. Verify your CSI driver capabilities.
* **CDI Import Failures:** Check the `cdi-operator` and `cdi-deployment` pods. Missing proxy settings or invalid CA certificates are common culprits when importing images from external registries.

## Related Pages

- [Virtualization Tests](virt-tests.html)
- [Networking Tests](network-tests.html)
- [Infrastructure & Observability](infrastructure-observability.html)

---

Source: infrastructure-observability.md

# Infrastructure & Observability

Validate cluster-wide configurations and ensure OpenShift Virtualization's monitoring pipelines process metrics and alerts correctly. Use the components in the `observability/` and `infrastructure/` domains to verify that workload behavior accurately triggers Prometheus alerts, and that underlying node features (like NUMA and HugePages) are correctly allocated.

### Prerequisites
* A running OpenShift cluster with OpenShift Virtualization installed.
* Test nodes meeting any specialized infrastructure requirements (if testing hardware features like NUMA).
* The monitoring stack configured and accessible via the cluster.

### Quick Example

Test that a Prometheus metric increments after an action. This requires the session-scoped `prometheus` fixture, which automatically authenticates against the cluster's monitoring endpoint.

```python
import pytest
from utilities.monitoring import validate_metrics_value

def test_vmi_creation_metric(prometheus, running_vm):
    validate_metrics_value(
        prometheus=prometheus,
        metric_name=f"kubevirt_vmi_phase_count{{phase='running', name='{running_vm.name}'}}",
        expected_value="1",
        timeout=120
    )
```

### Step-by-Step: Testing Observability & Alerts

When testing metrics and alerts, the test suite relies on polling mechanisms because Prometheus scraping cycles take time.

#### 1. Inject the Prometheus Fixture
Use the globally available `prometheus` fixture located in `tests/conftest.py`. This fixture handles token retrieval, TLS, and client initialization.

#### 2. Query or Validate a Metric
Instead of writing manual loops to poll Prometheus, use the functions in `utilities/monitoring.py`. For simple assertions, `validate_metrics_value` polls until the expected value appears.

```python
from utilities.monitoring import validate_metrics_value

def test_vm_network_metric(prometheus, running_vm):
    # Trigger an action that should emit a metric
    running_vm.stop(wait=True)

    # Wait for the metric to reflect the change
    validate_metrics_value(
        prometheus=prometheus,
        metric_name=f"kubevirt_vmi_phase_transition_time_from_deletion_seconds_count{{name='{running_vm.name}'}}",
        expected_value="1",
    )
```

#### 3. Wait for Firing Alerts
To test that a critical condition triggers a notification in the OpenShift alert manager, use `wait_for_alert`.

```python
from utilities.monitoring import wait_for_alert

def test_vm_crash_alert(prometheus, crashing_vm):
    # Wait up to the timeout for the specific alert to appear in the firing list
    alert = wait_for_alert(
        prometheus=prometheus,
        alert={
            "alertname": "KubeVirtVMHighMemoryUsage",
            "namespace": crashing_vm.namespace
        }
    )
    assert alert, f"Expected memory alert for VM {crashing_vm.name} did not fire"
```

> **Tip:** You can fetch all currently firing alerts for debugging purposes using `get_all_firing_alerts(prometheus)`.

### Advanced Usage

#### Testing Node Infrastructure (NUMA & HugePages)

Infrastructure testing verifies that VMs correctly consume physical cluster resources. Tests targeting specialized hardware must declare `pytest.mark.special_infra` alongside the specific hardware requirement.

```python
import pytest
from tests.utils import assert_numa_cpu_allocation, get_vm_cpu_list, get_numa_node_cpu_dict

# Ensure this test only runs on clusters with NUMA and HugePages
pytestmark = [pytest.mark.special_infra, pytest.mark.hugepages, pytest.mark.numa]

@pytest.mark.polarion("EXAMPLE-12367")
def test_numa_cpu_allocation(admin_client, created_vm_cx1_instancetype):
    # Validates CPUs are pinned correctly on NUMA nodes
    assert_numa_cpu_allocation(
        vm_cpus=get_vm_cpu_list(vm=created_vm_cx1_instancetype, admin_client=admin_client),
        numa_nodes=get_numa_node_cpu_dict(vm=created_vm_cx1_instancetype, admin_client=admin_client),
    )
```

> **Warning:** Without `@pytest.mark.special_infra`, hardware-specific tests will fail on generic standard CI clusters. See [Implementing New Tests](implementing-tests.html) for marker definitions.

#### Key Functions and Utilities

Use these utility functions to standardize your tests. Do not reinvent metric polling or hardware assertions.

| Function / Fixture | Scope / Location | Purpose |
| --- | --- | --- |
| `prometheus` | `session` (`tests/conftest.py`) | Provides an authenticated Prometheus client. |
| `validate_metrics_value` | `utilities/monitoring.py` | Polls Prometheus until a metric evaluates to an exact string/integer. |
| `wait_for_alert` | `utilities/monitoring.py` | Polls the AlertManager for a firing alert matching the given dictionary. |
| `get_metrics_value` | `utilities/monitoring.py` | Performs a single Prometheus query and returns the current metric value. |
| `assert_numa_cpu_allocation` | `tests/utils.py` | Verifies CPU requests match the physical NUMA topology boundaries. |
| `verify_hugepages_1gi` | `tests/utils.py` | Verifies the target nodes have 1Gi HugePages allocated and available. |

### Troubleshooting

* **Metrics taking too long to appear:** The default scrape interval in OpenShift Virtualization is typically 30 seconds. If `validate_metrics_value` times out, increase the timeout parameter (e.g., `timeout=120`) to allow at least two full scrape cycles. See [Resource Lifecycle & Validation](resource-lifecycle.html) for best practices on timeouts.
* **Alerts not clearing:** Some Prometheus alerts have lingering evaluation cycles and might remain in `pending` or `firing` states temporarily after a resource is deleted. Use `wait_for_firing_alert_clean_up` from `utilities/monitoring.py` if a test requires a clean slate before starting.
* **Tests skipped on your cluster:** If your infrastructure tests skip immediately, check that your cluster provides the underlying hardware features and that you haven't explicitly filtered out markers using Pytest arguments. See [Running and Filtering Tests](running-tests.html).

## Related Pages

- [Operations & Chaos](operations-chaos.html)
- [Scale & Upgrades Testing](scale-upgrades.html)
- [Networking Tests](network-tests.html)

---

Source: operations-chaos.md

# Operations & Chaos

Evaluate OpenShift Virtualization robustness under duress, execute disruptive component testing, and validate disaster recovery mechanics. Operations and chaos testing ensures your system gracefully handles data loss, ungraceful node restarts, disrupted storage paths, and pod terminations without permanently degrading workloads.

## Prerequisites

*   A running OpenShift cluster with OpenShift Virtualization deployed.
*   OpenShift API for Data Protection (OADP) configured (if executing `data_protection/` test cases).
*   Correct markers specified for execution: `@pytest.mark.chaos` is required for standard disruption tests.
*   The `multiprocessing_start_method_fork` fixture activated (usually via `pytestmark`) for suites relying on continuous background process failures.

## Quick Example: Component Disruption

To verify that VirtualMachines can still be deployed and managed when control-plane pods are repeatedly deleted, utilize the parameterized `pod_deleting_process` indirect fixture.

```python
import pytest
from ocp_resources.deployment import Deployment
from utilities.constants.namespaces import NamespacesNames
from utilities.constants.timeouts import TIMEOUT_5SEC, TIMEOUT_5MIN
from utilities.virt import running_vm

pytestmark = [
    pytest.mark.chaos,
    pytest.mark.usefixtures(
        "multiprocessing_start_method_fork",
        "chaos_namespace",
        "cluster_monitoring_process"
    ),
]

@pytest.mark.parametrize(
    "chaos_vms_list_rhel9, pod_deleting_process",
    [
        pytest.param(
            {"number_of_vms": 3},
            {
                "pod_prefix": "apiserver",
                "resource": Deployment,
                "namespace_name": NamespacesNames.OPENSHIFT_APISERVER,
                "ratio": 0.5,
                "interval": TIMEOUT_5SEC,
                "max_duration": TIMEOUT_5MIN,
            },
        )
    ],
    indirect=True,
)
def test_pod_delete_openshift_apiserver(pod_deleting_process, chaos_vms_list_rhel9):
    """
    Verifies that VMs can be deployed while openshift-apiserver pods are continuously deleted.
    """
    for vm in chaos_vms_list_rhel9:
        vm.deploy()
        running_vm(vm=vm, wait_for_interfaces=False, check_ssh_connectivity=False)
```

> **Note:** The `pod_deleting_process` fixture initiates a background process that consistently deletes a defined ratio (e.g., `0.5` or 50%) of the target pods at a set interval while your test logic executes.

## Implementing Chaos Experiments

When designing disruptive tests, the primary sequence is to introduce an active interference, perform a standard workload procedure (e.g., VM creation), and assert the system tolerated the disruption.

### 1. Select the Disruption Vector
The suite contains various disruption patterns. Use the utilities in `tests/chaos/utils.py` to trigger them:
*   **Process Manipulation:** Call `create_pod_deleting_process` or `create_pod_deleting_thread` to force restarts of critical operators or daemonsets.
*   **Hardware and Node Interference:** Yield the `rebooting_node` generator to suddenly restart cluster nodes where workloads reside.

### 2. Track System Degradation
Add `cluster_monitoring_process` (scoped at the module level in `tests/chaos/conftest.py`) to your test's `usefixtures` list. This parallel worker logs pod, daemonset, and node states out to a `chaos-monitoring.txt` log file so you can correlate system health with any test failures.

### 3. Assert Component Recovery
Always confirm the system heals after the disruption event finishes. Use `pod_deleting_process_recover` to poll deployments and daemonsets until they report full health and replica matching before passing the test.

## Advanced Usage

### Data Protection & Disaster Recovery (OADP)

The `tests/data_protection/oadp/` directory evaluates snapshot, backup, and restore capabilities via Velero. The aim is to ensure workload data, such as database files within a VM, remains perfectly intact when restored.

```python
import pytest
from ocp_resources.datavolume import DataVolume
from utilities.constants import Images
from utilities.oadp import check_file_in_running_vm
from utilities.constants.oadp import FILE_NAME_FOR_BACKUP, TEXT_TO_TEST

@pytest.mark.parametrize(
    "rhel_vm_with_data_volume_template",
    [
        pytest.param(
            {
                "dv_name": "filesystem-dv",
                "vm_name": "filesystem-vm",
                "volume_mode": DataVolume.VolumeMode.FILE,
                "rhel_image": Images.Rhel.RHEL9_3_IMG,
            },
        ),
    ],
    indirect=True,
)
@pytest.mark.usefixtures("velero_restore_first_namespace_with_datamover")
def test_backup_vm_data_volume_template_with_datamover(rhel_vm_with_data_volume_template):
    # Validates the VM state was perfectly restored by Velero
    check_file_in_running_vm(
        vm=rhel_vm_with_data_volume_template,
        file_name=FILE_NAME_FOR_BACKUP,
        file_content=TEXT_TO_TEST
    )
```

**Key OADP Fixtures:**
*   `velero_backup_single_namespace`: (Session scoped) Triggers a backup of a dedicated namespace, polling Velero until it confirms partial or full completion.
*   `velero_restore_first_namespace_with_datamover`: (Module scoped) Restores resources using a CSI Datamover, yielding when workloads report completion.
*   `rhel_vm_with_data_volume_template`: Seeds VMs with test data natively before a backup initiates.

### Host IO & CPU Stressing

To validate performance boundaries, trigger host-level utilities like `stress-ng` using the `chaos_worker_background_process` parameterized fixture.

```python
@pytest.mark.parametrize(
    "chaos_worker_background_process",
    [
        pytest.param(
            {
                "max_duration": TIMEOUT_2MIN,
                "background_command": "stress-ng --io 5 -t 120s",
                "process_name": "stress-ng",
            },
        ),
    ],
    indirect=True,
)
def test_host_io_stress(
    vm_with_nginx_service,
    nginx_monitoring_process,
    chaos_worker_background_process,
):
    chaos_worker_background_process.join()
    nginx_monitoring_process.join()

    assert nginx_monitoring_process.exitcode == 0, "NGINX failed to respond under IO load"
    assert chaos_worker_background_process.exitcode == 0, "Stress workload failed to execute"
```

## Functions and Modules Summary

| Resource / Module | Path | Purpose |
| :--- | :--- | :--- |
| **Disruption Utils** | `tests/chaos/utils.py` | Contains process threads (`create_pod_deleting_thread`), process killers, and node reboot handlers. |
| **Chaos Standard** | `tests/chaos/standard/test_standard.py` | Standard experimental test flows asserting VM operation during pod starvation and component destruction. |
| **OADP/Velero Workflows** | `tests/data_protection/oadp/test_velero.py` | Full-stack backup and CSI data-mover restore validation routines across file and block volume modes. |
| **OADP Verification** | `utilities.oadp` | Utility operations specific to reading marker files off recovered VMs (`check_file_in_running_vm`). |

> **Tip:** Do not write defensive polling loops (e.g. infinite retries waiting for an API response) when a failure condition is actively invoked. Use native timeout patterns, and if a workload action fails during chaos, the test should explicitly fail. See [Resource Lifecycle & Validation](resource-lifecycle.html) for more on safe polling mechanics.

## Related Pages

- [Infrastructure & Observability](infrastructure-observability.html)
- [Scale & Upgrades Testing](scale-upgrades.html)
- [Test Quarantine Process](quarantine-process.html)

---

Source: fixture-strategy.md

# Pytest Fixture Strategy

Fixtures are the backbone of test state management in the `openshift-virtualization-tests` suite. They handle resource creation, configuration injection, and teardown logic. Because the test suite interacts with complex, stateful OpenShift clusters, mastering our fixture strategy is critical to writing reliable, independent, and fast tests.

This guide explains how fixtures are organized across `conftest.py` files, strict naming conventions, scoping rules, and dependency management.

---

## The Big Picture: `conftest.py` Architecture

To avoid a massive, unmaintainable global state, we strictly layer our `conftest.py` files. Fixtures must be placed as close to their usage as possible, moving to broader shared directories only when proven necessary.

| Location | Purpose | Rules |
| :--- | :--- | :--- |
| **Root `conftest.py`**<br>`/conftest.py` | Pytest hooks and global configuration. | **NO fixtures.** Extract complex logic into pytest plugins instead. |
| **Shared Conftest**<br>`tests/conftest.py` | Exposes broadly shared fixtures across all test domains. | **Import only.** Do not write fixture logic directly here. Import from `tests/fixtures/`. |
| **Domain Fixtures**<br>`tests/fixtures/<domain>/` | Domain-specific modules providing reusable setups (e.g., `tests/fixtures/network/cluster.py`). | Group by domain. Only place fixtures here if used by multiple feature directories. |
| **Feature Conftest**<br>`tests/<domain>/<feature>/conftest.py` | Local feature setups (e.g., `tests/network/l2_bridge/conftest.py`). | The default home for new fixtures. Move up to shared locations only when another test domain needs them. |

> **Note:** Fixture files must ONLY contain fixtures. Helper functions, utility logic, and constants belong in the `utilities/` directory.

---

## Key Concepts and Mandatory Rules

### 1. Noun-Based Naming

Fixtures must describe **what they provide**, not the action they perform. They act as injected resources, not functions.

*   ✅ **Correct:** `vm_with_disk`, `migrated_vm`, `isolated_namespace`
*   ❌ **Incorrect:** `create_vm_with_disk`, `migrate_vm`, `setup_namespace`

### 2. The Single Action Principle

A fixture must perform exactly **ONE action** (single responsibility). If a test requires a namespace, a network attachment definition, and a VirtualMachine, use three separate fixtures injected into the test. Do not create a monolithic `setup_complex_vm_environment` fixture.

### 3. Yield vs. Return

Always `yield` resources rather than `return` them, even if the fixture does not require complex teardown. This ensures the test can inspect the resource, and any required cleanup logic executes predictably after the test completes.

```python
import pytest
from ocp_resources.namespace import Namespace

@pytest.fixture(scope="module")
def isolated_namespace():
    with Namespace(name="test-isolated-ns") as ns:
        yield ns
```

### 4. Dependency Injection Ordering

When a fixture relies on other fixtures, request them in this strict order:
1. Pytest native fixtures (e.g., `request`, `tmpdir`)
2. Session-scoped fixtures
3. Module/Class-scoped fixtures
4. Function-scoped fixtures

> **Warning:** Mixing scope ordering can lead to `ScopeMismatch` errors from pytest, especially when a broader-scoped fixture attempts to depend on a narrower-scoped one.

---

## Fixture Scoping Rules

Proper scoping minimizes API load on the OpenShift cluster and speeds up test execution. Use the narrowest scope necessary, but take advantage of broader scopes for immutable or expensive setups.

| Scope | When to use | Examples |
| :--- | :--- | :--- |
| `scope="function"` | **(Default)** Use when the setup must be completely isolated, modifies state, or creates per-test resources. | Individual VMs, DataVolumes, mutating workloads. |
| `scope="class"` | Use for setup shared across a specific test class. | Shared network configurations for a subset of test cases. |
| `scope="module"` | Use for expensive setups within a single test file (`test_*.py`). | A target VM used by multiple non-destructive connection tests. |
| `scope="session"` | Use for setups that persist across the entire test run. | Base storage classes, global namespaces, cluster-wide feature gates. |

> **Warning:** **NEVER** use a broader scope (like `module` or `session`) if the test modifies the resource's state. If Test A deletes a disk from a module-scoped VM, Test B will fail when it expects that disk to exist.

---

## How it Affects the User

### Parameterized Fixtures
When tests require different permutations of a setup (e.g., testing multiple storage protocols), use parameterized fixtures with `request.param`. Use dictionaries for complex parameters.

```python
@pytest.fixture(scope="function", params=[{"protocol": "iscsi"}, {"protocol": "nfs"}])
def storage_endpoint(request):
    protocol = request.param["protocol"]
    # Setup protocol endpoint...
    yield endpoint
```

### Unused but Required Fixtures
If a test requires a fixture for its side effects (like applying a cluster configuration) but does not directly interact with the yielded resource, you **must** use `@pytest.mark.usefixtures`.

*   ✅ **Correct:** Decorating the test or class with `@pytest.mark.usefixtures("cluster_config")`.
*   ❌ **Incorrect:** Requesting `cluster_config` in the test function signature and never referencing the variable.

### Test Independence vs. Dependencies
Tests must be independent. Use shared fixtures to provide resources. Only use `@pytest.mark.dependency` when Test B absolutely must run after Test A (e.g., sequential cluster state transitions). If you use `@pytest.mark.dependency`, a code comment explaining **why** the dependency exists is mandatory.

---

## Related Pages

*   See [Resource Lifecycle & Validation](resource-lifecycle.html) for guidelines on managing OpenShift resources within your fixtures.
*   See [Implementing New Tests](implementing-tests.html) to learn how to inject these fixtures into functional test cases.
*   See [Configuration & Global Contexts](configuration-constants.html) to understand how fixtures leverage global parameters across different cloud and architecture topologies.

## Related Pages

- [Resource Lifecycle & Validation](resource-lifecycle.html)
- [Configuration & Global Contexts](configuration-constants.html)
- [Implementing New Tests](implementing-tests.html)

---

Source: resource-lifecycle.md

# Resource Lifecycle & Validation

Properly managing the lifecycle of OpenShift resources—from creation and readiness polling to teardown—is crucial for maintaining a reliable automation suite. This page details the standard patterns used across the `openshift-virtualization-tests` framework to safely orchestrate test objects and validate cluster states without introducing hidden bugs or flaky behavior.

By strictly adhering to fail-fast methodologies and standardized polling tools, we ensure that infrastructure anomalies are explicitly caught rather than silently ignored.

## The Big Picture: Resource Lifecycle Flow

Every cluster resource (like a VirtualMachine, NetworkAttachmentDefinition, or DataVolume) goes through a strict sequence of phases during a test. Managing this flow correctly ensures that tests remain independent and that teardowns always succeed, even when assertions fail.

| Phase | Action | Enforced Pattern |
|---|---|---|
| **1. Definition** | Declaring the resource configuration. | Use dedicated classes. Never construct raw YAML dictionaries or use raw `subprocess` calls. |
| **2. Creation** | Deploying the resource to the OpenShift cluster. | Instantiated within a pytest fixture (`yield`) or a standard Python `with` context manager. |
| **3. Readiness** | Waiting for the cluster to report the resource as active. | Use the `TimeoutSampler` to poll for readiness. Never block the thread blindly. |
| **4. Validation** | Asserting properties of the running resource. | Trust the object schema. Perform explicit pytest assertions on returned data. |
| **5. Teardown** | Removing the object and releasing cluster compute. | Handled automatically by the context manager exit or the fixture's post-`yield` execution. |

## Key Concepts

### Polling and Timeouts (`TimeoutSampler`)

Network latencies and cluster reconciliation loops mean that objects rarely transition to a "Ready" state instantly. You must proactively poll the cluster.

> **Warning:** NEVER use `time.sleep()` in test loops. Hardcoded sleep intervals make test suites artificially slow and highly prone to flakiness under load.

Instead, always use the `TimeoutSampler` tool. It cleanly encapsulates the polling loop, handles logging on timeouts, and yields control back to the test efficiently.

**Example usage:**
```python
from timeout_sampler import TimeoutSampler

# Poll until the VM reports its status as running
for sample in TimeoutSampler(wait_timeout=120, sleep=5, func=check_vm_status, vm=my_vm):
    if sample == "Running":
        break
```
> **Tip:** Do not duplicate the internal logging of `TimeoutSampler`. Only log additional context (e.g., specific resource configurations) rather than logging "waiting for 5 seconds..." inside your loop.

### Avoiding Defensive Programming

One of the strict rules of this codebase is **No defensive programming**. We believe in a "fail-fast" paradigm. If an OpenShift cluster returns a malformed object or a linter throws an error, the code must break explicitly so the bug is visible.

Do not hide bugs behind fake defaults or empty checks.

**Anti-Patterns (Prohibited):**
* Checking attributes that are guaranteed to exist (`if hasattr(vm, 'name')`).
* Verifying parameters defined strictly by schema.
* Silently swallowing generic exceptions using `try: ... except Exception: pass`.

**Acceptable Defensive Checks (Exceptions):**
1. **Destructors/Cleanup Phase:** Resources may fail during partial initialization; teardowns must safely check for `None` before attempting deletion.
2. **Optional Parameters:** When a configuration argument is explicitly typed as `Type | None = None`.
3. **Lazy Initialization:** Variables strictly intended to populate on first access.
4. **Platform Constants:** Checking if a specific hardware feature is available on `s390x` vs `amd64`.
5. **Unversioned APIs:** Interacting with external endpoints lacking backward compatibility guarantees.

### Exception Contextualization

When a cluster resource encounters an error, stack traces should provide immediate clarity on what resource failed and why. Catching errors and silently returning `False` is prohibited.

If you must catch an exception to transform it, always re-raise it with context to preserve the original stack trace:

```python
try:
    execute_cluster_operation()
except ConnectionError as original_error:
    # Always include expected vs actual states or resource metadata
    raise ConfigurationError(f"Failed to connect to VM {vm.name} on node {node.name}") from original_error
```

## How it Affects the User

For test authors and engineers diagnosing CI failures, these lifecycle guardrails translate to predictable, clear outcomes:

* **No Ghost Failures:** Because teardowns rely on `yield` blocks and context managers, a test crashing midway will not leave orphaned resources consuming memory on the cluster.
* **Readable Stack Traces:** Fail-fast rules combined with explicit exception contextualization mean you immediately see *"VM failed to transition to Running state"* instead of a cryptic `NoneType has no attribute 'status'` deeply buried in a helper function.
* **Deterministic Execution:** Eliminating `time.sleep()` prevents "works on my machine" bugs where slower CI nodes fail solely due to strict hardcoded timings.

## Related Pages

* Explore how to properly implement these concepts in your code on the [Implementing New Tests](implementing-tests.html) guide.
* Learn about injecting your robustly managed resources into test logic using the [Test Design Workflow (STP/STD)](test-design-workflow.html).
* Understand how to condition your lifecycles across different environments in [Configuration & Global Contexts](configuration-constants.html).

## Related Pages

- [Pytest Fixture Strategy](fixture-strategy.html)
- [Configuration & Global Contexts](configuration-constants.html)
- [Project Utilities](utilities-reference.html)

---

Source: configuration-constants.md

# Configuration & Global Contexts

## What are Configuration and Global Contexts?

In a complex test suite that spans multiple cloud providers, local data centers, and different CPU architectures (amd64, arm64, s390x), relying on hardcoded strings or test-specific setups creates brittle code.

The `openshift-virtualization-tests` framework manages global test context and domain-specific constants centrally. This allows the test suite to:
- Dynamically adapt to the cluster's architecture without manual intervention.
- Provide a consistent set of configuration matrices (e.g., storage classes, networks, instances) across all test runs.
- Prevent typos and duplication by enforcing strict, domain-specific constant files.

By understanding how global config and constants flow through the system, users can easily write tests that scale across any deployment environment.

## The Big Picture: Architecture and Flow

The project splits global parameters into two main categories: **Runtime Configurations** and **Static Domain Constants**.

| Component | Location | Purpose | Dynamic Behavior |
| --- | --- | --- | --- |
| **Global Config** | `tests/global_config.py` | Environment-wide variables, default states, and parametrizing matrices. | Dynamically loads sub-configs (e.g., `global_config_amd64.py`) based on cluster architecture. |
| **Domain Constants** | `utilities/constants/` | Domain-specific static strings (e.g., timeouts, namespaces, component names). | Architecture-specific image URLs are calculated dynamically at module load time. |

### Config Initialization Flow

1. **Architecture Detection:** At test collection time, `utilities.architecture.get_cluster_architecture()` queries Kubernetes Node labels (or reads the `OPENSHIFT_VIRTUALIZATION_TEST_IMAGES_ARCH` env var in CI) to determine if the cluster is single-arch or `multiarch`.
2. **Sub-config Loading:** `tests/global_config.py` uses `pytest_testconfig` to load the appropriate child configuration file (`tests/global_config_{cluster_type}.py`).
3. **Variable Export:** Variables defined in the module are mapped into a global `config` dictionary, making them accessible to any test that imports them.
4. **Constants Binding:** Domain modules export architecture-specific values dynamically so tests request generic names (e.g., `Images.Rhel.RHEL9_REGISTRY_GUEST_IMG`) and receive the correct URL for the cluster's architecture.

## Key Concepts

### Global Configuration Matrices
`global_config.py` defines large matrices used to parametrize tests across standard permutations. Instead of redefining these in every test file, tests import the matrices directly.

Examples of global matrices include:
- `storage_class_matrix`: Supported storage provisioners and volume modes.
- `nic_models_matrix`: Supported NIC models (virtio, e1000e).
- `run_strategy_matrix`: VirtualMachine run strategies (Manual, Always, Halted).

### Domain-Specific Constants
Constants are strictly organized into domain files under `utilities/constants/` rather than a single monolithic file.

- **`architecture.py`**: Strings for CPU architectures (`AMD_64`, `ARM_64`) and multi-arch logic.
- **`namespaces.py`**: Core OpenShift and Virtualization namespaces.
- **`storage.py`**: Storage classes, datavolume access modes, and volume modes.
- **`timeouts.py`**: Standardized wait times (e.g., `TIMEOUT_5SEC`, `TIMEOUT_5MIN`) to use with `TimeoutSampler`.

> **Note:** The `__init__.py` in `utilities/constants/` restricts what is globally exported. Constants must be explicitly imported from their domain module (e.g., `from utilities.constants.timeouts import TIMEOUT_5MIN`).

### Architecture-Aware Variables
The framework abstracts architecture-specific logic. The `Images` class dynamically resolves the container image path depending on whether the cluster runs on x86, ARM, or s390x.

## How it Affects the User

- **Test Parametrization:** By using `pytest.mark.parametrize` with matrices imported from `global_config.py`, your tests automatically run against the project's standard permutations. If a new storage class becomes standard, adding it to the global matrix automatically expands coverage across all tests using it.
- **Multi-Cloud Portability:** Since you import `default_storage_class` or namespace variables rather than hardcoding `"hostpath-csi"` or `"openshift-cnv"`, your test will run smoothly on AWS, bare metal, or vSphere.
- **Multi-Arch Resilience:** Referencing `Images.Rhel...` guarantees the test won't crash trying to pull an `amd64` image on an `arm64` node.
- **No Magic Strings:** If you need a standard timeout or component name, search the `utilities/constants/` directory. Direct string literals are frequently flagged by maintainers during code review.

## Related Pages
- See [Multi-Architecture Support](multi-architecture-testing.html) for detailed guidelines on writing tests for heterogeneous clusters.
- See [Implementing New Tests](implementing-tests.html) for examples of how to parametrize new test cases using global matrices.
- See [Project Utilities](utilities-reference.html) to understand the helper functions that depend on these constants.
- See [Resource Lifecycle & Validation](resource-lifecycle.html) for how to use `timeouts.py` with polling tools.

## Related Pages

- [Pytest Fixture Strategy](fixture-strategy.html)
- [Resource Lifecycle & Validation](resource-lifecycle.html)
- [Project Utilities](utilities-reference.html)

---

Source: test-design-workflow.md

# Test Design Workflow (STP/STD)

Design and document your test scenarios before writing automation code to ensure alignment with the Software Test Plan (STP). The Software Test Description (STD) workflow separates test design from implementation, reducing rework by getting approval on behavior and assertions early.

*   An approved Software Test Plan (STP) document or Jira epic.
*   Familiarity with the target feature being tested.
*   See [Implementing New Tests](implementing-tests.html) for general test development context.

Here is a complete Phase 1 STD stub ready for design review:

```python
def test_flat_overlay_ping_between_vms():
    """
    Test that VMs on the same flat overlay network can communicate.

    STP Reference: https://example.com/stp/vm-network

    Markers:
        - gating

    Preconditions:
        - Flat overlay Network Attachment Definition created
        - Client VM running and attached to network
        - Server VM running and attached to network

    Steps:
        1. Get IP address of Server VM
        2. Execute ping from Client VM to Server VM

    Expected:
        - Ping succeeds with 0% packet loss
    """

test_flat_overlay_ping_between_vms.__test__ = False
```

### 1. Write the STD Stub (Phase 1)
Create the test signature and add a Google-format docstring containing exactly three required sections: `Preconditions:`, `Steps:`, and `Expected:`. Assign `__test__ = False` to the function to prevent pytest from collecting the empty test.

> **Tip:** In your `Preconditions:`, name resources by their functional role (e.g., "Client VM", "Target Node") rather than generic labels or implementation fixture names (like "VM-A" or `vm_to_restart`).

### 2. Add Traceability and Markers
Include a direct link to the STP in the docstring (at the module, class, or test level). List intended pytest markers in the `Markers:` section of the docstring. Do not use actual `@pytest.mark` decorators yet.

### 3. Submit the Design PR
Open a pull request containing ONLY the test stubs. Reviewers will validate the design, coverage, and assertion logic before you spend time writing automation.

### 4. Implement the Automation (Phase 2)
After the STD design is approved and merged, write the automation code in a new pull request.

| Task | Phase 1 (Design PR) | Phase 2 (Implementation PR) |
|---|---|---|
| **Test Collection** | `__test__ = False` | Remove `__test__ = False` |
| **Pytest Markers** | Listed in `Markers:` docstring block | Applied via actual `@pytest.mark` decorators |
| **Automation Logic** | None (Docstring only) | Test code, fixtures, and assertions |

```python
import pytest

@pytest.mark.gating
def test_flat_overlay_ping_between_vms(client_vm, server_vm):
    """
    Test that VMs on the same flat overlay network can communicate.

    STP Reference: https://example.com/stp/vm-network
    ...
    """
    server_ip = server_vm.get_ipv4()
    # Ping helper handles the assertion mapped to the 'Expected' docstring section
    client_vm.ping(server_ip)
```

## Advanced Usage

### Shared Preconditions (Class Level)
Group related tests inside a class to share common preconditions. Place shared setup in the class docstring and test-specific setup in the test docstring. To skip an entire class during Phase 1, assign `__test__ = False` at the class level. See [Pytest Fixture Strategy](fixture-strategy.html) for implementing these shared states.

```python
class TestSnapshotRestore:
    """
    Tests for VM snapshot restore functionality.

    Preconditions:
        - Running VM with a data disk
        - Snapshot created from VM
    """
    __test__ = False

    def test_preserves_original_file(self):
        """
        Test that files created before a snapshot are preserved after restore.

        Steps:
            1. Read file /data/original.txt from the restored VM

        Expected:
            - File content equals "data-before-snapshot"
        """
```

### Negative Tests
When verifying failure scenarios, prefix the test description with the `[NEGATIVE]` indicator to clearly communicate the intent.

```python
def test_isolated_vms_cannot_communicate():
    """
    [NEGATIVE] Test that VMs on separate flat overlay networks cannot ping each other.
    ...
    Expected:
        - Ping fails with 100% packet loss
    """
```

### Parametrized Tests
Indicate test matrices using a `Parametrize:` block in your STD. You can also specify inline markers for specific parameter values using `[Markers: ...]`.

```text
    Parametrize:
        - ip_family:
            - ipv4 [Markers: ipv4]
            - ipv6 [Markers: ipv6]
```

## Troubleshooting

*   **Test runs empty or is skipped unexpectedly:** Ensure you removed `__test__ = False` when submitting the Phase 2 implementation.
*   **Review blocked on missing traceability:** Verify the STP link is included in either the module, class, or test docstring. Partial coverage without documented exclusions blocks PR merges.
*   **Tests depending on each other:** Tests must be independent. If strict ordering is required for a sequence of operations, group them in a class and use the `@pytest.mark.incremental` decorator.

## Related Pages

- [Implementing New Tests](implementing-tests.html)
- [Test Quarantine Process](quarantine-process.html)
- [Pull Request Discipline](pr-discipline.html)

---

Source: implementing-tests.md

# Implementing New Tests

Writing clear, robust, and independent test cases ensures our virtualization features remain stable over time. By following standard patterns for assertions, utility reuse, and infrastructure tagging, you prevent flaky test runs and speed up the review process.

## Prerequisites

* Your development environment must be configured with `uv`. See [Quickstart & Setup](quickstart.html).
* The feature test plan (STP) must be approved and documented.
* Basic understanding of pytest fixtures and OpenShift virtualization primitives.

## Quick Example

The simplest working example of a test verifies VM state transitions. It defines an isolated resource via a fixture, runs the steps sequentially, and asserts conditions directly.

```python
import logging
import pytest

from utilities.virt import (
    VirtualMachineForTests,
    fedora_vm_body,
    running_vm,
    wait_for_vm_interfaces,
)

LOGGER = logging.getLogger(__name__)

@pytest.fixture()
def vm_to_restart(unprivileged_client, namespace):
    name = "vm-to-restart"
    with VirtualMachineForTests(
        client=unprivileged_client,
        name=name,
        namespace=namespace.name,
        body=fedora_vm_body(name=name),
    ) as vm:
        running_vm(vm=vm)
        yield vm

@pytest.mark.polarion("EXAMPLE-1497")
def test_vm_restart(vm_to_restart):
    """
    Test VM restart operations.

    Preconditions:
        - A running Fedora VM exists.

    Steps:
        1. Restart the VM.
        2. Stop the VM.
        3. Start the VM.

    Expected:
        - VM reaches running state and network interfaces are ready.
    """
    LOGGER.info("Restarting VM")
    vm_to_restart.restart(wait=True)

    LOGGER.info("Stopping VM")
    vm_to_restart.stop(wait=True)

    LOGGER.info("Starting VM")
    vm_to_restart.start(wait=True)
    vm_to_restart.vmi.wait_until_running()
    wait_for_vm_interfaces(vmi=vm_to_restart.vmi)

    assert vm_to_restart.ssh_exec.executor().is_connective(), "Failed to SSH into the restarted VM"
```

## Step-by-Step

Follow these steps to implement a new feature test successfully:

### 1. Write the Test Signature and Docstring
Begin by defining the test function and attaching its required tracking IDs and markers. The docstring MUST contain `Preconditions:`, `Steps:`, and `Expected:` blocks matching your STP.

```python
@pytest.mark.polarion("EXAMPLE-1234")
def test_new_feature_validation(my_feature_fixture):
    """
    Verify the new feature behaves correctly.
    ...
    """
```

### 2. Prepare Resources via Fixtures
Never create test objects directly in the test body if they need teardown. Use a pytest fixture with a context manager (using `with`) and `yield` the object to the test. Ensure the fixture does exactly ONE action and is named using a noun (e.g., `vm_with_disk`).

### 3. Implement the Test Logic
Keep tests focused on verifying one aspect. Use existing utility libraries from `utilities/` for common operations instead of calling raw OpenShift YAMLs or subprocesses.

* Use `LOGGER.info` to record phase transitions.
* Use `LOGGER.warning` for missing optional configurations.
* Use `LOGGER.error` to catch and wrap specific exceptions with full context.

### 4. Assert Expected Outcomes
Replace custom boolean logic with straightforward `assert` statements. Include a clear error message on failure to describe what went wrong.

```python
# Bad
if not vmi.exists:
    raise Exception("VMI is missing")

# Good
assert vmi.exists, f"VirtualMachineInstance {vmi.name} failed to create in {namespace.name}"
```

## Advanced Usage

### Polling Conditions with TimeoutSampler
Hardcoded `time.sleep()` delays cause flaky and slow tests. Whenever you wait for an asynchronous state change, use `TimeoutSampler` from the `timeout_sampler` library.

```python
from timeout_sampler import TimeoutSampler

def _check_condition(resource):
    return resource.status == "Ready"

for sample in TimeoutSampler(
    wait_timeout=60,
    sleep=5,
    func=_check_condition,
    resource=my_resource
):
    if sample:
        break
```

> **Tip:** Do not duplicate timeout exception logs. The `TimeoutSampler` automatically logs the duration and any caught exceptions. Only log additional context if needed.

### Tagging Special Infrastructure
If your test requires specialized cluster hardware or configurations (e.g., GPUs, SR-IOV networks, or DPDK), you must explicitly flag it so CI orchestrates it correctly. Include the `@pytest.mark.special_infra` marker alongside the hardware requirement.

```python
@pytest.mark.special_infra
@pytest.mark.gpu
def test_gpu_workloads(gpu_vm):
    # Specialized hardware test logic
```

For platform-specific differences, see [Multi-Architecture Support](multi-architecture-testing.html).

### Adding Temporary Exclusions
When your test identifies a genuine product defect, do not disable the test entirely. Attach a conditional marker tying it to the issue tracker.

```python
@pytest.mark.jira("EXAMPLE-99999", run=False)
def test_known_bug(my_fixture):
    pass
```

For more details on when to use xfail versus Jira markers, see the [Test Quarantine Process](quarantine-process.html).

## Troubleshooting

* **Linter failures locally:** Always commit via `uv run pre-commit run --all-files`. If mypy complains about missing types in new utility functions, add standard Python type hints.
* **Dead code warnings:** Ensure all fixtures and variables are referenced. Unused code will cause `ruff` to fail.
* **Fixture scope issues:** If a test modifies a global resource unexpectedly, check your fixture's scope parameter. Resources that mutate state per-test must use the default `scope="function"`. See [Pull Request Discipline](pr-discipline.html) for guidelines on ensuring clean test implementations.

## Related Pages

- [Test Design Workflow (STP/STD)](test-design-workflow.html)
- [Pytest Fixture Strategy](fixture-strategy.html)
- [Pull Request Discipline](pr-discipline.html)

---

Source: quarantine-process.md

# Test Quarantine Process

Temporarily disable failing tests to keep continuous integration (CI) lanes green while teams investigate root causes. Categorize failures correctly to ensure product bugs are tracked separately from automated test code issues.

## Prerequisites

- A failing test identified in the CI pipeline.
- An open Jira ticket (Bug for product defects, Task for automation issues).
- Local testing environment configured (See [Quickstart & Setup](quickstart.html)).

## Quick Examples

### Quarantining an Automation Issue

Disable a test when the failure is caused by a test framework or automation script bug:

```python
import pytest
from utilities.constants.pytest import QUARANTINED

@pytest.mark.xfail(
    reason=f"{QUARANTINED}: VM goes into running state unexpectedly, EXAMPLE-12345",
    run=False,
)
def test_vm_lifecycle():
    ...
```

### Skipping for a Product Bug

Disable a test conditionally when an actual product defect causes the failure:

```python
import pytest

@pytest.mark.jira("EXAMPLE-54321", run=False)
def test_new_feature():
    ...
```

## Quarantine Methods Compared

| Failure Type | Description | Required Action | Pytest Marker | Re-enable Condition |
| --- | --- | --- | --- | --- |
| **Product Bug** | Feature is broken in the product. | Open a Jira Bug. | `@pytest.mark.jira("...", run=False)` | Automatic when the Jira bug transitions to resolved. |
| **Automation Issue** | Test code is flawed or flaky. | Open a Jira Task. | `@pytest.mark.xfail(reason=..., run=False)` | Manual PR to remove the marker after a proven fix. |
| **Environment** | Infrastructure/cluster issue. | Open a DevOps ticket. | **Do NOT Quarantine** | N/A (fix the environment). |

## Step-by-Step: Managing a Failing Test

1. **Analyze the Failure**
   Determine if the failure originates from the OpenShift cluster (Product Bug), the test framework (Automation Issue), or external dependencies (System Issue).
2. **Verify Manually**
   Attempt the test steps manually on a test cluster. If the manual steps succeed but the automation fails, you have an automation issue.
3. **Open a Jira Ticket**
   Include specific failure logs, error messages, and steps to reproduce. Avoid generic titles like "Test failed".
4. **Apply the Correct Marker**
   Add either the `@pytest.mark.xfail` or `@pytest.mark.jira` marker directly above the test function definition.
5. **Submit a Pull Request**
   Prefix the PR title with `Quarantine: ` and include a brief explanation alongside a link to the Jira ticket in the description.

## Advanced Usage

### Running Quarantined Tests Locally

The test suite automatically assigns the `quarantined` marker to tests using the `xfail` format. You can isolate these tests to verify potential fixes:

```bash
# Run only quarantined tests
uv run pytest -m quarantined

# Run tests that are NOT quarantined
uv run pytest -m "not quarantined"

# Combine with other domain markers
uv run pytest -m "quarantined and storage"
```

> **Tip:** See [Running and Filtering Tests](running-tests.html) for more selection commands.

### The De-Quarantine Process

Before removing an `xfail` quarantine marker, you must prove the test is stable.

1. Fix the underlying automation issue.
2. Enhance assertion messages if the original failure lacked context.
3. Run the test consecutively 25 times against an identical cluster setup.

```bash
# Verify stability locally
uv run pytest --repeat-scope=session --count=25 path/to/test_file.py
```

4. Submit a PR removing the `xfail` marker and update the Jira ticket with your local verification results.

## Troubleshooting

- **Test runs despite Jira marker:** Ensure the Jira bug is open. The `pytest_jira` plugin automatically executes the test if the ticket is closed.
- **Marker not detected:** Ensure you include `run=False` in your `xfail` or `jira` markers to prevent test execution.
- **Merge Blocked:** Check if you accidentally used `@pytest.mark.skip` or `pytest.skip()`. The project strictly forbids native skip methods in favor of the workflow above.

## Related Pages

- [Test Design Workflow (STP/STD)](test-design-workflow.html)
- [Running and Filtering Tests](running-tests.html)
- [Operations & Chaos](operations-chaos.html)

---

Source: utilities-reference.md

# 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](resource-lifecycle.html) for guidelines on using these utilities, and [External Ecosystem Wrappers](external-wrappers.html) 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. |

```python
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. |

```python
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. |

```python
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. |

```python
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. |

```python
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`. |

```python
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. |

```python
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"`. |

```python
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](fixture-strategy.html) and [Running and Filtering Tests](running-tests.html) 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., executing `virtctl_upload_dv` in a `tests/storage/conftest.py` setup module).
*   `scope="session"`: Binds cluster-wide administrative changes.

> **Warning:** Never use `__test__ = False` directly 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_verify` relies on live-migration topology. Tests using this utility typically belong in `tests/virt/` or are flagged with appropriate markers (e.g., `@pytest.mark.tier2`).
*   Networking utilities like `ping` require advanced cluster configurations if testing SRIOV or DPDK. Mark these with `@pytest.mark.special_infra` and ensure nodes are properly allocated.

## Related Pages

- [External Ecosystem Wrappers](external-wrappers.html)
- [Resource Lifecycle & Validation](resource-lifecycle.html)
- [Configuration & Global Contexts](configuration-constants.html)

---

Source: external-wrappers.md

# External Ecosystem Wrappers

This page documents the external ecosystem wrappers used across the test suite for command execution and OpenShift resource management. Do not use standard library `subprocess` or construct raw Kubernetes YAML dicts.

---

## `pyhelper-utils` (Shell Execution)

The `pyhelper-utils` library is the mandatory method for executing local commands and SSH commands. Never use `subprocess.run()`.

### `run_command`
Executes a local shell command securely as a subprocess.

| Parameter       | Type        | Default | Description |
|-----------------|-------------|---------|-------------|
| `command`       | `list[str]` | (Required) | The command to execute, typically parsed via `shlex.split()`. |
| `check`         | `bool`      | `True`  | If `True`, raises `pyhelper_utils.exceptions.CommandExecFailed` if the command fails. |
| `verify_stderr` | `bool`      | `True`  | If `True`, validates standard error output. |

```python
from pyhelper_utils.shell import run_command
import shlex

# Example: Run command locally and capture output
success, stdout, stderr = run_command(
    command=shlex.split("oc get pods -n my-namespace"),
    check=False,
    verify_stderr=False
)
```

### `run_ssh_commands`
Executes commands on a remote host via SSH.

| Parameter      | Type                           | Default | Description |
|----------------|--------------------------------|---------|-------------|
| `host`         | `rrmngmnt.Host` / `SSHClient`  | (Required) | The SSH connection object, typically accessed via `vm.ssh_exec`. |
| `commands`     | `list[str] \| str`             | (Required) | The command(s) to execute on the remote host. |
| `check_rc`     | `bool`                         | `True`  | If `True`, validates the command's return code. |
| `wait_timeout` | `int`                          | `120`   | Maximum time (seconds) to wait for the command to complete. |
| `sleep`        | `int`                          | `1`     | Polling interval (seconds) while waiting. |

```python
from pyhelper_utils.shell import run_ssh_commands
from utilities.constants.timeouts import TIMEOUT_2MIN, TIMEOUT_5SEC

# Example: Execute command inside a VM
output = run_ssh_commands(
    host=vm.ssh_exec,
    commands=["cat", "/etc/os-release"],
    wait_timeout=TIMEOUT_2MIN,
    sleep=TIMEOUT_5SEC
)[0]
```

---

## `ocp-resources` (OpenShift Resource Management)

The `ocp-resources` package provides Pythonic abstractions for OpenShift and Kubernetes resources.

### OpenShift Resource Classes
Object-oriented wrappers for OpenShift APIs. Instead of passing dictionaries to Kubernetes client functions, instantiate and use these classes (e.g., `VirtualMachine`, `DataVolume`, `Pod`, `Namespace`).

| Parameter   | Type            | Default | Description |
|-------------|-----------------|---------|-------------|
| `name`      | `str`           | `None`  | The name of the resource. |
| `namespace` | `str`           | `None`  | The namespace where the resource resides. |
| `client`    | `DynamicClient` | `None`  | A preconfigured Kubernetes dynamic client. |
| `teardown`  | `bool`          | `True`  | If `True`, deletes the resource when exiting a context manager block. |
| `body`      | `dict`          | `None`  | Optional raw dictionary of the resource definition (e.g. `spec`). |

```python
from ocp_resources.virtual_machine import VirtualMachine

# Use context managers to ensure automatic cleanup
with VirtualMachine(
    name="test-vm",
    namespace="my-namespace",
    body=vm_dict_body
) as vm:
    # Interaction logic here
    pass
```

> **Note:** See [Resource Lifecycle & Validation](resource-lifecycle.html) for detailed guidelines on the context manager strategy.

### Resource Lifecycle Methods
All resource objects inherit lifecycle state mechanisms from the base `Resource` class.

| Method | Description |
|--------|-------------|
| `.create()` | Explicitly creates the resource if not using a context manager. |
| `.delete()` | Explicitly deletes the resource. |
| `.wait_for_condition(condition, status, timeout)` | Blocks until the given API condition equals the specified boolean status. |
| `.wait_for_status(status, timeout)` | Blocks until the `.status.phase` (or similar) matches the given status string. |

```python
# Example: Waiting for a specific condition
vm.wait_for_condition(
    condition=VirtualMachine.Condition.READY,
    status=True,
    timeout=120
)

# Example: Using exists property
if vm.exists:
    vm.delete()
```

### `ResourceEditor`
Provides a context manager or explicit execution block to safely patch API resources and optionally rollback.

| Parameter  | Type   | Default | Description |
|------------|--------|---------|-------------|
| `patches`  | `dict` | (Required) | A dictionary mapping the `Resource` object to a nested dictionary representing the patch payload. |

```python
from ocp_resources.resource import ResourceEditor

# Example: Update a field using the context manager (auto-rolls back if applicable)
with ResourceEditor(patches={vm: {"spec": {"running": False}}}):
    vm.wait_for_status(status=vm.Status.STOPPED)

# Example: Permanent update via direct execution
ResourceEditor(patches={node: {"metadata": {"labels": {"test-label": "true"}}}}).update()
```

---

## `openshift-python-wrapper` (`ocp_utilities`)

Provides helper modules that wrap complex OpenShift components like Prometheus monitoring and infrastructure polling.

### `Prometheus`
Simplifies querying the OpenShift observability stack.

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `client`  | `DynamicClient` | (Required) | The OpenShift dynamic client instance. |

```python
from ocp_utilities.monitoring import Prometheus

# Example: Execute a PromQL query
prometheus = Prometheus(client=client)
query_results = prometheus.query("kubevirt_vmi_phase_count")

for metric in query_results:
    print(metric["metric"], metric["value"])
```

### `get_pods_by_name_prefix`
Helper utility to fetch running pods dynamically based on their prefix.

| Parameter   | Type  | Default | Description |
|-------------|-------|---------|-------------|
| `prefix`    | `str` | (Required) | The string prefix of the pod name. |
| `namespace` | `str` | (Required) | Target namespace to search. |

```python
from ocp_utilities.infra import get_pods_by_name_prefix

# Example: Find all virt-launcher pods
virt_launcher_pods = get_pods_by_name_prefix(
    prefix="virt-launcher",
    namespace=vm.namespace
)
```

> **Tip:** You can review how test setups apply context managers and these utilities by reading the [Pytest Fixture Strategy](fixture-strategy.html) page.

## Related Pages

- [Project Utilities](utilities-reference.html)
- [Implementing New Tests](implementing-tests.html)
- [Virtualization Tests](virt-tests.html)

---

Source: code-quality.md

# Code Quality & Pre-commits

Ensure your contributions meet project standards by automatically validating formatting, strict type hints, and eliminating dead code before every commit. Running local checks avoids CI failures and reduces back-and-forth during pull request reviews.

- Python and `uv` installed on your system.
- Dependencies initialized in your local environment.

```bash
# Format code, check types, and run linters across all files
uv run pre-commit run --all-files

# Verify no unused code exists in the repository
uv run tox -e unused-code
```

1. **Initialize Git Hooks:**
   Run `uv run pre-commit install` once to configure Git to automatically run checks during `git commit`.
2. **Write and Format Code:**
   The project uses `ruff` for code formatting and linting. Run `uv run pre-commit run ruff --all-files` to automatically format files and fix common linting errors.
3. **Verify Type Hints:**
   Strict type hints are enforced via `mypy`. Check your types by running `uv run pre-commit run mypy --all-files`.
4. **Eliminate Dead Code:**
   Every function, variable, and fixture must be used. Validate that you haven't left unused code behind by running `uv run tox -e unused-code`.

## Advanced Usage

### Skipping Checks on Specific Code

Sometimes the dead-code scanner needs a hint. If a piece of code is valid but not directly referenced in a way the scanner detects, you can append a `# skip-unused-code` comment on the same line to bypass the check.

> **Warning:** The project explicitly prohibits `# noqa`, `# type: ignore`, and `# pylint: disable`. If a linter complains, fix the code. Do not disable linter rules to work around issues.

### Running Utilities Unit Tests

When modifying shared helper functions or utilities, ensure your changes don't break existing logic and maintain the 95% test coverage threshold:

```bash
uv run tox -e utilities-unittests
```

### Validating Architecture Configurations

You can also trigger specialized test environments via `tox`. To verify that tests correctly collect across all supported CPU architectures locally without needing a cluster:

```bash
uv run tox -e pytest-check-multiarch
```

## Troubleshooting

- **Pre-commit hook fails on commit:** The tool will often fix formatting errors automatically but fail the commit process. Simply run `git add` on the modified files and execute `git commit` again.
- **Tox environment issues:** If `tox` fails due to stale dependencies, force recreation of the environment by appending the recreate flag: `uv run tox -e unused-code --recreate`.
- **Commit signing errors:** Ensure your commits include a `Signed-off-by` trailer (DCO), which is enforced by CI. See [Pull Request Discipline](pr-discipline.html) for details.

## Related Pages

- [Pull Request Discipline](pr-discipline.html)
- [Test Design Workflow (STP/STD)](test-design-workflow.html)
- [Quickstart & Setup](quickstart.html)

---

Source: pr-discipline.md

# Pull Request Discipline

You want to contribute code and have your Pull Request (PR) merged as quickly as possible. Following these structural guidelines ensures your PR passes automated checks, remains focused on a single topic, and meets the review standards required for merging into the repository.

## Prerequisites

- Git installed and configured with your real name and email.
- The `uv` tool installed for managing dependencies and running test environments.

## Quick Example: The Ideal PR Workflow

The fastest path to getting code merged looks like this:

```bash
# 1. Branch for ONE specific feature or fix
git checkout -b fix-network-l2-bridge-timeout

# ... make your changes ...

# 2. Run all local checks before committing
uv run pre-commit run --all-files
uv run tox
uv run tox -e utilities-unittests

# 3. Commit with a descriptive title and DCO signature (-s)
git commit -s -m "fix(network): increase timeout for L2 bridge creation"

# 4. Push and create PR
git push -u origin fix-network-l2-bridge-timeout
```

## Step-by-Step Guide

### Step 1: Keep It Focused (Single-Topic Rule)

Every PR must address exactly ONE topic.

- **Do NOT** bundle unrelated changes together.
- **Do NOT** slip in "drive-by" refactoring unless it directly supports your fix.
- If you notice unrelated typos or bugs, create a separate branch and a separate PR.

### Step 2: Ensure the PR Title is Accurate

Your PR title must reflect the *actual change* being made, not merely a side effect.

| Incorrect Title | Correct Title | Why it's better |
|---|---|---|
| `skip artifactory` | `switch data source to DataSource API` | Describes *what* the code actually does |
| `fix failing test` | `fix(storage): update expected pvc size in test_disk.py` | Specific about the domain and fix |

### Step 3: Run Gating CI Checks Locally

Before you commit, you must verify your code passes all linting, formatting, and unit tests. Do not wait for the GitHub CI to tell you there are formatting errors.

```bash
# Run linters and formatters
uv run pre-commit run --all-files

# Run full CI collection and architecture checks
uv run tox

# Verify utility functions have required coverage
uv run tox -e utilities-unittests
```

> **Warning:** Fix all failures before committing. **Never** use `--no-verify` to bypass hooks.

### Step 4: Sign Your Commits (DCO)

All commits require a Developer Certificate of Origin (DCO) signature. This is enforced by CI.

Use the `-s` or `--signoff` flag when committing:
```bash
git commit -s -m "feat(virt): add STD for hotplug scenarios"
```

This automatically appends the required trailer to your commit message:
`Signed-off-by: Your Name <your.email@example.com>`

### Step 5: Fill Out the PR Template Completely

When you open a PR, a template is provided automatically. You **must** keep the mandatory headings and fill them out with meaningful content.

- `##### What this PR does / why we need it:` - Explain the *motivation* (the "why"). Do not leave this blank or use placeholders like `N/A`, `TBD`, or `-`.
- `##### Which issue(s) this PR fixes:` - List associated bugs or epics.
- `##### Special notes for reviewer:` - Highlight areas where you want specific feedback.
- `##### jira-ticket:` - Provide the full URL to the Jira ticket, or explicitly write `NONE`.

## Advanced Usage

### Using Draft PRs

Use Draft PRs to signal that work is intentionally incomplete.

- **When to use:** You have open design questions, failing CI that you are still debugging, or unresolved blockers.
- **When NOT to use:** The PR is ready for review.
- **Rule:** Never merge a PR (or ask for it to be merged) if it has known unresolved issues. Fix them, or document them in Jira and link them before marking the PR as Ready for Review.

### Fixing a Missing DCO Signature

If you forgot to sign a commit and the DCO check fails in GitHub, you can fix it locally:

**If it's just the last commit:**
```bash
git commit --amend -s --no-edit
git push --force-with-lease
```

**If you need to sign multiple older commits:**
```bash
git rebase --signoff origin/main
git push --force-with-lease
```

## Troubleshooting

### CI Fails with "Linter suppressions found"
The project strictly prohibits linter suppressions.
- **Problem:** You added `# noqa`, `# type: ignore`, or `# pylint: disable`.
- **Solution:** Remove the comment and **fix the code** so the linter is satisfied. If you believe the linter is genuinely wrong, you must get explicit approval during code review.

### Tests fail collection during `tox`
- **Problem:** Code fails in the `pytest-check-*` tox environments.
- **Solution:** This usually means a test file has a syntax error, a bad import, or a `__test__ = False` directive being misused. Ensure implemented tests do not have `__test__ = False` (this is only for placeholder STDs). Check your fixture definitions and imports. See [Code Organization](docs/CODE_ORGANIZATION.md) for import and fixture rules.

## Related Pages

- [Code Quality & Pre-commits](code-quality.html)
- [Test Design Workflow (STP/STD)](test-design-workflow.html)
- [Implementing New Tests](implementing-tests.html)

---

Source: multi-architecture-testing.md

# Multi-Architecture Support

Multi-architecture testing support enables the test suite to execute seamlessly against diverse cluster topologies, including single-architecture (homogeneous) and multi-architecture (heterogeneous) OpenShift environments. By avoiding hardcoded assumptions about the underlying CPU architecture (`amd64`, `arm64`, `s390x`), you ensure that OpenShift Virtualization tests are portable, reliable, and compliant across different hardware platforms.

> **Note:** Tests are evaluated on `amd64` by default. When tests target non-x86 hardware or heterogeneous environments, the framework dynamically adjusts configuration data, OS images, and node selectors to match the target architecture.

## The Big Picture: Cluster Types and Run Modes

The test framework detects the cluster's topology using worker node labels. Based on the configuration passed via the `--cpu-arch` flag, the test suite executes in one of three modes:

| Test Suite Scope | Cluster Type | `--cpu-arch` value | What Happens at Runtime |
| :--- | :--- | :--- | :--- |
| **Standard Regression** | Homogeneous (Single Arch) | *(Omitted)* | Auto-detects architecture from node labels. Runs standard regression tests on that specific architecture. |
| **Multiarch Regression** | Heterogeneous (Multi Arch) | Single value (e.g., `arm64`) | Scopes standard regression tests strictly to nodes matching the specified architecture. |
| **Multiarch-Dedicated** | Heterogeneous (Multi Arch) | Comma-separated (e.g., `amd64,arm64`) | Only runs tests decorated with `@pytest.mark.multiarch`. Validates cross-architecture scheduling and behaviors. |

> **Warning:** Running with `--cpu-arch=amd64,arm64` disables automatic test filtering and standard helper logic (like single-OS generation). Tests running in this mode must manually filter nodes and orchestrate cross-arch capabilities.

## Key Concepts and Conditional Logic

Designing multi-architecture tests relies on shared platform constants, dynamic fixtures, and conditional logic.

### Platform Constants

Instead of using raw strings, always import architecture definitions from the central constants file `utilities/constants/architecture.py`.

```python
from utilities.constants.architecture import AMD_64, ARM_64, S390X, MULTIARCH
```

### Architecture Fixtures and Scoping

Several built-in fixtures (located in `tests/conftest.py`) provide contextual architecture data to your tests dynamically:

* **`nodes_cpu_architecture`** (Session scope): Returns the CPU architecture string currently being targeted (e.g., `arm64`). Useful for conditionally skipping steps or mutating configurations.
* **`is_s390x_cluster`** (Session scope): A convenience boolean fixture to quickly alter configuration behavior on IBM Z hardware.
* **`schedulable_nodes`** (Session scope): Automatically filters the list of available OpenShift nodes to only return those matching the architecture being tested (defined by `nodes_cpu_architecture`).

### Conditional Logic in Test Design

When a test or fixture must adapt to the platform, use `nodes_cpu_architecture` rather than making external API calls. The framework evaluates these fixtures before resources are instantiated.

**Example: Adapting CPU Model by Architecture (`tests/conftest.py`)**
```python
@pytest.fixture(scope="session")
def host_cpu_model(schedulable_nodes, nodes_cpu_architecture):
    # ARM_64 environments do not expose specific host-model-cpus in the same way x86 does
    if nodes_cpu_architecture == ARM_64:
        return None
    return get_host_model_cpu(nodes=schedulable_nodes)
```

### Explicit Architecture Placement (Multiarch-Dedicated)

When writing multiarch-dedicated tests (tests that explicitly verify cross-architecture functionality), you must hardcode architecture targets in your resource specifications to ensure VMs are placed correctly.

**Example: Instantiating Specific Architecture VMs (`tests/fixtures/network/multiarch.py`)**
```python
@pytest.fixture(scope="class")
def arm_vm(namespace, unprivileged_client):
    spec = base_vmspec()
    spec.template.spec.architecture = ARM_64  # Force ARM placement
    with fedora_vm(namespace=namespace.name, name="arm-vm", client=unprivileged_client, spec=spec) as vm:
        vm.start(wait=True)
        yield vm
```

## How It Affects the User

As a test writer, your interaction with multi-architecture support revolves primarily around using markers and testing isolated conditional logic.

1. **Architecture Exclusion Markers:**
   If a test verifies a feature not supported on a specific architecture, exclude it at the test collection phase rather than using `if/else` inside the test body.
   To run specific suites locally against architectures, you use standard pytest marker invocation (e.g., `pytest -m s390x` or `pytest -m arm64`). The framework assumes `amd64` availability implicitly on unmarked tests.

2. **The `multiarch` Marker:**
   If you are writing a cross-architecture test (e.g., verifying a golden image import to both `amd64` and `arm64` simultaneously), you **must** apply the `@pytest.mark.multiarch` marker.
   ```python
   pytestmark = [pytest.mark.multiarch]
   ```
   > **Tip:** If a non-multiarch test is collected during a dedicated multi-arch run (`--cpu-arch=amd64,arm64`), pytest will raise an `UnsupportedCPUArchitectureError`.

3. **Silent Handling on Homogeneous Clusters:**
   If your multiarch test runs on a standard single-architecture cluster during CI, the framework automatically deselects it at collection time via the `filter_multiarch_tests` hook. You do not need to add defensive skips for environment availability.

## Related Pages

- See [Running and Filtering Tests](running-tests.html) for detailed command-line arguments and test execution workflows.
- See [Configuration & Global Contexts](configuration-constants.html) for understanding how `--tc-file=tests/global_config.py` impacts architecture discovery.
- See [Pytest Fixture Strategy](fixture-strategy.html) for rules regarding fixture implementation and scope mapping.

## Related Pages

- [Running and Filtering Tests](running-tests.html)
- [Scale & Upgrades Testing](scale-upgrades.html)
- [Infrastructure & Observability](infrastructure-observability.html)

---

Source: scale-upgrades.md

# Scale & Upgrades Testing

Scale and Upgrade testing ensures that OpenShift Virtualization handles extreme limits, massive concurrency, and complex operator lifecycle changes without dropping workloads. For contributors, working in the `tests/scale/` and `tests/install_upgrade_operators/` directories requires a significant mindset shift.

Instead of simple "arrange-act-assert" flows, test developers must carefully choreograph massive cluster state changes, prevent sudden API server exhaustion through staggered batching, sequence operator transitions via strict fixture chains, and prioritize forensic evidence gathering over aggressive teardowns when failures occur.

## The Big Picture

Testing massive scale and product upgrades relies on structured data flow and fixture dependency graphs.

### Scale Testing Architecture Flow

1. **Parameter Ingestion**: Configuration is driven externally (e.g., `tests/scale/scale_params.yaml`), defining OS flavors, storage variants, and batch sizes.
2. **Golden Image Pre-provisioning**: To prevent storage backend DDoS, `DataVolume` clones are deployed at the class level and shared.
3. **Batched Execution**: VMs are spawned and powered on in defined batches with calculated sleep intervals to throttle API server load.
4. **Asynchronous Polling**: Instead of asserting each VM's state sequentially, global samplers evaluate the entire cluster configuration asynchronously.
5. **Fail-Open Debugging**: When massive workloads crash the cluster, default teardowns are suspended to preserve logs for forensic analysis.

### Operator Upgrade Pipeline Flow

The upgrade test framework treats the upgrade process as a pipeline, mapped entirely onto `pytest` fixtures.

1. **Pre-Upgrade Baseline**: Captures active Machine Config Pools (MCP) and firing Prometheus alerts before touching the cluster.
2. **Catalog and Subscription Setup**: Bypasses default OperatorHub sources, applies custom Konflux Image Digest Mirror Sets (IDMS), and updates the HyperConverged Operator (HCO) catalog source.
3. **Trigger Upgrades**: Mutates the subscription channels for CNV or OpenShift to force a new InstallPlan.
4. **Replacement Monitoring**: Actively watches ClusterServiceVersion (CSV) progress, `OperatorCondition` status (`Upgradeable=True`), and operator Pod replacement.
5. **Post-Upgrade Validation**: Ensures VMs migrated successfully and the system stabilized.

## Key Concepts

### Managing Massive State Safely

Scale testing relies on specific utilities and scoping to prevent resource exhaustion and ensure debuggability.

| Name | Type | Scope | Purpose & Location |
|---|---|---|---|
| `failure_finalizer` | Function | N/A | Aborts normal teardown on scale failure, collects `must-gather` logs, and leaves resources intact. (`tests/scale/test_scale_benchmark.py`) |
| `scale_vms` | Fixture | `class` | Translates YAML scale parameters into structured batches of `VirtualMachineForTestsFromTemplate` objects. (`tests/scale/test_scale_benchmark.py`) |
| `all_vms_running` | Function | N/A | Evaluates an entire array of VM objects simultaneously to check if all achieved `RUNNING` status. (`tests/scale/test_scale_benchmark.py`) |

> **Warning:** Never use standard sequential assertions (`for vm in vms: assert vm.ready()`) in scale testing. Always use collective `TimeoutSampler` evaluations to prevent timeout cascades.

### Fixture-Driven Upgrade Routines

Operator upgrades in `tests/install_upgrade_operators/product_upgrade/conftest.py` rely on sequential fixture execution to safely mutate cluster state.

| Fixture Name | Scope | Lifecycle Action |
|---|---|---|
| `cnv_upgrade` | `session` | Global feature flag that determines whether CNV upgrade tests should run. |
| `updated_custom_hco_catalog_source_image` | `function` | Mutates the HCO catalog source with the targeted upgrade test image. |
| `approved_cnv_upgrade_install_plan` | `function` | Approves the newly generated `InstallPlan` to initiate the upgrade. |
| `upgraded_cnv` | `function` | The final execution barrier: waits for CSV `SUCCEEDED` state, `Upgradeable` condition, and pod replacements. |

### Longevity and Upgrade Storms

Tests under `tests/virt/cluster/longevity_tests/` monitor long-running operator resilience. For example, `test_multi_vm_upgrade_and_reboot.py` executes "upgrade storms" — forcing bulk Windows VMs to run internal Windows/WSL2 updates and trigger massive uncoordinated reboots, validating that the underlying CNV operator and storage layers remain stable under sustained stress.

## How it Affects the User

When writing or modifying tests in these domains, adhere to the following behavioral patterns:

* **Embrace the `keep_resources` parameter**: Scale tests must respect external configuration dictating whether to wipe the cluster. Use `fail-open` logic (like `failure_finalizer`) so failures don't destroy hours of state provisioning.
* **Map upgrade steps to fixtures, not test logic**: The core test method (e.g., `test_cnv_upgrade_process` in `tests/install_upgrade_operators/product_upgrade/test_upgrade.py`) should be nearly empty. All heavy lifting, pre-flight checks, and waiting routines must be defined in the requested fixtures.
* **Utilize Pytest Dependencies**: Use `@pytest.mark.dependency()` to string together execution phases.
  * Example: `test_mass_vm_live_migration` relies on the success of `test_scale_vms_running_stability`. If the VMs fail to run, the migration test automatically skips.
* **Know your markers**:
  * `@pytest.mark.scale` isolates volume testing to prevent triggering inside standard CI gating.
  * `@pytest.mark.longevity` flags tests that will consume significant time.
  * `@pytest.mark.gating` applied alongside `@pytest.mark.cnv_upgrade` ensures product upgrades block release pipelines if broken.

## Related Pages

* See [Resource Lifecycle & Validation](resource-lifecycle.html) for detailed patterns on using `TimeoutSampler` to avoid defensive programming and strict timeouts.
* See [Pytest Fixture Strategy](fixture-strategy.html) for rules regarding standard fixture scopes (`session` vs `class`) and noun-based naming.
* See [Operations & Chaos](operations-chaos.html) for adjacent tests regarding node disruption and disaster recovery.
* See [Configuration & Global Contexts](configuration-constants.html) to understand how `global_config.py` interacts with external YAML files for test configurations.

## Related Pages

- [Multi-Architecture Support](multi-architecture-testing.html)
- [Operations & Chaos](operations-chaos.html)
- [Infrastructure & Observability](infrastructure-observability.html)

---