integration: Adds BaseSourceMinionPoolProvider to test source provider

TestExportProvider now implements BaseSourceMinionPoolProvider,
mirroring TestImportProvider's container-backed minion pattern.

Creates common.py for the test providers, containing common code between
then, which include the minion pool related code.

coriolis/tests/integration/test_provider/common.py

@@ -0,0 +1,183 @@
+# Copyright 2026 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+"""
+Shared functionality between the import and export test providers.
+"""
+
+import os
+import uuid
+
+import paramiko
+
+from coriolis import utils as coriolis_utils
+from coriolis.tests.integration import utils as test_utils
+
+
+class TestProviderMixin:
+    """Shared provider methods between TestImportProvider and TestExportProvider."""
+
+    def __init__(self, event_handler):
+        self._event_handler = event_handler
+
+    # BaseProvider / BaseEndpointProvider
+
+    def get_connection_info_schema(self):
+        return {
+            "type": "object",
+            "properties": {
+                "pkey_path": {"type": "string"},
+                "role": {"type": "string"},
+            },
+            "required": ["pkey_path"],
+        }
+
+    def validate_connection(self, ctxt, connection_info):
+        pkey_path = connection_info["pkey_path"]
+        if not os.path.exists(pkey_path):
+            raise ValueError("SSH private key not found: %s" % pkey_path)
+
+    def _create_minion(
+        self,
+        name_prefix,
+        connection_info,
+        devices=None,
+        volumes=None,
+        device_cgroup_rules=None,
+    ):
+        """Create a data-minion container and return its SSH connection info."""
+        pkey_path = connection_info["pkey_path"]
+        container_name = "%s-%s" % (name_prefix, uuid.uuid4().hex[:8])
+
+        container_id = test_utils.run_container(
+            test_utils.DATA_MINION_IMAGE,
+            container_name,
+            is_systemd=True,
+            ssh_key=f"{pkey_path}.pub",
+            devices=devices,
+            volumes=volumes,
+            device_cgroup_rules=device_cgroup_rules,
+        )
+
+        try:
+            container_ip = test_utils.get_container_ip(container_id)
+            test_utils.wait_for_ssh(container_ip, 22, "root", pkey_path)
+
+            pkey = paramiko.RSAKey.from_private_key_file(pkey_path)
+            ssh_conn_info = {
+                "ip": container_ip,
+                "port": 22,
+                "username": "root",
+                "pkey": coriolis_utils.serialize_key(pkey),
+            }
+
+            return {
+                "container_id": container_id,
+                "ssh_connection_info": ssh_conn_info,
+            }
+        except Exception:
+            test_utils.remove_container(container_id)
+            raise
+
+    # BaseSourceMinionPoolProvider / BaseDestinationMinionPoolProvider
+
+    def validate_minion_compatibility_for_transfer(
+        self, ctxt, connection_info, export_info, environment_options, minion_properties
+    ):
+        pass
+
+    def validate_minion_pool_environment_options(
+        self, ctxt, connection_info, environment_options
+    ):
+        pass
+
+    def set_up_pool_shared_resources(
+        self, ctxt, connection_info, environment_options, pool_identifier
+    ):
+        return {}
+
+    def tear_down_pool_shared_resources(
+        self, ctxt, connection_info, environment_options, pool_shared_resources
+    ):
+        pass
+
+    def delete_minion(self, ctxt, connection_info, minion_properties):
+        container_id = (minion_properties or {}).get("container_id")
+        if container_id:
+            test_utils.remove_container(container_id)
+
+    def shutdown_minion(self, ctxt, connection_info, minion_properties):
+        container_id = (minion_properties or {}).get("container_id")
+        if container_id:
+            test_utils.stop_container(container_id)
+
+    def start_minion(self, ctxt, connection_info, minion_properties):
+        container_id = (minion_properties or {}).get("container_id")
+        if container_id:
+            test_utils.start_container(container_id)
+
+    def attach_volumes_to_minion(
+        self,
+        ctxt,
+        connection_info,
+        minion_properties,
+        minion_connection_info,
+        volumes_info,
+    ):
+        container_id = minion_properties["container_id"]
+
+        for vol in volumes_info:
+            if "volume_dev" in vol:
+                # Destination side: the device was already resolved by
+                # deploy_replica_disks, or left empty for a shared disk owned by another
+                # instance of a clustered transfer, in which case there is nothing to
+                # attach here.
+                device_path = vol["volume_dev"]
+                if not device_path:
+                    continue
+            else:
+                # Source side: derive it from the disk_id.
+                device_path = "/dev/%s" % vol["disk_id"]
+
+            test_utils.hotplug_device_to_container(container_id, device_path)
+            vol["volume_dev"] = device_path
+
+        return {
+            "minion_properties": minion_properties,
+            "volumes_info": volumes_info,
+        }
+
+    def detach_volumes_from_minion(
+        self,
+        ctxt,
+        connection_info,
+        minion_properties,
+        minion_connection_info,
+        volumes_info,
+    ):
+        container_id = (minion_properties or {}).get("container_id")
+        if not container_id:
+            return
+
+        for vol in volumes_info or []:
+            dev_path = vol.get("volume_dev")
+            if not dev_path:
+                continue
+
+            test_utils.unplug_device_from_container(container_id, dev_path)
+
+        return {
+            "minion_properties": minion_properties,
+            "volumes_info": volumes_info,
+        }
+
+    def healthcheck_minion(
+        self, ctxt, connection_info, minion_properties, minion_connection_info
+    ):
+        ip = minion_connection_info.get("ip")
+        port = minion_connection_info.get("port", 22)
+        username = minion_connection_info.get("username", "root")
+        pkey = minion_connection_info.get("pkey")
+
+        client = coriolis_utils.connect_ssh(ip, port, username, pkey=pkey)
+        client.close()

coriolis/tests/integration/test_provider/exp.py

@@ -14,7 +14,6 @@ import os
 import unittest
 import uuid
 
-import paramiko
 from oslo_config import cfg
 from oslo_log import log as logging
 
@@ -27,10 +26,12 @@ from coriolis.providers.base import (
     BaseEndpointSourceOptionsProvider,
     BaseReplicaExportProvider,
     BaseReplicaExportValidationProvider,
+    BaseSourceMinionPoolProvider,
     BaseUpdateSourceReplicaProvider,
 )
 from coriolis.tests.integration import provider_test_base
 from coriolis.tests.integration import utils as test_utils
+from coriolis.tests.integration.test_provider import common
 
 CONF = cfg.CONF
 LOG = logging.getLogger(__name__)
@@ -49,6 +50,7 @@ _CONTAINER_PREFIXES = ("coriolis-replicator-",)
 
 
 class TestExportProvider(
+    common.TestProviderMixin,
     BaseEndpointInstancesProvider,
     BaseEndpointInventoryExportProvider,
     BaseEndpointSourceOptionsProvider,
@@ -56,6 +58,7 @@ class TestExportProvider(
     BaseReplicaExportProvider,
     BaseReplicaExportValidationProvider,
     provider_test_base.BaseTestExportProvider,
+    BaseSourceMinionPoolProvider,
 ):
     """Source-side provider backed by a local loop device.
 
