CherryPicked: [v2.10] feat: add MTV operator upgrade/update migration testing

README.md

@@ -257,6 +257,7 @@ The Quick Start runs **tier0** tests (smoke tests). You can run other test categ
 | `copyoffload` | Fast migrations via shared storage | Testing storage arrays |
 | `copyoffload_sanity` | Copy-offload sanity subset (see below) | Quick copy-offload validation |
 | `warm` | Warm migrations (VMs stay running) | Specific scenario testing |
+| `upgrade` | Migration across MTV operator upgrades | Validating upgrade compatibility |
 
 ### Copy-Offload Sanity Tests
 
@@ -629,6 +630,49 @@ Configure pod affinity/anti-affinity and node affinity rules.
 
 ---
 
+## Upgrade Tests
+
+Upgrade tests validate that migration resources created on a pre-upgrade MTV version survive the
+operator upgrade and that migrations execute successfully afterward. The test flow is:
+
+1. Create migration resources (StorageMap, NetworkMap, Plan) on the current MTV version
+2. Upgrade the MTV operator to the target version using an external upgrade script
+3. Verify the upgraded operator version and that existing Plan CRs remain in Ready state
+4. Execute the migration on the upgraded operator
+5. Validate the migrated VMs
+
+### Required `--tc` Parameters
+
+Upgrade tests require additional `--tc` parameters beyond the standard cluster and provider options:
+
+| Parameter | Required | Description |
+| --------- | -------- | ----------- |
+| `upgrade_repo_url` | Yes | URL of the Git repository containing the upgrade script |
+| `upgrade_repo_ref` | Yes | Git ref (branch or tag) to checkout |
+| `upgrade_script_path` | Yes | Relative path to the upgrade script within the repository |
+| `mtv_upgrade_to_version` | Yes | Target MTV version to upgrade to (e.g., `2.7.0`) |
+| `mtv_upgrade_to_source` | Yes | MTV source identifier (e.g., `brew`, `released`) |
+| `mtv_upgrade_image_index` | No | Image index override (pass empty string `""` to skip) |
+
+### Example
+
+```bash
+uv run pytest -m upgrade -v \
+  --tc=cluster_host:https://api.your-cluster.com:6443 \
+  --tc=cluster_username:kubeadmin \
+  --tc=cluster_password:${CLUSTER_PASSWORD} \
+  --tc=source_provider:vsphere-8.0.1 \
+  --tc=storage_class:YOUR-STORAGE-CLASS \
+  --tc=upgrade_repo_url:https://github.com/org/mtv-autodeploy.git \
+  --tc=upgrade_repo_ref:main \
+  --tc=upgrade_script_path:scripts/upgrade.sh \
+  --tc=mtv_upgrade_to_version:2.7.0 \
+  --tc=mtv_upgrade_to_source:released \
+  --tc=mtv_upgrade_image_index:""
+```
+
+---
+
 ## Useful Test Options
 
 ### Debug and Troubleshooting Flags

exceptions/exceptions.py

@@ -125,3 +125,7 @@ class InvalidVMNameError(Exception):
 
 class GuestCommandError(Exception):
     """Raised when a command executed via VMware Guest Operations exits with non-zero code."""
+
+
+class MtvUpgradeError(Exception):
+    """Raised when MTV operator upgrade fails."""

pyproject.toml

@@ -64,4 +64,5 @@ dependencies = [
   "filelock>=3.16.0",
   "python-dotenv>=1.2.1",
   "requests>=2.32.5",
+  "gitpython>=3.1.50",
 ]

pytest.ini

@@ -28,5 +28,6 @@ markers =
     rhv: RHV provider-specific tests
     openstack: OpenStack provider-specific tests
     openshift: OpenShift provider-specific tests
+    upgrade: MTV operator upgrade tests
 
 junit_logging = all

tests/tests_config/config.py

@@ -13,7 +13,6 @@
 snapshots_interval: int = 2
 mins_before_cutover: int = 5
 plan_wait_timeout: int = 3600
