tests: Improve coverage for transfers

Adds verify_checksum source_environment flag through the test provider's
replicate_disks call (False by default). test_incremental_replica_transfer
will now also check the checksum.

Adds a "backup_writer_backend" destination-environment option through
_create_minion, so a test can request the SSH backend for one transfer.
Adds a transfer test using the mentioned SSH backup writer.

data_transfer.compression_proxy falls back to in-process gzip / zlib
compression when compressor_address is unset (None by default). That means
that the external compressor service scenario is not covered. Spins up a
unix-socket HTTP service in-process, points compressor_address at it for the
duration of one transfer, and asserts it was invoked.

.github/workflows/integration-tests.yml

@@ -36,6 +36,12 @@ jobs:
       run: |
         sudo apt-get install -y linux-modules-extra-$(uname -r)
 
+    - name: Build write_data resource binary
+      shell: bash
+      run: |
+        sudo apt-get install -y build-essential zlib1g-dev
+        make -C coriolis/resources
+
     - name: Build Docker image for integration test minions
       shell: bash
       run: |

coriolis/tests/integration/test_provider/exp.py

@@ -294,7 +294,10 @@ class TestExportProvider(
         backup_writer = backup_writers.BackupWritersFactory(
             target_conn_info, volumes_info).get_writer()
 
-        replicator.replicate_disks(source_volumes_info, backup_writer)
+        replicator.replicate_disks(
+            source_volumes_info, backup_writer,
+            verify_checksum=source_environment.get(
+                "verify_disk_integrity", False))
         return volumes_info
 
     def delete_replica_source_snapshots(

coriolis/tests/integration/test_provider/imp.py

@@ -66,6 +66,8 @@ class TestImportProvider(
     ``target_environment`` (per-transfer destination settings) has the form::
 
         {
+            # optional; "HTTPS" (default) or "SSH"
+            "data_transfer_mechanism": "HTTPS",
         }
     """
 
@@ -126,7 +128,9 @@ class TestImportProvider(
     def get_target_environment_schema(self):
         return {
             "type": "object",
-            "properties": {},
+            "properties": {
+                "data_transfer_mechanism": {"type": "string"},
+            },
             "required": [],
         }
 
@@ -181,8 +185,14 @@ class TestImportProvider(
     def deploy_replica_target_resources(
             self, ctxt, connection_info, target_environment, volumes_info):
         devices = [vol["volume_dev"] for vol in volumes_info]
+        data_transfer_mechanism = target_environment.get(
+            "data_transfer_mechanism",
+            backup_writers.DATA_TRANSFER_MECHANISM_HTTPS)
+        writer_backend = backup_writers.DATA_TRANSFER_MECHANISM_MAP[
+            data_transfer_mechanism]
         result = self._create_minion(
-            "coriolis-writer", connection_info, devices)
+            "coriolis-writer", connection_info, devices,
+            writer_backend=writer_backend)
 
         return {
             "volumes_info": volumes_info,
@@ -192,7 +202,8 @@ class TestImportProvider(
 
     def _create_minion(
             self, name_prefix, connection_info, devices=None, volumes=None,
-            device_cgroup_rules=None, setup_writer=True):
+            device_cgroup_rules=None, 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])
 
@@ -223,13 +234,20 @@ class TestImportProvider(
                 "ssh_connection_info": ssh_conn_info,
             }
             if setup_writer:
-                bootstrapper = backup_writers.HTTPBackupWriterBootstrapper(
-                    ssh_conn_info, WRITER_TEST_PORT)
-                writer_conn_details = bootstrapper.setup_writer()
-                info["backup_writer_connection_info"] = {
-                    "backend": "http_backup_writer",
-                    "connection_details": writer_conn_details,
-                }
+                if writer_backend == backup_writers.BACKUP_WRITER_SSH:
+                    info["backup_writer_connection_info"] = {
+                        "backend": backup_writers.BACKUP_WRITER_SSH,
+                        "connection_details": ssh_conn_info,
+                    }
+                else:
+                    bootstrapper = (
+                        backup_writers.HTTPBackupWriterBootstrapper(
+                            ssh_conn_info, WRITER_TEST_PORT))
+                    writer_conn_details = bootstrapper.setup_writer()
+                    info["backup_writer_connection_info"] = {
+                        "backend": backup_writers.BACKUP_WRITER_HTTP,
+                        "connection_details": writer_conn_details,
+                    }
 
             return info
         except Exception:

coriolis/tests/integration/transfers/test_transfer.py

@@ -7,12 +7,32 @@ Integration tests for the replica transfer pipeline.
 Must be run as root.
 """
 
+import gzip
+import http.server
+import os
+import shutil
+import socketserver
+import tempfile
+import threading
+from unittest import mock
+import zlib
+
+from oslo_config import cfg
+
+from coriolis import data_transfer
+from coriolis.providers import backup_writers
 from coriolis.tests.integration import base
 from coriolis.tests.integration import utils as test_utils
 
+CONF = cfg.CONF
 
-class ReplicaTransferIntegrationTest(base.ReplicaIntegrationTestBase):
-    """Full-pipeline replica transfer integration tests."""
+_COMPRESS_FUNC = {
+    "gzip": gzip.compress,
+    "zlib": zlib.compress,
+}
+
+
+class _ReplicaTransferTestsMixin:
 
     def test_transfer(self):
         # List the transfer
@@ -42,7 +62,9 @@ class ReplicaTransferIntegrationTest(base.ReplicaIntegrationTestBase):
           Coriolis REST API (using coriolisclient).
         - Execute the transfer and wait for it to complete.
         - Overwrite a single chunk on the source device.
+        - Update the transfer to enable disk integrity verification.
         - Execute a second transfer run (incremental=True).
+        - Assert that the destination disk checksum was computed.
 
         The content is verified only if the test import provider is being used.
         """
@@ -67,8 +89,37 @@ class ReplicaTransferIntegrationTest(base.ReplicaIntegrationTestBase):
                 "Devices should differ after mutating the source",
             )
 
+        # Enable disk integrity verification for the incremental run.
+        # Note that the HTTP backup writer is the only one implementing
+        # get_disk_checksum().
+        execution = self._client.transfers.update(
+            self._transfer.id,
+            {
+                "source_environment": {"verify_disk_integrity": True},
+                "destination_environment": {
+                    "data_transfer_mechanism": (
+                        backup_writers.DATA_TRANSFER_MECHANISM_HTTPS),
+                },
+            })
+        self.assertExecutionCompleted(execution.id)
+
+        checksummed_disks = []
+        original = backup_writers.HTTPBackupWriterImpl.get_disk_checksum
+
+        def _recording_get_disk_checksum(writer, *args, **kwargs):
+            checksummed_disks.append(writer._disk_id)
+            return original(writer, *args, **kwargs)
+
         # Second run: incremental
-        self._execute_and_wait(self._transfer.id)
+        with mock.patch.object(
+                backup_writers.HTTPBackupWriterImpl, "get_disk_checksum",
+                _recording_get_disk_checksum):
+            self._execute_and_wait(self._transfer.id)
+
+        self.assertEqual(
+            [os.path.basename(self._src_device)],
+            checksummed_disks,
+            "The checksum was not computed for the transferred disk")
 
         if self._harness.uses_core_test_import_provider():
             self.assertTrue(
@@ -77,8 +128,93 @@ class ReplicaTransferIntegrationTest(base.ReplicaIntegrationTestBase):
             )
 
 
+class ReplicaTransferIntegrationTest(
+    base.ReplicaIntegrationTestBase, _ReplicaTransferTestsMixin):
+    """Full-pipeline replica transfer integration tests."""
+
+    def test_transfer_with_ssh_backup_writer(self):
+        # NOTE: for minion pools, updating the data_transfer_mechanism will not
+        # set up the new transfer mechanism into existing minions.
+        execution = self._client.transfers.update(
+            self._transfer.id,
+            {
+                "destination_environment": {
+                    "data_transfer_mechanism": (
+                        backup_writers.DATA_TRANSFER_MECHANISM_SSH),
+                },
+            })
+        self.assertExecutionCompleted(execution.id)
+
+        # Record the writers handed out during the run.
+        writer_types = set()
+        original = backup_writers.BackupWritersFactory.get_writer
+
+        def _recording_get_writer(factory, *args, **kwargs):
+            writer = original(factory, *args, **kwargs)
+            writer_types.add(type(writer))
+            return writer
+
+        with mock.patch.object(
+                backup_writers.BackupWritersFactory, "get_writer",
+                _recording_get_writer):
+            self._execute_and_wait(self._transfer.id)
+
+        self.assertEqual(
+            {backup_writers.SSHBackupWriter},
+            writer_types,
+            "The transfer did not use the expected SSH backup writer")
+
+    def test_transfer_with_external_compressor(self):
+        # Compression is typically done in-process. For this test, point
+        # compressor_address at a unix-socket HTTP service.
+        socket_dir = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, socket_dir, ignore_errors=True)
+        socket_path = os.path.join(socket_dir, "compressor.sock")
+
+        call_count = 0
+
+        class _CompressorHandler(http.server.BaseHTTPRequestHandler):
+            def do_POST(self):
+                nonlocal call_count
+                call_count += 1
+
+                length = int(self.headers["Content-Length"])
+                body = self.rfile.read(length)
+                fmt = self.headers["X-Compression-Format"]
+                compressed = _COMPRESS_FUNC[fmt](body)
+
+                self.send_response(200)
+                self.send_header("Content-Length", str(len(compressed)))
+                self.end_headers()
+                self.wfile.write(compressed)
+
+            def address_string(self):
+                # Default impl indexes client_address, which is empty '' for
+                # a unix socket, causing an IndexError while logging.
+                return self.client_address
+
+        server = socketserver.UnixStreamServer(
+            socket_path, _CompressorHandler)
+        server_thread = threading.Thread(
+            target=server.serve_forever, daemon=True)
+        server_thread.start()
+        self.addCleanup(server_thread.join)
+        self.addCleanup(server.shutdown)
+
+        CONF.set_override("compressor_address", socket_path)
+        self.addCleanup(CONF.clear_override, "compressor_address")
+
+        with mock.patch.object(data_transfer.LOG, "exception") as mock_exc:
+            self._execute_and_wait(self._transfer.id)
+
+        self.assertGreater(
+            call_count, 0,
+            "External compressor service was never invoked")
+        mock_exc.assert_not_called()
+
+
 class MinionPoolTransferTest(
-        base.MinionPoolReplicaTestBase, ReplicaTransferIntegrationTest):
+        base.MinionPoolReplicaTestBase, _ReplicaTransferTestsMixin):
     """Transfer execution that uses a pre-allocated destination minion pool."""
 
     def test_transfer(self):