test(MTV-5663): add plan archive PVC cleanup regression test

.pre-commit-config.yaml

@@ -41,7 +41,7 @@ repos:
         exclude: ^docs/
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.16.0
+    rev: v0.16.1
     hooks:
       - id: ruff
       - id: ruff-format

AGENTS.md

@@ -598,12 +598,17 @@ with ResourceEditor(node) as editor:
 
 ## Critical Constraints
 
-### Test Execution Prohibition
+### Test Execution Requirements (MUST)
 
-AI must NEVER run tests directly (`pytest`, `uv run pytest`). Tests require live clusters, provider connections, and credentials.
+Tests interact with live OpenShift clusters and source providers. Before running tests:
 
-AI can: Read/analyze/write/fix tests, suggest improvements, review structure
-AI cannot: Execute tests, validate by running
+- **Cluster access:** A dedicated test cluster with `kubeadmin` or equivalent credentials is required
+- **Provider credentials:** A valid `.providers.json` with source provider connection details
+- **Isolation:** Tests create unique namespaces per session (`session_uuid`) to prevent OCP resource collisions between parallel runs
+- **Parallel safety:** Namespace isolation does not prevent source-provider VM conflicts when multiple
+  runs share the same source VM names — use separate provider configurations or VM cloning for true
+  parallel safety
+- **Cleanup:** Tests clean up resources via `fixture_store` teardown — use `--skip-teardown` to preserve resources for debugging
 
 ### No Module-Level Provider Loading (MUST)
 
