Skip to content

AGENTS.md

Latest commit

 

History

History
1082 lines (819 loc) · 46 KB

AGENTS.md

File metadata and controls

1082 lines (819 loc) · 46 KB
Raw

MTV API Tests - Claude Instructions

This document provides project-specific instructions for the MTV API Tests codebase.

AI Workflow

  1. User Prompt - User requests fix/new test/feature/enhancement
  2. Create Branch - Create a feature branch (e.g., feat/description or fix/description)
  3. Agent Selection - Route to appropriate specialist agent
  4. Code Changes - Specialist implements the changes
  5. Code Review - Delegate to code-reviewer agent after ANY code change
  6. Review Cycle - Repeat steps 4-5 until no more changes needed
  7. Pre-commit - Run pre-commit run --all-files and fix any failures (formatting, linting - no re-review needed)
  8. Completion - All changes reviewed, tests pass, ready to commit

Rules

  • Run agents in PARALLEL when possible
  • Never skip code-reviewer after code changes
  • (MUST) Update README.md when code changes affect usage/requirements/installation/configuration
  • (MUST) Update CLAUDE.md when methodology or coding patterns change. Show proposed changes to user and get approval before committing (These updates happen during the work, not as separate workflow steps)
  • (MUST) CLAUDE.md must have NO duplications - define information once, reference elsewhere. AI context is limited.
  • (MUST) CLAUDE.md must have clear, unambiguous instructions - avoid vague terms without definitions
  • (MUST) Do not dismiss review comments without a resolution comment. If out of scope, create a GitHub issue and reference it
  • (MUST) README and documentation must match actual code — verify examples, markers, and config references

Commands Reference

  • Package Installation: uv sync
  • Pre-commit: pre-commit run --all-files
  • Container Build: podman build -f Dockerfile -t mtv-api-tests

Auto-Generated Files (MUST)

The docs/ folder is auto-generated by docsfy and must NOT be manually edited.

Code Standards

  • Type Annotations (MUST): All new functions and functions with signature changes must have complete type annotations. Use built-in Python typing (dict, list, tuple). Using Any must be justified — don't use it as a shortcut when the actual type is known. Default None on non-optional parameters is a mypy error — use str | None = None, not str = None.
  • Package Management: Use uv for all dependency management
  • Pre-commit (MUST): Must pass before any commit - never use --no-verify
  • No Auto-Skip: Never use pytest.skip() or pytest.fail() for validation inside fixtures or test methods. Validation = checking required inputs/config exist before test execution (belongs in fixtures). Assertions = verifying test outcomes (belongs in test methods). Use @pytest.mark.skipif at class/test level for conditional skipping.
  • Every OpenShift resource: Must use create_and_store_resource() function
  • Logging Format: Use f-strings for logging by default. Use parameterized format (%s) only for expensive operations (e.g., large_object.to_json()) where lazy evaluation matters.

No Inline Imports (MUST)

All imports must be at the top of the file. Never import inside functions, methods, or conditional blocks.

# Wrong
def create_plan_resource(...):
    from utilities.resources import create_and_store_resource
    return create_and_store_resource(...)

# Correct
from utilities.resources import create_and_store_resource

def create_plan_resource(...):
    return create_and_store_resource(...)

Only exception: TYPE_CHECKING block for type-only imports.

OpenShift/Kubernetes Resource Interactions

All cluster interactions must use openshift-python-wrapper. Direct kubernetes package usage is forbidden at runtime.

# Correct imports
from ocp_resources.namespace import Namespace
from ocp_resources.secret import Secret
from ocp_resources.virtual_machine import VirtualMachine
from ocp_utilities.infra import get_client

# Forbidden imports (runtime)
from kubernetes import client  # Never
from kubernetes.dynamic import DynamicClient  # Never instantiate directly

DynamicClient Rules:

Usage Allowed
Import inside TYPE_CHECKING block Yes
String annotation "DynamicClient" Yes
Instantiate DynamicClient(...) directly No - use get_client()
Use in isinstance() or runtime checks No
Import kubernetes.dynamic.exceptions Yes
Import other kubernetes.* modules No

Use Existing Libraries (MUST)

Do not reimplement functionality that exists in project dependencies.

Need Use Don't
OpenShift resources openshift-python-wrapper (ocp_resources.*) kubernetes.client.*
SSH commands python-rrmngmnt Custom paramiko wrappers
Shell commands pyhelper-utils run_ssh_commands Custom subprocess wrappers
VM inventory data ForkliftInventory Direct API calls
Resource constants/statuses Contribute to openshift-python-wrapper upstream Hardcode strings in test code

When upstream is missing a feature (e.g., a Plan status, a resource field):

  1. Open a PR to the upstream library (e.g., openshift-python-wrapper)
  2. Use the upstream addition in our code after the PR is merged. Do not hardcode workarounds.

Function Size and Responsibilities (SHOULD)

  • Primary: Single responsibility — if you need "and" to describe it in the docstring, split it
  • Secondary: Keep functions under 50 lines when possible. Longer functions need clear section comments
  • Extract helpers with _ prefix for sub-tasks
  • Function names must clearly describe WHAT they do (e.g., is_warm_migration_supported not is_supported)
