integration: Abstract the source provider

Adds BaseTestExportProvider class, meant to be implemented by export
providers to be used for testing. Similar to BaseTestImportProvider,
it has initialize and teardown, meant to record initial resources and
flag leaked resources when tests finish.

Moves the src loop device creation logic into the source provider,
served by calling get_test_instance.

TestExportProvider will now report leaked "coriolis-replicator-*"
containers on test teardown.

coriolis/tests/integration/base.py

@@ -15,7 +15,6 @@ Subclasses must be run as root.
 import os
 import time
 import unittest
-import uuid
 from unittest import mock
 
 import oslo_messaging as messaging
@@ -31,7 +30,6 @@ from coriolis.db import api as db_api
 from coriolis.providers import factory as providers_factory
 from coriolis.tests import test_base
 from coriolis.tests.integration import harness
-from coriolis.tests.integration import utils as test_utils
 
 CONF = cfg.CONF
 LOG = logging.getLogger(__name__)
@@ -62,6 +60,7 @@ class CoriolisIntegrationTestBase(test_base.CoriolisBaseTestCase):
         cls._workdir = cls._harness.workdir
         cls._lock_path = cls._harness.lock_path
         cls._api_port = cls._harness.api_port
+        cls._exp_provider = cls._harness.exp_provider
         cls._exp_platform = cls._harness.exp_provider_platform
         cls._exp_conn_info = cls._harness.exp_conn_info
 
@@ -296,29 +295,18 @@ class ReplicaIntegrationTestBase(CoriolisIntegrationTestBase):
     def setUp(self):
         super().setUp()
 
-        self._src_device = test_utils.create_loop_device(
-            self._SRC_DEVICE_SIZE_MB * 1024 * 1024
-        )
-        self.addCleanup(test_utils.remove_loop_device, self._src_device)
-
-        # Write a test pattern on the src device.
-        # Incremental transfer tests update the second chunk (offset=4096).
-        test_utils.write_test_pattern(self._src_device, 8192)
-
-        # 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 = "%s-%s" % (
-            os.path.basename(self._src_device),
-            uuid.uuid4().hex[:8],
+        self._instance_name, extra_source_env = self._exp_provider.get_test_instance(
+            size_mb=self._SRC_DEVICE_SIZE_MB,
         )
+        self.addCleanup(self._exp_provider.delete_test_instance, self._instance_name)
+
         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={
-                "instance_block_devices": {self._instance_name: [self._src_device]},
+                **extra_source_env,
                 **self._EXTRA_SOURCE_ENVIRONMENT,
             },
         )
@@ -338,6 +326,11 @@ class ReplicaIntegrationTestBase(CoriolisIntegrationTestBase):
             mocker.start()
             self.addCleanup(mocker.stop)
 
+    @property
+    def _src_device(self):
+        """Local loop device path backing the source instance, if any."""
+        return self._exp_provider.get_test_instance_device(self._instance_name)
+
     @property
     def _dst_device(self):
         """First destination dev path from the transfer's volumes_info."""

coriolis/tests/integration/harness.py

@@ -370,10 +370,16 @@ class _IntegrationHarness:
         # Init exporter.
         self.exp_provider_class = _get_provider(_TEST_EXPORT_PROVIDER)
         self.exp_provider_platform = self.exp_provider_class.platform
+        self.exp_provider = providers_factory.get_provider(
+            self.exp_provider_platform,
+            constants.PROVIDER_TYPE_TRANSFER_EXPORT,
+            event_handler=mock.MagicMock(),
+        )
         self.exp_conn_info = {
             "pkey_path": self.ssh_key_path,
             "role": "source",
         }
+        self.exp_provider.initialize(self.exp_conn_info)
 
         # Init importer.
         imp_provider_cls = providers_config["destination"]["provider_cls"]
@@ -412,6 +418,7 @@ class _IntegrationHarness:
         sqlalchemy_api._facade = None
         rpc_module._TRANSPORT = None
 
+        atexit.register(self.exp_provider.teardown, self.exp_conn_info)
         atexit.register(self.imp_provider.teardown, self.imp_conn_info)
         atexit.register(self._teardown)
 

coriolis/tests/integration/provider_test_base.py

@@ -2,12 +2,13 @@
 # All Rights Reserved.
 
 """
-Abstract base class for test import providers.
+Abstract base classes for test import / export providers.
 
 Based on the Base* provider convention from coriolis/providers/base.py.
 
-The BaseTestImportProvider contains provider-specific logic not currently
-defined in the import providers, meant to be used for testing-only purposes:
+BaseTestImportProvider and BaseTestExportProvider contain provider-specific
+logic not currently defined in the import / export providers, meant to be
+used for testing-only purposes:
     - detect leaked resources
     - delete deployed replicas
 """
@@ -19,6 +20,49 @@ from oslo_log import log as logging
 LOG = logging.getLogger(__name__)
 
 