@@ -708,6 +713,20 @@ namespace = Namespace(client=ocp_admin_client, name="my-namespace")
 namespace.deploy()
 ```
 
+When intentionally deleting a resource mid-test (e.g., testing plan archive+delete), unregister it
+from teardown to prevent session cleanup from operating on a missing resource:
+
+```python
+# Assumes Plan, Migration, archive_plan, get_migration_for_plan are already imported
+from utilities.resources import unregister_teardown_resource
+
+migration_name = get_migration_for_plan(plan).name
+archive_plan(plan=self.plan_resource)
+self.plan_resource.clean_up(wait=True)
+unregister_teardown_resource(fixture_store=fixture_store, kind=Plan.kind, name=self.plan_resource.name)
+unregister_teardown_resource(fixture_store=fixture_store, kind=Migration.kind, name=migration_name)
+```
+
 ## Test Structure Pattern
 
 All tests follow a class-based structure with 5 base test methods:
@@ -1070,12 +1089,14 @@ When multiple issues exist, address them in this order:
 
 ## Parallel Execution (pytest-xdist)
 
-Tests are parallel-safe because:
+Tests are parallel-safe for OpenShift resources because:
 
 - Unique namespaces per session via `session_uuid`
 - Each worker has isolated `fixture_store`
 - `create_and_store_resource()` generates unique names
 
+**Note:** Namespace isolation does not prevent source-provider VM conflicts — see Test Execution Requirements above.
+
 Rules:
 
 - Always use fixtures for namespaces (never hardcode)

conftest.py

@@ -329,11 +329,7 @@ def pytest_collection_modifyitems(session, config, items):
             if source_provider_type != Provider.ProviderType.VSPHERE:
                 vsphere_only_skip = pytest.mark.skip(reason="Test is only applicable to vSphere source providers")
                 for item in items:
-                    if (
-                        "copyoffload" in item.keywords
-                        or "shared_disk" in item.keywords
-                        or "deep_inspection" in item.keywords
-                    ):
+                    if any(kw in item.keywords for kw in ("copyoffload", "shared_disk", "deep_inspection", "aap")):
                         item.add_marker(vsphere_only_skip)
 
             # Skip CA cert tests for providers that don't use CA certificates.

pytest.ini

@@ -32,6 +32,7 @@ markers =
     openshift: OpenShift provider-specific tests
     deep_inspection: Deep Inspection / Conversion CR tests (vSphere only)
     ca_crt: CA certificate field (ca.crt) provider secret tests
+    aap: AAP (Ansible Automation Platform) hook integration tests (vSphere only)
     upgrade: MTV operator upgrade tests
 
 junit_logging = all

tests/hooks/conftest.py

@@ -0,0 +1,301 @@
+"""
+Fixtures for AAP (Ansible Automation Platform) hook integration tests.
+
+Session-scoped fixtures deploy AWX on the cluster and configure MTV
+to use it. Class-scoped fixtures create AAP Hook CRs for each test class.
+"""
+
+from __future__ import annotations
+
+import tempfile
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Generator
+
+import filelock
+import pytest
+from ocp_resources.forklift_controller import ForkliftController
+from ocp_resources.resource import ResourceEditor
+from simple_logger.logger import get_logger
+
+from utilities.aap import (
+    AAP_TEST_PLAYBOOKS_REPO,
+    AWX_ADMIN_USERNAME,
+    AWX_NAMESPACE,
+    AWX_POSTHOOK_PLAYBOOK,
+    AWX_POSTHOOK_TEMPLATE_NAME,
+    AWX_PREHOOK_PLAYBOOK,
+    AWX_PREHOOK_TEMPLATE_NAME,
+    AWX_PROJECT_NAME,
+    create_aap_token_secret,
+    create_awx_auth_token,
+    create_awx_instance,
+    create_awx_inventory,
+    create_awx_job_template,
+    create_awx_project,
+    deploy_awx_via_helm,
+    get_awx_admin_password,
+    is_awx_installed,
+    get_awx_route_url,
+    teardown_awx,
+    wait_for_awx_api_ready,
+    wait_for_awx_project_sync,
+    wait_for_awx_ready,
+)
+from utilities.forklift_controller_populator import ensure_secure_shared_lock_dir
+from utilities.hooks import create_hook_for_plan
+
+if TYPE_CHECKING:
+    from kubernetes.dynamic import DynamicClient
+
+LOGGER = get_logger(__name__)
+
+FORKLIFT_CONTROLLER_NAME: str = (
+    "forklift-controller"  # Well-known name of the ForkliftController CR created by MTV operator
+)
+AAP_FORKLIFT_CONTROLLER_LOCK_TIMEOUT: int = 3600  # seconds; covers entire session-scoped fixture
+
+
+def _get_aap_forklift_controller_lock_path() -> Path:
+    """Return the cross-worker lock path for ForkliftController AAP setting changes.
+
+    Returns:
+        Path: File lock path under a secured shared temp directory.
+    """
+    lock_dir = Path(tempfile.gettempdir()) / "pytest-shared-forklift"
+    ensure_secure_shared_lock_dir(lock_dir=lock_dir)
+    return lock_dir / "aap-settings.lock"
+
+
+@pytest.fixture(scope="session")
+def awx_deployment(
+    ocp_admin_client: "DynamicClient",
+    fixture_store: dict[str, Any],
+) -> Generator[str, None, None]:
+    """Deploy AWX via Helm and return the route URL.
+
+    Installs the AWX Operator, creates an AWX instance with CephFS storage,
+    and waits for all pods to be ready. AWX is a third-party CRD installed
+    at runtime — cleanup is handled by namespace deletion in teardown_awx().
+
+    Args:
+        ocp_admin_client: OpenShift admin client.
+        fixture_store: Fixture store for resource tracking.
+
+    Yields:
+        str: AWX web route URL (https://...).
+    """
+    helm_installed_by_us = not is_awx_installed()
+    awx_cr_created_by_us = False
+    try:
+        deploy_awx_via_helm()
+        awx_cr_created_by_us = create_awx_instance(
+            ocp_admin_client=ocp_admin_client, fixture_store=fixture_store, namespace=AWX_NAMESPACE
+        )
+        wait_for_awx_ready(ocp_admin_client=ocp_admin_client, namespace=AWX_NAMESPACE)
+        awx_url = get_awx_route_url(ocp_admin_client=ocp_admin_client, namespace=AWX_NAMESPACE)
+        password = get_awx_admin_password(ocp_admin_client=ocp_admin_client, namespace=AWX_NAMESPACE)
+        wait_for_awx_api_ready(awx_url=awx_url, username=AWX_ADMIN_USERNAME, password=password)
+        yield awx_url
+    finally:
+        if helm_installed_by_us or awx_cr_created_by_us:
+            teardown_awx()
+
+
+@pytest.fixture(scope="session")
+def awx_api_token(
+    ocp_admin_client: "DynamicClient",
+    awx_deployment: str,
+) -> str:
+    """Create an AWX OAuth2 API token.
+
+    Args:
+        ocp_admin_client: OpenShift admin client.
+        awx_deployment: AWX route URL.
+
+    Returns:
+        str: OAuth2 token for AWX API authentication.
+    """
+    password = get_awx_admin_password(
+        ocp_admin_client=ocp_admin_client,
+        namespace=AWX_NAMESPACE,
+    )
+    return create_awx_auth_token(
+        awx_url=awx_deployment,
+        username=AWX_ADMIN_USERNAME,
+        password=password,
+    )
+
+
+@pytest.fixture(scope="session")
+def awx_job_templates(
+    awx_deployment: str,
+    awx_api_token: str,
+) -> dict[str, int]:
+    """Create AWX project and job templates for pre/post hooks.
+
+    Creates a project from the mtv-aap-test-playbooks git repo,
+    waits for SCM sync, then creates two job templates.
+
+    Args:
+        awx_deployment: AWX route URL.
+        awx_api_token: AWX API token.
+
+    Returns:
+        dict[str, int]: Mapping of hook type to template ID:
+            ``{"pre_hook": <id>, "post_hook": <id>}``
+    """
+    project_id = create_awx_project(
+        awx_url=awx_deployment,
+        token=awx_api_token,
+        name=AWX_PROJECT_NAME,
+        scm_url=AAP_TEST_PLAYBOOKS_REPO,
+    )
+    wait_for_awx_project_sync(
+        awx_url=awx_deployment,
+        token=awx_api_token,
+        project_id=project_id,
+    )
+
+    inventory_id = create_awx_inventory(
+        awx_url=awx_deployment,
+        token=awx_api_token,
+    )
+
+    pre_hook_id = create_awx_job_template(
+        awx_url=awx_deployment,
+        token=awx_api_token,
+        name=AWX_PREHOOK_TEMPLATE_NAME,
+        project_id=project_id,
+        playbook=AWX_PREHOOK_PLAYBOOK,
+        inventory_id=inventory_id,
+    )
+    post_hook_id = create_awx_job_template(
+        awx_url=awx_deployment,
+        token=awx_api_token,
+        name=AWX_POSTHOOK_TEMPLATE_NAME,
+        project_id=project_id,
+        playbook=AWX_POSTHOOK_PLAYBOOK,
+        inventory_id=inventory_id,
+    )
+
+    return {"pre_hook": pre_hook_id, "post_hook": post_hook_id}
+
+
+@pytest.fixture(scope="session")
+def aap_mtv_settings(
+    ocp_admin_client: "DynamicClient",
+    fixture_store: dict[str, Any],
+    mtv_namespace: str,
+    session_uuid: str,
+    awx_deployment: str,
+    awx_api_token: str,
+) -> Generator[None, None, None]:
+    """Configure MTV to use AWX for AAP hooks.
+
+    Creates a session-unique AWX token Secret in the MTV namespace and patches
+    the ForkliftController with ``aap_token_secret_name``, ``aap_url``, and
+    ``aap_insecure_skip_verify``. A file lock serializes ForkliftController
+    changes across pytest-xdist workers for the entire session duration.
+
+    Args:
+        ocp_admin_client: OpenShift admin client.
+        fixture_store: Fixture store for resource tracking.
+        mtv_namespace: MTV operator namespace (e.g., openshift-mtv).
+        session_uuid: Session UUID for unique resource naming.
+        awx_deployment: AWX route URL.
+        awx_api_token: AWX OAuth2 token.
+
+    Yields:
+        None
+
+    Raises:
+        TimeoutError: If the cross-worker file lock cannot be acquired.
+    """
+    lock_path = _get_aap_forklift_controller_lock_path()
+    try:
+        with filelock.FileLock(lock_path, timeout=AAP_FORKLIFT_CONTROLLER_LOCK_TIMEOUT):
+            token_secret_name = f"{session_uuid}-awx-aap"
+            create_aap_token_secret(
+                ocp_admin_client=ocp_admin_client,
+                fixture_store=fixture_store,
+                namespace=mtv_namespace,
+                token=awx_api_token,
+                name=token_secret_name,
+            )
+
+            forklift_controller = ForkliftController(
+                client=ocp_admin_client,
+                name=FORKLIFT_CONTROLLER_NAME,
+                namespace=mtv_namespace,
+                ensure_exists=True,
+            )
+
+            LOGGER.info(
+                f"Patching ForkliftController with aap_url='{awx_deployment}', "
+                f"aap_token_secret_name='{token_secret_name}', and aap_insecure_skip_verify=true"
+            )
+            editor = ResourceEditor(
+                patches={
+                    forklift_controller: {
+                        "spec": {
+                            "aap_url": awx_deployment,
+                            "aap_token_secret_name": token_secret_name,
+                            "aap_insecure_skip_verify": "true",
+                        }
+                    }
+                }
+            )
+            editor.update(backup_resources=True)
+            try:
+                forklift_controller.wait_for_condition(
+                    status=forklift_controller.Condition.Status.TRUE,
+                    condition=forklift_controller.Condition.Type.SUCCESSFUL,
+                    timeout=300,
+                )
+                yield
+            finally:
+                editor.restore()
+    except filelock.Timeout as err:
+        raise TimeoutError(
+            f"Timeout ({AAP_FORKLIFT_CONTROLLER_LOCK_TIMEOUT}s) waiting for ForkliftController "
+            f"AAP settings lock at {lock_path}. Another worker may be running AAP hook tests."
+        ) from err
+
+
+@pytest.fixture(scope="class")
+def aap_hook_refs(
+    prepared_plan: dict[str, Any],
+    fixture_store: dict[str, Any],
+    ocp_admin_client: "DynamicClient",
+    target_namespace: str,
+    awx_job_templates: dict[str, int],
+    aap_mtv_settings: None,
+) -> None:
+    """Create AAP Hook CRs and store references in prepared_plan.
+
+    Creates pre-hook and post-hook Hook CRs with ``spec.aap.jobTemplateId``
+    pointing to AWX job templates. Stores ``_pre_hook_name``,
+    ``_pre_hook_namespace``, ``_post_hook_name``, ``_post_hook_namespace``
+    in the prepared_plan dict for ``test_create_plan`` to use.
+
+    Args:
+        prepared_plan: The prepared migration plan dict.
+        fixture_store: Fixture store for resource tracking.
+        ocp_admin_client: OpenShift admin client.
+        target_namespace: Target namespace for Hook CR creation.
+        awx_job_templates: AWX job template IDs.
+        aap_mtv_settings: Ensures MTV is configured with AAP token.
+    """
+    for hook_type, template_key in [("pre", "pre_hook"), ("post", "post_hook")]:
+        hook_config: dict[str, Any] = {
+            "aap_job_template_id": awx_job_templates[template_key],
+        }
+        hook_name, hook_namespace = create_hook_for_plan(
+            hook_config=hook_config,
+            hook_type=hook_type,
+            fixture_store=fixture_store,
+            ocp_admin_client=ocp_admin_client,
+            target_namespace=target_namespace,
+        )
+        prepared_plan[f"_{hook_type}_hook_name"] = hook_name
+        prepared_plan[f"_{hook_type}_hook_namespace"] = hook_namespace