# Wrong — migration function also creates storage map
def migrate_vms_with_copyoffload(...):
    storage_secret = create_storage_secret(...)
    storage_map = get_storage_migration_map(...)  # Should not be here
    execute_migration(...)

# Correct — each function does one thing, caller orchestrates
storage_secret = create_storage_secret(...)
storage_map = get_storage_migration_map(...)
execute_migration(...)

What to flag:

  • Functions over 50 lines (likely doing too much)
  • Functions that create AND use resources in the same body

Code Quality Rules

Fail Fast - Validate Content Not Just Existence (MUST)

Code must never result in None when None is not valid. Fail early with clear errors.

# Wrong - allows None to propagate
def get_vm_firmware(template):
    return template.spec.domain.get("firmware")

# Correct - fail fast
def get_vm_firmware(template):
    firmware = template.spec.domain.get("firmware")
    if firmware is None:
        raise ValueError(f"Firmware not found in template '{template.name}'")
    return firmware

Note: Validate content when applicable. Use if value is None: for None checks. Use if not value: only when empty containers and False are also invalid.

Pass Objects Over Values (SHOULD)

Functions should receive objects and extract needed values internally. This improves API simplicity and maintainability.

# Wrong - extracting values before passing
def create_plan(
    source_provider_name: str,
    source_provider_namespace: str,
    storage_map_name: str,
    storage_map_namespace: str,
):
    ...

create_plan(
    source_provider_name=provider.ocp_resource.name,
    source_provider_namespace=provider.ocp_resource.namespace,
    storage_map_name=storage_map.name,
    storage_map_namespace=storage_map.namespace,
)

# Correct - passing objects
def create_plan(
    source_provider: BaseProvider,
    storage_map: StorageMap,
):
    # Extract values inside the function
    name = source_provider.ocp_resource.name
    namespace = source_provider.ocp_resource.namespace
    ...

create_plan(
    source_provider=provider,
    storage_map=storage_map,
)

When to apply: When you control both the function signature and caller. Existing APIs that require extracted values may accept values.

Variables Must Have Consistent Types (MUST)

# Wrong
actual_affinity = vm.get("affinity")  # Could be dict, list, or None

# Correct
actual_affinity: dict[str, Any] = vm.get("affinity") or {}

Trust Required Arguments (MUST)

Don't check if required function arguments exist.

# Wrong
def compare_labels(expected_labels: dict, actual_labels: dict) -> bool:
    if expected_labels and actual_labels:
        return expected_labels == actual_labels
    return False

# Correct
def compare_labels(expected_labels: dict, actual_labels: dict) -> bool:
    return expected_labels == actual_labels

Clarification: "Trust" means don't check if required arguments were passed (they always are). "Validate at Source" means fixtures validate the VALUE they produce is valid.

No Duplicate Code (MUST)

If the same logic appears twice, extract it into a function. Copy-paste is the indicator. Don't create abstractions for single-use or merely "similar" code.

What to flag:

  • Identical code blocks in multiple test classes or functions
  • Same validation/extraction pattern repeated in different files
  • Same expression computed multiple times — store in a variable and reuse
# Wrong — same check in two places
if "copyoffload" in source_provider_data_copy:
    provider_annotations = {FEATURE_COPY_OFFLOAD: "true"}
# ... 50 lines later ...
if "copyoffload" in source_provider_data_copy:
    setup_copyoffload(...)

# Correct — compute once, reuse
is_copyoffload = "copyoffload" in source_provider_data_copy
if is_copyoffload:
    provider_annotations = {FEATURE_COPY_OFFLOAD: "true"}
# ...
if is_copyoffload:
    setup_copyoffload(...)

No Unnecessary Variables (MUST)

Avoid intermediate variables that add no clarity.

# Wrong
@pytest.fixture
def my_fixture():
    result = create_resource()
    yield result

# Correct
@pytest.fixture
def my_fixture():
    yield create_resource()

Exception Types (MUST)

Use specific exception types instead of generic RuntimeError. Create custom exceptions for domain-specific errors.

Instead of RuntimeError Use
Invalid input/config ValueError
Missing resource ValueError
Type issues TypeError
Key not found KeyError (let propagate)
Connection failures ConnectionError
Domain-specific errors Custom exception class (preferred)

Custom exceptions are encouraged for domain-specific errors. They provide clearer error handling and better debugging:

# Custom exceptions are encouraged for domain-specific errors
from exceptions.exceptions import MigrationTimeoutError, ProviderConnectionError

# Example custom exception usage
if not provider.is_connected():
    raise ProviderConnectionError(f"Failed to connect to provider '{provider.name}'")

if not migration.wait_for_completion(timeout=3600):
    raise MigrationTimeoutError(f"Migration '{migration.name}' timed out after 1 hour")

Location: All custom exceptions must be defined in exceptions/exceptions.py, not scattered in other modules. This centralizes exception definitions for better discoverability.

