integration: Move destination device initialization to the provider

coriolis/tests/integration/README.md

@@ -98,6 +98,39 @@ sudo tox -e integration -- --no-discover coriolis.tests.integration.transfers.te
 > `sudo` is required because `tox` itself must run as root so that the
 > test process inherits root privileges.
 
+## Using an external destination provider
+
+By default, the harness uses the built-in Docker test provider for both source
+and destination. To run the integration suite against a real destination
+provider, install the provider package via `CORIOLIS_PROVIDER_PACKAGE` and
+supply provider configuration via `CORIOLIS_PROVIDERS_YAML`.
+
+### What the harness does with `providers.yaml`
+
+1. Registers the destination provider class with `oslo.config`.
+2. Creates a destination endpoint with `destination.connection_info`.
+3. Uses `destination.environment` as `destination_environment` and
+   `destination.storage_mappings` as `storage_mappings` for each transfer.
+
+### Running
+
+Set `CORIOLIS_PROVIDER_PACKAGE` to a local path or any pip-compatible specifier
+(`git+file://`, `git+https://`, etc.); tox installs it into the virtualenv
+before running the tests. Leave it unset to use only the built-in test provider.
+
+```bash
+sudo -E CORIOLIS_PROVIDER_PACKAGE=/path/to/provider \
+  CORIOLIS_PROVIDERS_YAML=./providers.yaml tox -e integration
+```
+
+Supply `CORIOLIS_CONFIG_FILE` when provider-specific configurations are required:
+
+```bash
+sudo -E CORIOLIS_PROVIDER_PACKAGE=/path/to/provider \
+  CORIOLIS_CONFIG_FILE=./provider.conf \
+  CORIOLIS_PROVIDERS_YAML=./providers.yaml tox -e integration
+```
+
 ## Test modules
 
 ### No block devices (extend `CoriolisIntegrationTestBase`)

coriolis/tests/integration/base.py

@@ -13,10 +13,10 @@
 """
 
 import os
-import subprocess
 import time
 import unittest
 from unittest import mock
+import uuid
 
 from coriolisclient import client as coriolis_client
 from keystoneauth1 import session as ks_session
@@ -67,8 +67,11 @@ def setUpClass(cls):
         cls._exp_platform = cls._harness.exp_provider_platform
         cls._exp_conn_info = cls._harness.exp_conn_info
 
+        cls._imp_provider = cls._harness.imp_provider
         cls._imp_platform = cls._harness.imp_provider_platform
         cls._imp_conn_info = cls._harness.imp_conn_info
+        cls._imp_env_options = cls._harness.imp_env_options
+        cls._storage_mappings = cls._harness.imp_storage_mappings
 
         cls._client = cls.get_client()
 
@@ -135,15 +138,19 @@ def _create_transfer(
             destination_environment=None, **kwargs):
         """Create a Replica transfer object and return its ID."""
 
+        destination_environment = (
+            destination_environment or self._imp_env_options
+        )
+
         transfer = self._client.transfers.create(
             origin_endpoint_id=src_id,
             destination_endpoint_id=dst_id,
             source_environment=source_environment or {},
-            destination_environment=destination_environment or {},
+            destination_environment=destination_environment,
             instances=instances,
             transfer_scenario=constants.TRANSFER_SCENARIO_REPLICA,
             network_map={},
-            storage_mappings={},
+            storage_mappings=self._storage_mappings,
             notes="integration test replica",
             skip_os_morphing=True,
             **kwargs,
@@ -192,7 +199,7 @@ def _safe_delete_pool(cls, pool_id):
             cls._client.minion_pools.delete(pool_id)
 
     @classmethod
-    def _wait_for_pool(cls, pool_id, terminal_statuses, timeout=180):
+    def _wait_for_pool(cls, pool_id, terminal_statuses, timeout=600):
         """Poll the DB until *pool_id* reaches one of *terminal_statuses*.
 
         :returns: minion pool ORM object.
@@ -239,17 +246,7 @@ class ReplicaIntegrationTestBase(CoriolisIntegrationTestBase):
 
     @classmethod
     def setUpClass(cls):
-        result = subprocess.run(
-            ["docker", "image", "inspect", test_utils.DATA_MINION_IMAGE],
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-        )
-        if result.returncode != 0:
-            raise unittest.SkipTest(
-                "Docker image not found; build it with: "
-                "docker build -t %s "
-                "coriolis/tests/integration/dockerfiles/data-minion/"
-                % test_utils.DATA_MINION_IMAGE)
+        harness._IntegrationHarness.get().imp_provider.check_prerequisites()
 
         super().setUpClass()
 
@@ -289,8 +286,6 @@ def setUp(self):
 
         self._src_device = test_utils.add_scsi_debug_device()
         self.addCleanup(test_utils.remove_scsi_debug_device)
-        self._dst_device = test_utils.add_scsi_debug_device()
-        self.addCleanup(test_utils.remove_scsi_debug_device)
 
         # Write a test pattern on the src device.
         # Incremental transfer tests update the second chunk (offset=4096).