+class BaseTestExportProvider(abc.ABC):
+    def initialize(self, connection_info: dict):
+        """One-time initialization, before any tests run.
+
+        Can be used to list the current resources on the source provider,
+        which can then be used to check if any test resources leaked and
+        clean them.
+        """
+
+    def teardown(self, connection_info: dict):
+        """One-time teardown called at atexit.
+
+        Can be used to check and clean any leaked test resources.
+        """
+
+    def check_prerequisites(self):
+        """Raise ``unittest.SkipTest`` if required infrastructure is absent."""
+
+    def get_test_instance(self, size_mb: int):
+        """Return ``(instance_name, source_environment)`` to use for one test.
+
+        *size_mb* is a hint for providers that create disposable block devices;
+        providers backed by a pre-existing VM ignore it.
+
+        The returned ``source_environment`` is merged into the transfer's
+        source_environment; it is only non-empty for providers that need to
+        advertise something about the instance they just created (e.g.: the
+        core test provider's ``instance_block_devices``).
+        """
+        raise NotImplementedError
+
+    def delete_test_instance(self, instance_name: str):
+        """Release resources allocated by ``get_test_instance()``, if any."""
+
+    def get_test_instance_device(self, instance_name: str):
+        """Return the local block-device path backing *instance_name*.
+
+        Only meaningful for providers that back test instances with a host-readable
+        device (the core test provider); other providers return ``None``.
+        """
+        return None
+
+
 class BaseTestImportProvider(abc.ABC):
     def initialize(self, connection_info: dict):
         """One-time initialization, before any tests run.

coriolis/tests/integration/test_provider/exp.py

@@ -11,6 +11,7 @@ manage the coriolis-replicator service and perform disk replication.
 import csv
 import io
 import os
+import unittest
 import uuid
 
 import paramiko
@@ -28,6 +29,7 @@ from coriolis.providers.base import (
     BaseReplicaExportValidationProvider,
     BaseUpdateSourceReplicaProvider,
 )
+from coriolis.tests.integration import provider_test_base
 from coriolis.tests.integration import utils as test_utils
 
 CONF = cfg.CONF
@@ -42,6 +44,9 @@ _TEST_NIC = {
     "mac_address": "fa:16:3e:12:34:56",
 }
 
+# Name prefixes used by deploy_replica_source_resources.
+_CONTAINER_PREFIXES = ("coriolis-replicator-",)
+
 
 class TestExportProvider(
     BaseEndpointInstancesProvider,
@@ -50,6 +55,7 @@ class TestExportProvider(
     BaseUpdateSourceReplicaProvider,
     BaseReplicaExportProvider,
     BaseReplicaExportValidationProvider,
+    provider_test_base.BaseTestExportProvider,
 ):
     """Source-side provider backed by a local loop device.
 
@@ -72,6 +78,54 @@ class TestExportProvider(
 
     def __init__(self, event_handler):
         self._event_handler = event_handler
+        self._test_devices = {}  # instance_name -> loop device path
+
+    # BaseTestExportProvider - test only
+
+    def initialize(self, connection_info: dict):
+        self._initial_containers = test_utils.list_containers(_CONTAINER_PREFIXES)
+
+    def teardown(self, connection_info: dict):
+        new_containers = test_utils.list_containers(_CONTAINER_PREFIXES)
+        leaked_containers = new_containers - self._initial_containers
+
+        if not leaked_containers:
+            return
+
+        for name in leaked_containers:
+            test_utils.remove_container(name)
+
+        raise AssertionError(
+            "Found leaked containers during teardown: %s" % leaked_containers
+        )
+
+    def check_prerequisites(self):
+        if not test_utils.container_image_exists(test_utils.DATA_MINION_IMAGE):
+            raise unittest.SkipTest(
+                "Docker image '%s' not found; build it with: "
+                "docker build -t %s "
+                "coriolis/tests/integration/dockerfiles/data-minion/"
+                % (test_utils.DATA_MINION_IMAGE, test_utils.DATA_MINION_IMAGE)
+            )
+
+    def get_test_instance(self, size_mb: int):
+        device = test_utils.create_loop_device(size_mb * 1024 * 1024)
+        test_utils.write_test_pattern(device, 8192)
+
+        instance_name = "%s-%s" % (os.path.basename(device), uuid.uuid4().hex[:8])
+        self._test_devices[instance_name] = device
+
+        return instance_name, {
+            "instance_block_devices": {instance_name: [device]},
+        }
+
+    def delete_test_instance(self, instance_name: str):
+        device = self._test_devices.pop(instance_name, None)
+        if device:
+            test_utils.remove_loop_device(device)
+
+    def get_test_instance_device(self, instance_name: str):
+        return self._test_devices.get(instance_name)
 
     def _event_manager(self):
         return events.EventManager(self._event_handler)