Exception: RuntimeError is allowed ONLY in pytest hooks for infrastructure failures (e.g., cluster unreachable, API timeout). Use ValueError for configuration errors (e.g., missing config key, invalid credentials file).

No except Exception (MUST)

Broad exception handling masks programming bugs (TypeError, AttributeError, KeyError) that should crash loudly.

# Wrong — swallows everything including bugs
try:
    vm = clone_vm(source_vm_name=name, ...)
except Exception as clone_error:
    raise VmNotFoundError(f"Failed to clone: {clone_error}") from clone_error

# Correct — catch specific exceptions, let bugs propagate
try:
    vm = clone_vm(source_vm_name=name, ...)
except vim.fault.VmConfigFault as err:
    raise VmNotFoundError(f"Failed to clone VM '{name}': {err}") from err

What to flag:

  • Any except Exception that is not in a pytest hook
  • Catching exceptions only to log them and continue
  • Re-wrapping specific exceptions into generic ones

No Fallbacks, No Silent Recovery (MUST)

Tests are deterministic. When something is wrong, fail immediately — never fall back to an alternative.

# Wrong — silent fallback hides the real problem
try:
    resource_pool = self.get_obj([vim.ResourcePool], "Resources")
except ValueError:
    LOGGER.warning("Could not find resource pool, using template's pool")
    relocate_spec.pool = source_vm.resourcePool

# Correct — fail if the expected resource doesn't exist
resource_pool = self.get_obj([vim.ResourcePool], "Resources")
relocate_spec.pool = resource_pool

What to flag:

  • except followed by a fallback code path
  • LOGGER.warning(...) used to paper over a failure instead of raising
  • try/except with a different code path in the except block

No Unnecessary Logging (MUST)

Pytest captures all output. Do not add redundant logging that duplicates what pytest already provides. Progress logging for long-running operations (migration execution, provider connections) is acceptable — flag only redundant per-step logging within those operations.

What to flag:

  • Logging input parameters already visible in pytest output
  • Multiple sequential LOGGER.info() calls that could be one line
  • Reporter/progress logging in test methods
  • Debug logging left in production code

No Dead Code or Unnecessary Comments (MUST)

Code must be clean. Remove anything that adds no value.