@@ -301,15 +296,19 @@ def setUp(self):
         # Create transfer replica.
         # Use basename as instance name; real VM names do not contain slashes,
         # and some providers use the name as is in resource indentifiers.
-        self._instance_name = os.path.basename(self._src_device)
+        self._instance_name = "%s-%s" % (
+            os.path.basename(self._src_device), uuid.uuid4().hex[:8])
         self._transfer = self._create_transfer(
             self._src_endpoint.id,
             self._dst_endpoint.id,
             instances=[self._instance_name],
             destination_minion_pool_id=self._pool_id,
             source_environment={"block_device_path": self._src_device},
-            destination_environment={"devices": [self._dst_device]},
         )
+        # Safety-net cleanup for destination devices allocated by the provider.
+        # Must be registered after the transfer, so it runs (LIFO) before the
+        # transfer delete, while the volumes_info is still in the DB.
+        self.addCleanup(self._cleanup_provider_dst_devices)
 
         # mock a few commands that are going to be ran through ssh; they won't
         # pass anyway.
@@ -325,6 +324,48 @@ def setUp(self):
             mocker.start()
             self.addCleanup(mocker.stop)
 
+    @property
+    def _dst_device(self):
+        """First destination dev path from the transfer's volumes_info."""
+        ctxt = self._get_db_context()
+
+        transfer = db_api.get_transfer(
+            ctxt, self._transfer.id, include_task_info=True)
+        info = transfer.get("info", {}).get(self._instance_name, {})
+        for vol in info.get("volumes_info", []):
+            if vol.get("volume_dev"):
+                return vol["volume_dev"]
+
+        return None
+
+    def _cleanup_provider_dst_devices(self):
+        """Remove any devices the provider allocated for this test."""
+        ctxt = self._get_db_context()
+
+        try:
+            transfer = db_api.get_transfer(
+                ctxt, self._transfer.id, include_task_info=True)
+            volumes_info = transfer.get("info", {}).get(
+                self._instance_name, {}).get('volumes_info', [])
+        except Exception as ex:
+            LOG.warn("Could not get volumes info for cleanup. Ex: %s", ex)
+            return
+
+        if not volumes_info:
+            LOG.info("No volume info. Nothing to cleanup.")
+            return
+
+        try:
+            self._imp_provider.delete_replica_disks(
+                ctxt,
+                self._imp_conn_info,
+                self._imp_env_options,
+                volumes_info,
+            )
+        except Exception as ex:
+            LOG.warn(
+                "Could not clean up provider dst devices. Ex: %s", ex)
+
     def _execute_and_wait(self, transfer_id, timeout=300):
         """Trigger one execution of *transfer_id* and wait for completion."""
         execution = self._client.transfer_executions.create(
@@ -426,6 +467,9 @@ def _cleanup_deployment(self, deployment_id):
         that occurs when a deployment is still in-flight at cleanup time, which
         can happen with slow providers when a test fails or times out before
         the deployment completes.
+
+        Calls ``_imp_provider.delete_deployed_instance`` for every deployment
+        instance, so that finalized VMs at the destination are destroyed.
         """
         ctxt = self._get_db_context()
         deployment = db_api.get_deployment(ctxt, deployment_id)
@@ -434,13 +478,24 @@ def _cleanup_deployment(self, deployment_id):
                 "Deployment '%s' not found. Skip cleanup.", deployment_id)
             return
 
+        instances = list(deployment.instances or [])
+
         if deployment.last_execution_status in (
                 constants.ACTIVE_EXECUTION_STATUSES):
             self._client.deployments.cancel(deployment_id)
             self.wait_for_deployment(deployment_id, timeout=60)
 
         self._client.deployments.delete(deployment_id)
 
+        for instance_name in instances:
+            try:
+                self._imp_provider.delete_deployed_instance(
+                    self._imp_conn_info, instance_name)
+            except Exception as ex:
+                LOG.warning(
+                    "Could not clean up deployed instance '%s': %s",
+                    instance_name, ex)
+
     def wait_for_deployment(self, deployment_id, timeout=300,
                             desired_statuses=None):
         """Block until *deployment_id* reaches any terminal state.

coriolis/tests/integration/deployments/test_osmorphing.py

@@ -7,9 +7,11 @@
 installation in the target OS.
 """
 
+import unittest
 import uuid
 
 from coriolis.tests.integration import base as integration_base
+from coriolis.tests.integration import harness as integration_harness
 from coriolis.tests.integration import utils as test_utils
 
 
@@ -19,6 +21,14 @@ class OsMorphingDeploymentTest(integration_base.ReplicaIntegrationTestBase):
     # any new packages to be added during OS morphing.
     _SCSI_DEBUG_SIZE_MB = 256
 
+    @classmethod
+    def setUpClass(cls):
+        harness = integration_harness._IntegrationHarness.get()
+        if not harness.uses_test_import_provider():
+            raise unittest.SkipTest(
+                "OS morphing tests require local disk access")
+        super().setUpClass()
+
     def setUp(self):
         super().setUp()
         test_utils.write_os_image_to_disk(self._src_device, "ubuntu:24.04")