@@ -133,20 +136,13 @@ class TestExportProvider(
     def _make_replicator(self, conn_info, event_mgr, volumes_info, repl_state):
         """Build a Replicator that connects via SSH to *conn_info*.
 
-        *conn_info* must contain ``ip``, ``port``, ``username``, and
-        ``pkey_path`` keys. An optional ``use_tunnel`` key forces the
-        replicator client to connect through an SSH tunnel instead of
-        directly to the replicator's TCP port.
+        *conn_info* must contain ``ip``, ``port``, ``username``, and a ``pkey``, as
+        returned by ``TestProviderMixin._create_minion``'s ``ssh_connection_info``.
+        An optional ``use_tunnel`` key forces the replicator client to connect through
+        an SSH tunnel instead of directly to the replicator's TCP port.
         """
-        pkey = paramiko.RSAKey.from_private_key_file(conn_info["pkey_path"])
-        repl_conn_info = {
-            "ip": conn_info["ip"],
-            "port": conn_info.get("port", 22),
-            "username": conn_info.get("username", "root"),
-            "pkey": pkey,
-        }
         return replicator_module.Replicator(
-            repl_conn_info,
+            conn_info,
             event_mgr,
             volumes_info,
             repl_state,
@@ -154,23 +150,6 @@ class TestExportProvider(
             _allow_loop_devices=True,
         )
 
-    # BaseProvider / BaseEndpointProvider
-
-    def get_connection_info_schema(self):
-        return {
-            "type": "object",
-            "properties": {
-                "pkey_path": {"type": "string"},
-                "role": {"type": "string"},
-            },
-            "required": ["pkey_path"],
-        }
-
-    def validate_connection(self, ctxt, connection_info):
-        pkey_path = connection_info["pkey_path"]
-        if not os.path.exists(pkey_path):
-            raise ValueError("SSH private key not found: %s" % pkey_path)
-
     # BaseExportInstanceProvider
 
     def get_source_environment_schema(self):
@@ -313,28 +292,17 @@ class TestExportProvider(
     ):
         block_devices = source_environment.get("instance_block_devices", {})
         block_device_paths = block_devices.get(export_info["instance_name"], [])
-        pkey_path = connection_info["pkey_path"]
-
-        container_name = "coriolis-replicator-%s" % uuid.uuid4().hex[:8]
-        container_id = test_utils.run_container(
-            test_utils.DATA_MINION_IMAGE,
-            container_name,
-            is_systemd=True,
-            ssh_key=f"{pkey_path}.pub",
+
+        info = self._create_minion(
+            "coriolis-replicator",
+            connection_info,
             devices=block_device_paths,
         )
+        container_id = info["container_id"]
+        src_conn_info = info["ssh_connection_info"]
+        src_conn_info["use_tunnel"] = source_environment.get("use_tunnel", False)
 
         try:
-            container_ip = test_utils.get_container_ip(container_id)
-            test_utils.wait_for_ssh(container_ip, 22, "root", pkey_path)
-
-            src_conn_info = {
-                "ip": container_ip,
-                "port": 22,
-                "username": "root",
-                "pkey_path": pkey_path,
-                "use_tunnel": source_environment.get("use_tunnel", False),
-            }
             replicator = self._make_replicator(
                 src_conn_info, self._event_manager(), [], None
             )
@@ -375,13 +343,34 @@ class TestExportProvider(
     ):
         repl_state = _extract_repl_state(volumes_info) if incremental else None
 
+        disk_mappings = source_resources.get("disk_mappings")
+        reused_minion = disk_mappings is None
+        if disk_mappings is None:
+            # Minion pool case: "source_resources" only carries the pool minion's
+            # container_id. the block devices were never attached to the container,
+            # so hotplug them now.
+            container_id = source_resources["container_id"]
+            block_devices = source_environment.get("instance_block_devices", {})
+            block_device_paths = block_devices.get(instance_name, [])
+            for path in block_device_paths:
+                test_utils.hotplug_device_to_container(container_id, path)
+
+            disk_mappings = {
+                os.path.basename(path): path for path in block_device_paths
+            }
+
         replicator = self._make_replicator(
             source_conn_info, self._event_manager(), volumes_info, repl_state
         )
         replicator.init_replicator()
+        if reused_minion and incremental:
+            # The pool minion's replicator process persists across executions and it
+            # only computes whole-disk checksums once, at process startup. Without
+            # restarting it, "verify_disk_integrity" would keep comparing against that
+            # stale, pre-sync checksum on every subsequent execution against the minion.
+            replicator.update_state(repl_state or [], restart=True)
         replicator.wait_for_chunks()
 
-        disk_mappings = source_resources.get("disk_mappings", {})
         source_volumes_info = [
             {
                 "disk_id": vol["disk_id"],
@@ -429,6 +418,45 @@ class TestExportProvider(
     ):
         return {}
 
+    # BaseSourceMinionPoolProvider
+
+    def get_minion_pool_environment_schema(self):
+        return self.get_source_environment_schema()
+
+    def get_minion_pool_options(
+        self, ctxt, connection_info, env=None, option_names=None
+    ):
+        return self.get_source_environment_options(
+            ctxt, connection_info, env, option_names
+        )
+
+    def create_minion(
+        self,
+        ctxt,
+        connection_info,
+        environment_options,
+        pool_identifier,
+        pool_os_type,
+        pool_shared_resources,
+        new_minion_identifier,
+    ):
+        # Devices are hotplugged after container creation via mknod / nsenter.
+        # We must pre-authorize all block devices through the
+        # --device-cgroup-rule option, otherwise any device added will be
+        # inaccessible ("operation not permitted" error on open).
+        result = self._create_minion(
+            "coriolis-pool-minion",
+            connection_info,
+            device_cgroup_rules=["b *:* rwm"],
+        )
+
+        return {
+            "connection_info": result["ssh_connection_info"],
+            "minion_provider_properties": {
+                "container_id": result["container_id"],
+            },
+        }
+
 
 # Helpers
 def _get_block_device_size(device):

coriolis/tests/integration/test_provider/imp.py

@@ -11,13 +11,10 @@ target_conn_info that BackupWritersFactory expects.
 
 import os
 import unittest
-import uuid
 
-import paramiko
 from oslo_log import log as logging
 
 from coriolis import constants
-from coriolis import utils as coriolis_utils
 from coriolis.providers import backup_writers
 from coriolis.providers.base import (
     BaseDestinationMinionPoolProvider,
@@ -31,7 +28,7 @@ from coriolis.providers.base import (
 )
 from coriolis.tests.integration import provider_test_base
 from coriolis.tests.integration import utils as test_utils
-from coriolis.tests.integration.test_provider import osmorphing
+from coriolis.tests.integration.test_provider import common, osmorphing
 
 LOG = logging.getLogger(__name__)
 
@@ -47,6 +44,7 @@ _CONTAINER_PREFIXES = (
 
 
 class TestImportProvider(
+    common.TestProviderMixin,
     BaseEndpointProvider,
     BaseEndpointDestinationOptionsProvider,
     BaseEndpointNetworksProvider,
@@ -76,9 +74,6 @@ class TestImportProvider(
 
     platform = "test-dest"
 
-    def __init__(self, event_handler):
-        self._event_handler = event_handler
-
     @classmethod
     def supports_shared_disks(cls) -> bool:
         return True
@@ -111,23 +106,6 @@ class TestImportProvider(
                 % (test_utils.DATA_MINION_IMAGE, test_utils.DATA_MINION_IMAGE)
             )
 
-    # BaseProvider / BaseEndpointProvider
-
-    def get_connection_info_schema(self):
-        return {
-            "type": "object",
-            "properties": {
-                "pkey_path": {"type": "string"},
-                "role": {"type": "string"},
-            },
-            "required": ["pkey_path"],
-        }
-
-    def validate_connection(self, ctxt, connection_info):
-        pkey_path = connection_info["pkey_path"]
-        if not os.path.exists(pkey_path):
-            raise ValueError("SSH private key not found: %s" % pkey_path)
-
     # BaseImportInstanceProvider
 
     def get_target_environment_schema(self):
@@ -244,36 +222,17 @@ class TestImportProvider(
         setup_writer=True,
         writer_backend=backup_writers.BACKUP_WRITER_HTTP,
     ):
-        pkey_path = connection_info["pkey_path"]
-        container_name = "%s-%s" % (name_prefix, uuid.uuid4().hex[:8])
-
-        container_id = test_utils.run_container(
-            test_utils.DATA_MINION_IMAGE,
-            container_name,
-            is_systemd=True,
-            ssh_key=f"{pkey_path}.pub",
+        info = super()._create_minion(
+            name_prefix,
+            connection_info,
             devices=devices,
             volumes=volumes,
             device_cgroup_rules=device_cgroup_rules,
         )
 
         try:
-            container_ip = test_utils.get_container_ip(container_id)
-            test_utils.wait_for_ssh(container_ip, 22, "root", pkey_path)
-
-            pkey = paramiko.RSAKey.from_private_key_file(pkey_path)
-            ssh_conn_info = {
-                "ip": container_ip,
-                "port": 22,
-                "username": "root",
-                "pkey": coriolis_utils.serialize_key(pkey),
-            }
-
-            info = {
-                "container_id": container_id,
-                "ssh_connection_info": ssh_conn_info,
-            }
             if setup_writer:
+                ssh_conn_info = info["ssh_connection_info"]
                 if writer_backend == backup_writers.BACKUP_WRITER_SSH:
                     info["backup_writer_connection_info"] = {
                         "backend": backup_writers.BACKUP_WRITER_SSH,
@@ -291,7 +250,7 @@ class TestImportProvider(
 
             return info
         except Exception:
-            test_utils.remove_container(container_id)
+            test_utils.remove_container(info["container_id"])
             raise
 
     def delete_replica_target_resources(
@@ -466,26 +425,6 @@ class TestImportProvider(
             ctxt, connection_info, env, option_names
         )
 
-    def validate_minion_compatibility_for_transfer(
-        self, ctxt, connection_info, export_info, environment_options, minion_properties
-    ):
-        pass
-
-    def validate_minion_pool_environment_options(
-        self, ctxt, connection_info, environment_options
-    ):
-        pass
-
-    def set_up_pool_shared_resources(
-        self, ctxt, connection_info, environment_options, pool_identifier
-    ):
-        return {}
-
-    def tear_down_pool_shared_resources(
-        self, ctxt, connection_info, environment_options, pool_shared_resources
-    ):
-        pass
-
     def create_minion(
         self,
         ctxt,
@@ -521,74 +460,6 @@ class TestImportProvider(
             },
         }
 
-    def delete_minion(self, ctxt, connection_info, minion_properties):
-        container_id = (minion_properties or {}).get("container_id")
-        if container_id:
-            test_utils.remove_container(container_id)
-
-    def shutdown_minion(self, ctxt, connection_info, minion_properties):
-        container_id = (minion_properties or {}).get("container_id")
-        if container_id:
-            test_utils.stop_container(container_id)
-
-    def start_minion(self, ctxt, connection_info, minion_properties):
-        container_id = (minion_properties or {}).get("container_id")
-        if container_id:
-            test_utils.start_container(container_id)
-
-    def attach_volumes_to_minion(
-        self,
-        ctxt,
-        connection_info,
-        minion_properties,
-        minion_connection_info,
-        volumes_info,
-    ):
-        container_id = minion_properties["container_id"]
-        for vol in volumes_info:
-            device_path = vol["volume_dev"]
-            test_utils.hotplug_device_to_container(container_id, device_path)
-
-        return {
-            "minion_properties": minion_properties,
-            "volumes_info": volumes_info,
-        }
-
-    def detach_volumes_from_minion(
-        self,
-        ctxt,
-        connection_info,
-        minion_properties,
-        minion_connection_info,
-        volumes_info,
-    ):
-        container_id = (minion_properties or {}).get("container_id")
-        if not container_id:
-            return
-
-        for vol in volumes_info or []:
-            dev_path = vol.get("volume_dev")
-            if not dev_path:
-                continue
-
-            test_utils.unplug_device_from_container(container_id, dev_path)
-
-        return {
-            "minion_properties": minion_properties,
-            "volumes_info": volumes_info,
-        }
-
-    def healthcheck_minion(
-        self, ctxt, connection_info, minion_properties, minion_connection_info
-    ):
-        ip = minion_connection_info.get("ip")
-        port = minion_connection_info.get("port", 22)
-        username = minion_connection_info.get("username", "root")
-        pkey = minion_connection_info.get("pkey")
-
-        client = coriolis_utils.connect_ssh(ip, port, username, pkey=pkey)
-        client.close()
-
     def validate_osmorphing_minion_compatibility_for_transfer(
         self, ctxt, connection_info, export_info, environment_options, minion_properties
     ):

coriolis/tests/integration/utils.py

@@ -317,8 +317,20 @@ def _get_container_pid(container_id):
 
 
 def hotplug_device_to_container(container_id, device_path):
-    """Create a device node for *device_path* in *container_id*'s namespace."""
+    """Create a device node for *device_path* in *container_id*'s namespace.
+
+    Noop if the device node already exists in the container (e.g.: it was already
+    hotplugged by a previous call, such as an earlier incremental replication pass).
+    """
     pid = _get_container_pid(container_id)
+
+    exists = _run(
+        ["nsenter", "--target", str(pid), "--mount", "--", "test", "-e", device_path],
+        check=False,
+    )
+    if exists.returncode == 0:
+        return
+
     stat_result = os.stat(device_path)
     major = os.major(stat_result.st_rdev)
     minor = os.minor(stat_result.st_rdev)