-
 tests_params: dict = {
     "test_sanity_warm_mtv_migration": {
         "virtual_machines": [
@@ -606,6 +605,32 @@
         "post_hook": {"expected_result": "fail"},
         "expected_migration_result": "fail",
     },
+    "test_shared_disk_rhel_migration": {
+        "virtual_machines": [
+            {
+                "name": "mtv-feature-shared-rhel1",
+                "source_vm_power": "off",
+                "guest_agent": True,
+                "migrate_shared_disks": True,
+            },
+            {
+                "name": "mtv-feature-shared-rhel2",
+                "source_vm_power": "off",
+                "guest_agent": True,
+                "migrate_shared_disks": False,
+            },
+        ],
+        "warm_migration": False,
+        "migrate_shared_disks": True,
+        "shared_disk_device": "/dev/vdc",
+        "target_power_state": "on",
+    },
+    "test_upgrade_cold_migration": {
+        "virtual_machines": [
+            {"name": "mtv-tests-rhel8", "guest_agent": True},
+        ],
+        "warm_migration": False,
+    },
 }
 
 for _dir in dir():

tests/upgrade/conftest.py

@@ -0,0 +1,185 @@
+from __future__ import annotations
+
+import stat
+from typing import TYPE_CHECKING, Any
+
+import pytest
+from git import Repo
+from ocp_resources.network_map import NetworkMap
+from ocp_resources.plan import Plan
+from ocp_resources.storage_map import StorageMap
+from pytest_testconfig import config as py_config
+from simple_logger.logger import get_logger
+
+from utilities.mtv_migration import (
+    create_plan_resource,
+    get_network_migration_map,
+    get_storage_migration_map,
+)
+from utilities.utils import populate_vm_ids
+
+if TYPE_CHECKING:
+    from kubernetes.dynamic import DynamicClient
+
+    from libs.base_provider import BaseProvider
+    from libs.forklift_inventory import ForkliftInventory
+    from libs.providers.openshift import OCPProvider
+
+LOGGER = get_logger(name=__name__)
+
+
+@pytest.fixture(scope="session")
+def upgrade_script_path(tmp_path_factory: pytest.TempPathFactory) -> str:
+    """Clone the mtv-autodeploy repo and return the absolute path to the upgrade script.
+
+    Args:
+        tmp_path_factory (pytest.TempPathFactory): Pytest factory for session-scoped temporary directories.
+
+    Returns:
+        str: Absolute path to the executable upgrade script.
+
+    Raises:
+        ValueError: If required config values are missing or the script is not found after cloning.
+        git.exc.GitCommandError: If the git clone fails.
+    """
+    repo_url: str = py_config.get("upgrade_repo_url", "")
+    repo_ref: str = py_config.get("upgrade_repo_ref", "")
+    script_relative_path: str = py_config.get("upgrade_script_path", "")
+
+    if not repo_url:
+        raise ValueError("upgrade_repo_url must be provided via --tc=upgrade_repo_url:<url>")
+    if not repo_ref:
+        raise ValueError("upgrade_repo_ref must be provided via --tc=upgrade_repo_ref:<ref>")
+    if not script_relative_path:
+        raise ValueError("upgrade_script_path must be provided via --tc=upgrade_script_path:<relative-path>")
+
+    clone_dir = tmp_path_factory.mktemp("mtv-autodeploy")
+
+    LOGGER.info(f"Cloning {repo_url} (ref={repo_ref}) into {clone_dir}")
+    Repo.clone_from(url=repo_url, to_path=str(clone_dir), branch=repo_ref, depth=1)
+
+    script_path = clone_dir / script_relative_path
+    if not script_path.is_file():
+        raise ValueError(f"Upgrade script not found at '{script_path}' after cloning {repo_url}")
+
+    current_mode = script_path.stat().st_mode
+    script_path.chmod(current_mode | stat.S_IXUSR)
+
+    return str(script_path)
+
+
+@pytest.fixture(scope="class")
+def pre_upgrade_storage_map(
+    prepared_plan: dict[str, Any],
+    fixture_store: dict[str, Any],
+    ocp_admin_client: DynamicClient,
+    source_provider: BaseProvider,
+    destination_provider: BaseProvider,
+    source_provider_inventory: ForkliftInventory,
+    target_namespace: str,
+) -> StorageMap:
+    """Create StorageMap resource for upgrade migration tests.
+
+    Args:
+        prepared_plan (dict[str, Any]): Test plan configuration with VM details.
+        fixture_store (dict[str, Any]): Resource tracking dictionary.
+        ocp_admin_client (DynamicClient): OpenShift admin client.
+        source_provider (BaseProvider): Source provider connection.
+        destination_provider (BaseProvider): Destination provider.
+        source_provider_inventory (ForkliftInventory): Provider inventory.
+        target_namespace (str): Target namespace.
+
+    Returns:
+        StorageMap: The created StorageMap resource.
+    """
+    vms: list[str] = [vm["name"] for vm in prepared_plan["virtual_machines"]]
+    return get_storage_migration_map(
+        fixture_store=fixture_store,
+        source_provider=source_provider,
+        destination_provider=destination_provider,
+        source_provider_inventory=source_provider_inventory,
+        ocp_admin_client=ocp_admin_client,
+        target_namespace=target_namespace,
+        vms=vms,
+    )
+
+
+@pytest.fixture(scope="class")
+def pre_upgrade_network_map(
+    prepared_plan: dict[str, Any],
+    fixture_store: dict[str, Any],
+    ocp_admin_client: DynamicClient,
+    source_provider: BaseProvider,
+    destination_provider: BaseProvider,
+    source_provider_inventory: ForkliftInventory,
+    target_namespace: str,
+    multus_network_name: dict[str, str],
+) -> NetworkMap:
+    """Create NetworkMap resource for upgrade migration tests.
+
+    Args:
+        prepared_plan (dict[str, Any]): Test plan configuration with VM details.
+        fixture_store (dict[str, Any]): Resource tracking dictionary.
+        ocp_admin_client (DynamicClient): OpenShift admin client.
+        source_provider (BaseProvider): Source provider connection.
+        destination_provider (BaseProvider): Destination provider.
+        source_provider_inventory (ForkliftInventory): Provider inventory.
+        target_namespace (str): Target namespace.
+        multus_network_name (dict[str, str]): Multus network name mapping.
+
+    Returns:
+        NetworkMap: The created NetworkMap resource.
+    """
+    vms: list[str] = [vm["name"] for vm in prepared_plan["virtual_machines"]]
+    return get_network_migration_map(
+        fixture_store=fixture_store,
+        source_provider=source_provider,
+        destination_provider=destination_provider,
+        source_provider_inventory=source_provider_inventory,
+        ocp_admin_client=ocp_admin_client,
+        target_namespace=target_namespace,
+        multus_network_name=multus_network_name,
+        vms=vms,
+    )
+
+
+@pytest.fixture(scope="class")
+def pre_upgrade_plan_resource(
+    prepared_plan: dict[str, Any],
+    fixture_store: dict[str, Any],
+    ocp_admin_client: DynamicClient,
+    source_provider: BaseProvider,
+    destination_provider: OCPProvider,
+    source_provider_inventory: ForkliftInventory,
+    target_namespace: str,
+    pre_upgrade_storage_map: StorageMap,
+    pre_upgrade_network_map: NetworkMap,
+) -> Plan:
+    """Create MTV Plan CR resource for upgrade migration tests.
+
+    Args:
+        prepared_plan (dict[str, Any]): Test plan configuration with VM details.
+        fixture_store (dict[str, Any]): Resource tracking dictionary.
+        ocp_admin_client (DynamicClient): OpenShift admin client.
+        source_provider (BaseProvider): Source provider connection.
+        destination_provider (OCPProvider): Destination provider.
+        source_provider_inventory (ForkliftInventory): Provider inventory.
+        target_namespace (str): Target namespace.
+        pre_upgrade_storage_map (StorageMap): Storage map for the migration.
+        pre_upgrade_network_map (NetworkMap): Network map for the migration.
+
+    Returns:
+        Plan: The created Plan CR resource.
+    """
+    populate_vm_ids(prepared_plan, source_provider_inventory)
+    return create_plan_resource(
+        ocp_admin_client=ocp_admin_client,
+        fixture_store=fixture_store,
+        source_provider=source_provider,
+        destination_provider=destination_provider,
+        storage_map=pre_upgrade_storage_map,
+        network_map=pre_upgrade_network_map,
+        virtual_machines_list=prepared_plan["virtual_machines"],
+        target_namespace=target_namespace,
+        warm_migration=prepared_plan.get("warm_migration", False),
+    )