What to flag:

  • Comments that restate what the code does (# Create storage map above create_storage_map())
  • Comments like # Success!, # Retry, # original behavior - unchanged
  • Unreachable code after return, raise, or exhaustive loops
  • Debug code left in (listing all datastores, extensive logging of internal state)
  • Commented-out code blocks
  • type: ignore without a comment explaining why

Do Not Mutate Caller Data In-Place (SHOULD)

Functions that receive dicts or lists from callers should not modify them unless that is their documented purpose.

# Wrong — vm.pop() modifies the caller's data
for vm in virtual_machines_list:
    if migrate_shared := vm.pop("migrate_shared_disks", None):
        vm["migrateSharedDisks"] = migrate_shared

# Correct — work on a copy, collect results
processed = []
for vm in virtual_machines_list:
    vm_copy = dict(vm)
    if migrate_shared := vm_copy.pop("migrate_shared_disks", None):
        vm_copy["migrateSharedDisks"] = migrate_shared
    processed.append(vm_copy)

Magic Numbers and Hardcoded Values (SHOULD)

Numeric literals and string constants that aren't self-explanatory should be named constants with comments.

# Wrong — what does 4000 mean? What does -101 mean?
port_offset = 4000
disk_key = -101

# Correct
_SERIAL_PORT_OFFSET = 4000  # vSphere serial port numbering starts at 4000
_RDM_DISK_KEY = -101  # Placeholder key for RDM disk devices

hasattr / Duck-Typing Is Fragile (SHOULD)

Use isinstance() checks instead of hasattr() for method/attribute detection. A typo in hasattr silently skips logic.

# Wrong — typo in method name silently skips
if hasattr(source_provider, "relink_shared_disks"):
    source_provider.relink_shared_disks(...)

# Correct — type-safe, catches typos at development time
if isinstance(source_provider, VMWareProvider):
    source_provider.relink_shared_disks(...)

No Removed Functionality Without Justification (MUST)

Removing existing code (safety checks, validations, feature logic) requires explicit justification in the PR description or commit message.

What to flag:

  • Removed validation/health checks without explanation
  • Removed error handling without confirming the error case no longer exists
  • Changed defaults (e.g., clone=True → clone=False) without documenting why

Feature-Specific Logic Must Not Pollute General Code (MUST)

Adding a new feature (e.g., copy-offload) must not introduce feature-specific branching throughout general-purpose functions.

# Wrong — copy-offload branching injected into general naming logic
def generate_resource_name(plan, ...):
    if plan.get("copyoffload"):
        return f"xcopy-{base_name}"
    return base_name

# Correct — keep general functions general, add feature-specific functions
def generate_resource_name(plan, ...):
    return base_name  # Same for all features

What to flag:

  • if copyoffload: / if feature_x: branches inside general-purpose functions
  • Feature-specific parameters added to functions that serve all features

Use Empty Container Defaults (SHOULD)

Use empty containers as defaults for optional/nested data where absence is valid. Fail fast (see "Fail Fast" rule) for required data where absence indicates a configuration error. Note: This applies to external/provider data (e.g., template.spec), not to our own config — see "Deterministic Tests" for config access rules.

# Wrong
firmware_spec = template.spec.domain.get("firmware")
if firmware_spec is not None:
    boot_order = firmware_spec.get("bootOrder")

# Correct
firmware_spec: dict[str, Any] = template.spec.domain.get("firmware", {})
boot_order: list = firmware_spec.get("bootOrder", [])

Docstring Format (MUST)

All new functions must have docstrings with the standard sections as applicable:

  • Args: include when the function has parameters.
  • Returns: include only when returning a meaningful non-None value (omit for __init__ and other void functions).
  • Raises: include only when exceptions are actually raised.
def process_vm(vm: VirtualMachine, options: dict[str, Any]) -> MigrationResult:
    """Process a VM for migration.

    Args:
        vm (VirtualMachine): The VM resource to process
        options (dict[str, Any]): Processing options

    Returns:
        MigrationResult: The result of the migration processing

    Raises:
        ValueError: If VM is in an invalid state
    """

Validate at Source (MUST)

Definition: Validation = verifying required inputs, configuration values, or fixture dependencies are present and valid before test execution. This is distinct from assertions, which verify test outcomes during execution.

Validation must happen in fixtures (where values originate), not in utility functions or test methods.

# Wrong - validating in utility (too late)
def apply_node_label(labeled_worker_node, ...):
    if not labeled_worker_node:
        raise ValueError("No worker node provided")

# Wrong - validating in test method (see also: "Deterministic Tests - No Defaults for Our Config")
def test_create_storagemap(self, source_provider, ...):
    if not source_provider_data.get("storage_vendor_product"):
        pytest.fail("Missing storage_vendor_product")  # Should be in fixture

# Correct - validate in fixture
@pytest.fixture
def labeled_worker_node(worker_nodes, target_node_selector):
    node = find_node_with_selector(worker_nodes, target_node_selector)
    if not node:
        raise ValueError(f"No node found matching selector {target_node_selector}")
    return node

Distinction: Fixtures validate their own construction (is the fixture value valid?). Utility functions may validate external/provider data that varies at runtime.

Test methods should never contain validation logic - if a config value is required, create a fixture that validates it.

Test Methods Must Be Tests (SHOULD)

Before adding a test_ prefixed method, consider whether it belongs as a test or as a fixture/helper. If the method's sole purpose is a precondition for other tests and has no value as a standalone test step, it likely belongs in a fixture.

Fixture Rules (MUST)

  • Autouse sparingly: Only the autouse_fixtures fixture in conftest.py uses autouse=True. All other fixtures must be requested explicitly via parameters or @pytest.mark.usefixtures()
  • Noun names: Fixtures are nouns (they represent state), never actions
  • No magic skip: See "No Auto-Skip" rule in Code Standards
# Wrong
@pytest.fixture
def setup_provider():    # Action verb
def validate_copyoffload_config():  # Action verb

# Correct
@pytest.fixture
def source_provider():   # Noun — what it IS
def copyoffload_config():  # Noun — what it IS

What to flag:

  • Fixture names that start with verbs (setup_, validate_, create_, configure_)
  • Function/variable names that become misleading after code changes
  • Generic names like data, result, value, info without context

Fixture Request Patterns (MUST)

  • Method parameters: Use when you need the fixture value in the test
  • @pytest.mark.usefixtures(): Use for side-effect fixtures (e.g., cleanup_migrated_vms) that perform setup/teardown but whose return value isn't needed by the test
  • Never list both: Don't request via both parameter AND usefixtures - choose one

No Unnecessary Randomness (MUST)

Tests must be deterministic. Avoid random selection when order does not matter.

# Wrong
selected_node = random.choice(available_nodes)

# Correct
selected_node = available_nodes[0]

Use Context Managers for Cleanup (SHOULD)

Use context managers to ensure proper resource cleanup.

# Wrong
editor = ResourceEditor(node)
editor.add_label(label)
# Caller must remember to cleanup

# Correct
with ResourceEditor(node) as editor:
    editor.add_label(label)
    yield

Architecture Patterns

Provider Abstraction

  • Base class: BaseProvider in libs/base_provider.py
  • Implementations: VMware, RHV, OpenStack, OVA, OpenShift providers
  • Context manager support for provider connections

Critical Constraints

Test Execution Prohibition

AI must NEVER run tests directly (pytest, uv run pytest). Tests require live clusters, provider connections, and credentials.

AI can: Read/analyze/write/fix tests, suggest improvements, review structure AI cannot: Execute tests, validate by running

No Module-Level Provider Loading (MUST)

load_source_providers() must only be called within the pytest ecosystem (fixtures, hooks). Never call it at module level in test files — module-level code runs before pytest parses CLI args like --providers-json, causing the arg to be ignored.

# Wrong - module level, ignores --providers-json
_SOURCE_PROVIDER_TYPE = load_source_providers().get(...)

# Correct - use pytest_collection_modifyitems hook
# (see conftest.py for the documented pattern)

Deterministic Tests - No Defaults for Our Config (MUST)

Data source Access pattern Why
Our config (py_config, plan, tests_params) config["key"] (direct access) Fail fast with KeyError if missing
External/provider data .get("key") + validate Provider data varies at runtime
Optional feature flags .get("key", False) Explicitly optional 
# Wrong — default for our config
storage_class = py_config.get("storage_class", "default-storage")

# Correct — direct access for our config
storage_class = py_config["storage_class"]

# Correct — .get() for optional feature flags (not present in every test config)
warm_migration = plan.get("warm_migration", False)
copyoffload = plan.get("copyoffload", False)
xfs_compatibility = plan.get("xfs_compatibility", False)
skip_clone = plan.get("skip_clone", False)
per_nic_network_map = plan.get("per_nic_network_map", False)

# Correct — .get() with validation for external data
vm_id = provider_data.get("vm_id")
if not vm_id:
    raise ValueError(f"VM ID not found for VM '{vm_name}'")

What to flag:

  • .get("key", <default>) on py_config, plan, tests_params, or any dict we control
  • .get("key") on external data without validation afterward
  • Using False as default when the key is always present in config (exception: warm_migration, copyoffload, enable_nested_virtualization, xfs_compatibility, skip_clone, and per_nic_network_map are optional flags — .get() is correct)

Provider Config Key Access (MUST)

When accessing .providers.json fields in code, check providers_schema.json for whether the key is required or optional on the provider type:

Schema status Access pattern Example
Required field data["key"] source_provider_data["fqdn"]
Optional field data.get("key") source_provider_data.get("vddk_init_image")
Conditional field data.get("key") + validate copyoffload_config.get("ontap_svm")

What to flag:

  • .get() on a field marked required in the schema
  • Direct ["key"] access on a field marked optional in the schema
  • Adding new .providers.json fields without updating providers_schema.json

Schema and Example Maintenance (MUST)

When adding, removing, or renaming any property in .providers.json, update both:

  1. providers_schema.json — required/optional status, types, enums, descriptions
  2. .providers.json.example — add the new field with a placeholder value and comment

Resource Creation - create_and_store_resource()

Every OpenShift resource must use utilities/resources.py:create_and_store_resource().

def create_and_store_resource(
    client: DynamicClient,
    fixture_store: dict[str, Any],
    resource: type[Resource],
    test_name: str | None = None,
    **kwargs: Any,
) -> Any:

Features: auto-generates unique names, deploys and waits, stores in fixture_store["teardown"], handles conflicts, truncates to 63 chars.

# Correct
namespace = create_and_store_resource(
    fixture_store=fixture_store,
    resource=Namespace,
    client=ocp_admin_client,
    name="my-namespace",
)

# Wrong - bypasses tracking
namespace = Namespace(client=ocp_admin_client, name="my-namespace")
namespace.deploy()

Test Structure Pattern

All tests follow a class-based structure with 5 base test methods:

from pytest_testconfig import config as py_config
from ocp_resources.network_map import NetworkMap
from ocp_resources.storage_map import StorageMap
from ocp_resources.migration_toolkit_virtualization import Plan

from utilities.mtv_migration import (
    create_plan_resource,
    execute_migration,
    get_network_migration_map,
    get_storage_migration_map,
)
from utilities.post_migration import check_vms


@pytest.mark.parametrize(
    "class_plan_config",
    [pytest.param(py_config["tests_params"]["test_name_here"])],
    indirect=True,
    ids=["descriptive-test-id"],
)
@pytest.mark.usefixtures("cleanup_migrated_vms")
@pytest.mark.incremental
@pytest.mark.tier0  # optional: tier0, tier1, warm, remote, copyoffload
class TestNameHere:
    """Test description."""

    storage_map: StorageMap
    network_map: NetworkMap
    plan_resource: Plan

    def test_create_storagemap(self, prepared_plan, fixture_store, source_provider, destination_provider, ocp_admin_client, target_namespace, source_provider_inventory):
        """Create StorageMap resource."""
        self.__class__.storage_map = get_storage_migration_map(...)
        assert self.storage_map

    def test_create_networkmap(self, prepared_plan, fixture_store, source_provider, destination_provider, ocp_admin_client, target_namespace, source_provider_inventory, multus_network_name):
        """Create NetworkMap resource."""
        self.__class__.network_map = get_network_migration_map(
            multus_network_name=multus_network_name, ...
        )
        assert self.network_map

    def test_create_plan(self, prepared_plan, fixture_store, source_provider, destination_provider, ocp_admin_client, target_namespace):
        """Create MTV Plan CR resource."""
        self.__class__.plan_resource = create_plan_resource(
            storage_map=self.storage_map,
            network_map=self.network_map,
            ...
        )
        assert self.plan_resource

    def test_migrate_vms(self, fixture_store, ocp_admin_client, target_namespace):
        """Execute migration."""
        execute_migration(
            plan=self.plan_resource,
            ...
        )

    def test_check_vms(self, prepared_plan, source_provider, destination_provider, target_namespace, source_provider_data, source_vms_namespace, source_provider_inventory):
        """Validate migrated VMs."""
        check_vms(
            plan=prepared_plan,
            network_map_resource=self.network_map,
            storage_map_resource=self.storage_map,
            ...
        )

Key Patterns

  • Class-level parametrization: Use class_plan_config with indirect=True

  • Shared state: Store resources on class with self.__class__.attribute

  • Test ordering: Use @pytest.mark.incremental at class level for sequential test dependencies

  • 4-step plan-readiness pattern: verify_<feature> -> storagemap -> networkmap -> plan Used when the feature under test is exercised during provider/plan creation (e.g., CA cert field validation). No migration is executed — plan readiness proves the feature works.

  • 5-step pattern: storagemap -> networkmap -> plan -> migrate -> check_vms

  • 6-step shared-disk pattern (Linux): storagemap -> networkmap -> plan -> migrate -> verify_shared_disk_data -> check_vms Shared disk tests insert test_verify_shared_disk_data before test_check_vms. This step mounts, writes, and reads a shared disk from both VMs to verify bidirectional access after migration. Uses verify_shared_disk_data() from utilities/shared_disk.py.

  • 7-step shared-disk pattern (Windows): label_shared_disk -> storagemap -> networkmap -> plan -> migrate -> verify_shared_disk_data -> check_vms Windows shared disk tests prepend test_label_shared_disk which dynamically labels the shared NTFS volume on the source VM via VMware Guest Operations API before migration. Post-migration verification uses verify_shared_disk_data_windows() from utilities/shared_disk.py.

  • 6-step copy-offload pattern: storagemap -> networkmap -> plan -> migrate -> check_xcopy_used -> check_vms test_check_xcopy_used calls verify_xcopy_used() from utilities/copyoffload_migration.py. This step validates the transfer mechanism (infrastructure), not the migrated VM (application), and provides clearer failure diagnostics.

    Copy-offload test implementation:

    • Default approach: Use execute_copyoffload_migration() from utilities/copyoffload_migration.py for standard copy-offload tests. This function handles all orchestration: Migration CR creation, plan secret waiting, log capture callback setup, and migration polling.
    • Populator throttling tests: Use execute_migration_monitoring_populator_inflight() from utilities/copyoffload_migration.py for tests that need to track populator concurrency during migration. This function combines migration execution with populator in-flight monitoring and log capture callback setup, ensuring both concurrency tracking and populate pod logs are collected during migration polling.
    • Concurrent migrations only: For tests running multiple migrations simultaneously (see TestSimultaneousCopyoffloadMigrations.test_migrate_vms_simultaneously in tests/copyoffload/test_copyoffload_migration.py), use the low-level orchestration: wait_for_copyoffload_plan_secret(), create_log_capture_callback(), wait_for_dual_migration_completion() (from utilities/mtv_migration.py for two-plan tests), and wait_for_migration_complate(on_status_poll=...) to manage each migration independently.

    Do not wait for plan secret in create_plan_resource() — Forklift creates the plan populator secret when migration starts, not at Plan Ready.

  • 7-step copy-offload throttling pattern: storagemap -> networkmap -> plan -> migrate -> verify_populator_throttling -> check_xcopy_used -> check_vms Populator throttling tests insert test_verify_populator_throttling after test_migrate_vms and before test_check_xcopy_used. This step calls verify_populator_throttling() from utilities/copyoffload_migration.py to validate per-ESXi-host concurrency limits, PopulatorThrottled events, and sourceHost labels. Requires the populator_inflight_forkliftcontroller fixture.

  • 8-step copy-offload VM+populator throttling pattern: storagemap -> networkmap -> plan -> migrate -> verify_vm_inflight_throttling -> verify_populator_throttling -> check_xcopy_used -> check_vms Combined VM and populator inflight tests (MTV-777) split verification into two steps after test_migrate_vms: test_verify_vm_inflight_throttling (peak concurrent VMs per host) then test_verify_populator_throttling (sourceHost labels, PopulatorThrottled events, peak populate pods). With sequential VMs (VM_INFLIGHT_LIMIT=1), pass min_expected_throttled to verify_populator_throttling() as vm_count * max(0, disks_per_vm - limit) instead of the default pod_count - limit. Requires the vm_populator_inflight_forkliftcontroller fixture.

  • 6-step LUKS pattern: storagemap -> networkmap -> plan -> migrate -> verify_luks_encryption -> check_vms test_verify_luks_encryption calls verify_luks_encryption() from utilities/post_migration.py. LUKS secret setup is handled by the luks_vm_specs fixture in tests/luks/conftest.py, which resolves passphrases (per-VM override → provider fallback) and creates K8s Secrets.

  • 6-step XFS pattern: storagemap -> networkmap -> plan -> migrate -> verify_xfs_version -> check_vms XFS migration tests insert test_verify_xfs_version before test_check_vms. This step executes xfs_info on the migrated VM to verify XFS v4 filesystem compatibility (validates crc=0 in output). Uses check_vm_command_output() from utilities/post_migration.py. Requires xfs_compatibility: True in plan config and xfs_check config dict.

Test method naming: Base tests: test_create_storagemap, test_create_networkmap, test_create_plan, test_migrate_vms, test_check_vms. Shared-disk Linux tests: same through test_migrate_vms, then test_verify_shared_disk_data, test_check_vms. Shared-disk Windows tests: test_label_shared_disk, then the base five through test_migrate_vms, then test_verify_shared_disk_data, test_check_vms. Copy-offload tests: same through test_migrate_vms, then test_check_xcopy_used, test_check_vms. Copy-offload throttling tests: same through test_migrate_vms, then test_verify_populator_throttling, test_check_xcopy_used, test_check_vms. Copy-offload VM+populator throttling tests: same through test_migrate_vms, then test_verify_vm_inflight_throttling, test_verify_populator_throttling, test_check_xcopy_used, test_check_vms. LUKS tests: same through test_migrate_vms, then test_verify_luks_encryption, test_check_vms. XFS tests: same through test_migrate_vms, then test_verify_xfs_version, test_check_vms.

Fixture parameters: Each test method requests only the fixtures it needs. The example shows typical patterns.

Adding New Tests

  1. Add configuration to tests/tests_config/config.py:
tests_params: dict = {
    "test_my_new_test": {
        "virtual_machines": [{"name": "vm-name", "source_vm_power": "on", "guest_agent": True}],
        "warm_migration": False,
    },
}
  1. Create the test file in the feature subdirectory described in Test File Location (MUST) (for example, tests/<feature>/test_<feature>_migration.py)
  2. Create a test class with @pytest.mark.parametrize using class_plan_config and indirect=True
  3. Add pytest markers at class level (tier0, tier1, warm, remote, copyoffload)
  4. Implement the 5 base test methods. Some features need extra validation steps: see Key Patterns for the 6-step shared-disk (Linux), 7-step shared-disk (Windows), copy-offload, and LUKS patterns, or the 7-step copy-offload throttling pattern

VM Configuration Options:

Option Required Values
name Yes VM name in source provider
source_vm_power No "on" or "off"
guest_agent No True if installed
clone No True to clone before migration
disk_type No "thin", "thick-lazy", "thick-eager"
luks No True if VM has LUKS-encrypted disk
luks_passphrase No Per-VM passphrase override
migrate_shared_disks No True for owner VM in shared disk tests

Plan Configuration Options:

Option Required Description
warm_migration No True for warm migration
preserve_static_ips No True to preserve static IP addresses after migration
copyoffload No True to enable copy-offload (XCOPY) migration
xfs_compatibility No True to enable XFS v4 filesystem compatibility
migrate_shared_disks No True to enable shared disk migration at plan level
inventory_timeout No Per-VM Forklift inventory wait timeout, in seconds
clone_to_same_host No True to default VM2+ to VM1's ESXi host; explicit target_esxi_host overrides
disable_drs_for_vms No True to disable vSphere DRS per VM after cloning; not supported for OVA; requires a VMware clone provider
per_nic_network_map No True to create per-NIC network mappings (allows duplicate source network entries in NetworkMap)
skip_clone No True to skip VM cloning in prepared_plan fixture (for plan-readiness tests that use existing VMs)
rdm_as_lun No True to map RDM disks as LUN devices with SCSI bus instead of default virtio

Test Verification Configuration:

Option Required Description
xfs_check Yes* XFS filesystem verification config. Required for XFS tests
xfs_check.command Yes* Command to execute. Required when xfs_check is present
xfs_check.mount_point Yes* Mount point to check. Required when xfs_check is present
xfs_check.expected_output Yes* Expected string in output. Required when xfs_check is present

Fixture Patterns

Test File Location (MUST)

Test files must be placed in feature subdirectories under tests/, not directly in the tests/ root. Each subdirectory groups related tests (e.g., tests/cold/, tests/warm/, tests/copyoffload/).

conftest.py Structure and File Placement (MUST)

Every piece of code has exactly one correct location. Misplaced code is a review blocker.

Code type Correct location Wrong location
Pytest fixtures and hooks conftest.py closest to tests that use them Test files, utility modules
Helper/utility functions utilities/<module>.py conftest.py, test files
Provider-specific logic libs/providers/<provider>.py conftest.py, utilities
Constants Top of the module that owns them, after imports Inline before first usage
Custom exceptions See "Exception Types (MUST)" rule Scattered in other modules

What to flag:

  • Private helper functions (_func) defined in conftest.py — move to utilities/
  • Module-level constants defined in the middle of a file — move after imports
  • See also: "No Inline Imports" and "Validate at Source" rules

Fixture Scopes

Session-scoped - shared across all tests:

@pytest.fixture(scope="session")
def ocp_admin_client():
    return get_cluster_client()

Common: ocp_admin_client, session_uuid, target_namespace, source_provider, destination_provider, fixture_store

Class-scoped - per test class:

Two class-scoped fixtures work together (used with indirect=True parametrization):

  • class_plan_config: Raw test configuration from @pytest.mark.parametrize
  • prepared_plan: Processed config with cloned VMs, updated names, and source_vms_data.
    • Two-phase clone pattern: clone all VMs first, then batch inventory sync via wait_for_cloned_vms_in_forklift_inventory.
    • Applies to all non-OVA class tests with cloned VMs.
    • vSphere MTV-6066 workarounds (host/datastore wait, refresh on timeout) are gated on MTV-6072 via jira_issue_open in utilities/provider_inventory.py; disabled when MTV-6072 is resolved in Jira.
    • Use utilities/jira_helpers.py for Jira runtime checks in fixtures.

Test methods receive prepared_plan which is ready to use:

@pytest.fixture(scope="class")
def prepared_plan(class_plan_config, fixture_store, source_provider, ...):
    plan: dict[str, Any] = deepcopy(class_plan_config)
    # Clone VMs, update names
    plan["source_vms_data"] = {}  # Separate storage for source VM data
    yield plan
    # Track for cleanup

Function-scoped (default) - per test method:

Function-scoped fixtures are rarely needed in this codebase. Most fixtures are session or class scoped. If you need per-test isolation, use function scope but this is uncommon.

cleanup_migrated_vms Fixture

Class-scoped teardown fixture that cleans up migrated VMs after each test class completes.

Usage: Add via @pytest.mark.usefixtures("cleanup_migrated_vms") at class level.

Behavior:

  • Runs after all tests in the class complete (teardown-only fixture)
  • Uses vm_obj.clean_up() from ocp_resources for proper VM cleanup
  • Honors --skip-teardown flag (skips cleanup when flag is set)
  • Session-level teardown catches any leftover VMs not cleaned by class fixtures

fixture_store Structure

{
    "session_uuid": "auto-abc123",
    "base_resource_name": "auto-abc123-vsphere-8-0",
    "teardown": {
        "Namespace": [{"name": "ns1", ...}],
        "VirtualMachine": [{"name": "vm1", ...}],
    },
}

Test Markers

Marker Purpose
tier0 Core functionality (smoke tests)
tier1 Extended functionality tests
warm Warm migration tests
remote Remote cluster tests
copyoffload Copy-offload (XCOPY) tests
copyoffload_sanity Copy-offload sanity subset
copyoffload_snapshots Copy-offload snapshot tests (vSphere)
ca_crt CA certificate field (ca.crt) tests
vsphere VMware vSphere provider-specific tests
shared_disk Shared disk migration tests

Marker requirements for collection-time skipping (MUST):

Provider-type skip logic runs in pytest_collection_modifyitems (in conftest.py) and matches tests by marker keywords. Tests MUST use the correct markers for skipping to work:

  • Warm migration tests → @pytest.mark.warm
  • Copy-offload snapshot tests → @pytest.mark.copyoffload_snapshots
  • CA certificate field tests → @pytest.mark.ca_crt

See the documented hook in conftest.py for how to add new provider-type skip rules.

Test class naming for marker-based features (SHOULD):

Test classes for marker-gated features must include the feature name in the class name:

  • Warm migration → class name must contain Warm (e.g., TestSanityWarmMtvMigration)
  • Copy-offload snapshots → class name must contain both Copyoffload and Snapshot (e.g., TestCopyoffloadThinSnapshotsMigration)
  • Copy-offload → class name must contain Copyoffload (e.g., TestCopyoffloadThinMigration)
  • Tier1 features → class name must include the feature name (e.g., TestLuksColdMigration)

This ensures discoverability and consistency with the markers applied to the class.

@pytest.mark.tier0
@pytest.mark.warm
@pytest.mark.skipif(not get_value_from_py_config("remote_ocp_cluster"), reason="No remote cluster")
class TestRemoteWarmMigration:
    ...

Fixture Scope Changes (MUST)

Changing fixture scope (e.g., session → function, class → session) has performance and correctness implications.

What to flag:

  • Scope changes without a comment explaining why
  • session → function scope changes (creates resources per-test instead of once)
  • function → session scope changes (may introduce shared state issues)

Review Priority

When multiple issues exist, address them in this order:

  1. Correctness — Wrong behavior, data loss, security issues
  2. Architecture — Separation of concerns, file placement, function responsibilities
  3. Determinism — Fallbacks, broad exceptions, .get() defaults on our config
  4. Code quality — Naming, types, duplicate code, dead code
  5. Style — Comments, logging, magic numbers

Parallel Execution (pytest-xdist)

Tests are parallel-safe because:

  • Unique namespaces per session via session_uuid
  • Each worker has isolated fixture_store
  • create_and_store_resource() generates unique names

Rules:

  • Always use fixtures for namespaces (never hardcode)
  • Never share mutable state between tests