Adds management-related integration tests

Tests providers.list() (verifies configured platforms are listed),
providers.schemas_list() for connection info, destination environment,
and source environment schemas.

Tests diagnostics.get() (verifies at least one diagnostics entry is returned
from the in-process services).

Tests region CRUD APIs.

Tests services CRUD APIs.

coriolis/tests/integration/base.py

@@ -48,6 +48,8 @@ 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_platform = cls._harness.exp_provider_platform
+        cls._imp_platform = cls._harness.imp_provider_platform
         cls._client = cls.get_client()
 
     # Helpers for subclasses
@@ -141,7 +143,7 @@ class ReplicaIntegrationTestBase(CoriolisIntegrationTestBase):
         # Create endpoints.
         self._src_endpoint = self._create_endpoint(
             name="test-src",
-            endpoint_type="test-src",
+            endpoint_type=self._exp_platform,
             description="integration source endpoint",
             connection_info={
                 "block_device_path": self._src_device,
@@ -151,7 +153,7 @@ class ReplicaIntegrationTestBase(CoriolisIntegrationTestBase):
 
         self._dst_endpoint = self._create_endpoint(
             name="test-dest",
-            endpoint_type="test-dest",
+            endpoint_type=self._imp_platform,
             description="integration destination endpoint",
             connection_info={
                 "devices": [self._dst_device],

coriolis/tests/integration/harness.py

@@ -18,6 +18,7 @@ Must be run as root (scsi_debug block device setup requires it).
 """
 
 import atexit
+import importlib
 import os
 import queue
 import shutil
@@ -74,6 +75,13 @@ _TEST_IMPORT_PROVIDER = (
 _TEST_PROJECT_ID = 'integration-project'
 
 
+def _provider_platform(dotted_path):
+    """Return the ``platform`` attribute of the class at *dotted_path*."""
+    module_path, class_name = dotted_path.rsplit('.', 1)
+    cls = getattr(importlib.import_module(module_path), class_name)
+    return cls.platform
+
+
 class DaemonCherootWorker(cheroot_threadpool.WorkerThread):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
@@ -260,6 +268,9 @@ class _IntegrationHarness:
         # Policy enforcer: reset so it re-reads the new CONF (no policy file).
         policy_module.reset()
 
+        self.exp_provider_platform = _provider_platform(_TEST_EXPORT_PROVIDER)
+        self.imp_provider_platform = _provider_platform(_TEST_IMPORT_PROVIDER)
+
         self._wsgi_server = None
         self._wsgi_server_thread = None
         self.api_port = None

coriolis/tests/integration/management/test_diagnostics.py

@@ -0,0 +1,35 @@
+# Copyright 2026 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+"""Integration tests for the diagnostics API.
+
+Exercises diagnostics.get() via the Coriolis REST API.
+"""
+
+import netifaces
+import socket
+
+from coriolis.tests.integration import base
+from coriolis import utils
+
+
+class DiagnosticsTest(base.CoriolisIntegrationTestBase):
+
+    def test_get_diagnostics(self):
+        diag_list = self._client.diagnostics.get()
+
+        # Returns a list of Diagnostics resources, one per service.
+        self.assertIsInstance(diag_list, list)
+        self.assertTrue(
+            len(diag_list) > 0, "Expected at least one diagnostics entry")
+
+        diag = diag_list[0]
+        diag_ip_addr = diag.ip_addresses[0]
+        ifname = list(diag_ip_addr.keys())[0]
+        ip = netifaces.ifaddresses(ifname)[netifaces.AF_INET][0]["addr"]
+
+        self.assertEqual(diag_ip_addr[ifname]["ipv4"][0]["addr"], ip)
+        self.assertEqual(diag.os_info, utils._get_host_os_info())
+        self.assertEqual(diag.hostname, socket.gethostname())
+
+        self.assertEqual(diag.to_dict(), utils.get_diagnostics_info())

coriolis/tests/integration/management/test_providers.py

@@ -0,0 +1,45 @@
+# Copyright 2026 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+"""Integration tests for the providers APIs.
+
+Exercises providers.list() and providers.schemas_list() via the Coriolis
+REST API.
+"""
+
+from coriolis import constants
+from coriolis.tests.integration import base
+
+
+class ProvidersTest(base.CoriolisIntegrationTestBase):
+
+    def test_list_providers(self):
+        providers = self._client.providers.list()
+
+        provider_names = [p["name"] for p in providers.providers_list]
+        self.assertIn(self._imp_platform, provider_names)
+        self.assertIn(self._exp_platform, provider_names)
+
+    def test_schemas_list(self):
+        # PROVIDER_TYPE_ENDPOINT returns a "connection_info_schema" key.
+        schemas = self._client.providers.schemas_list(
+            self._imp_platform, constants.PROVIDER_TYPE_ENDPOINT)
+
+        schema_types = [s["type"] for s in schemas.provider_schemas]
+        self.assertIn("connection_info_schema", schema_types)
+
+        # PROVIDER_TYPE_TRANSFER_IMPORT returns a
+        # "destination_environment_schema" key.
+        schemas = self._client.providers.schemas_list(
+            self._imp_platform, constants.PROVIDER_TYPE_TRANSFER_IMPORT)
+
+        schema_types = [s["type"] for s in schemas.provider_schemas]
+        self.assertIn("destination_environment_schema", schema_types)
+
+        # PROVIDER_TYPE_TRANSFER_EXPORT returns a "source_environment_schema"
+        # key.
+        schemas = self._client.providers.schemas_list(
+            self._exp_platform, constants.PROVIDER_TYPE_TRANSFER_EXPORT)
+
+        schema_types = [s["type"] for s in schemas.provider_schemas]
+        self.assertIn("source_environment_schema", schema_types)

coriolis/tests/integration/management/test_region.py

@@ -0,0 +1,46 @@
+# Copyright 2026 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+"""Integration tests for the regions APIs.
+
+Exercises region CRUD operations via the Coriolis REST API.
+"""
+
+from coriolis.tests.integration import base
+
+
+class RegionTests(base.CoriolisIntegrationTestBase):
+
+    def _create_region(self, name, **kwargs):
+        region = self._client.regions.create(name, **kwargs)
+        self.addCleanup(
+            self._ignoreExc(self._client.regions.delete), region.id)
+
+        return region
+
+    def test_region_crud(self):
+        # Create.
+        region = self._create_region(
+            "test-region", description="integration test region")
+
+        # Get.
+        fetched = self._client.regions.get(region.id)
+        self.assertEqual(region.id, fetched.id)
+        self.assertEqual("test-region", fetched.name)
+
+        # List.
+        regions = self._client.regions.list()
+        ids = [r.id for r in regions]
+        self.assertIn(region.id, ids)
+
+        # Update.
+        updated = self._client.regions.update(
+            region.id, {"description": "updated"})
+        self.assertEqual("updated", updated.description)
+
+        # Delete.
+        self._client.regions.delete(region.id)
+
+        regions = self._client.regions.list()
+        ids = [r.id for r in regions]
+        self.assertNotIn(region.id, ids)

coriolis/tests/integration/management/test_service.py

@@ -0,0 +1,57 @@
+# Copyright 2026 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+"""Integration tests for the services APIs.
+
+Exercises service CRUD operations via the Coriolis REST API.
+"""
+
+import socket
+
+from coriolis.tests.integration import base
+
+
+class ServiceTests(base.CoriolisIntegrationTestBase):
+
+    def _create_service(self, host, binary, topic):
+        svc = self._client.services.create(
+            host=host, binary=binary, topic=topic, regions=[])
+
+        self.addCleanup(self._ignoreExc(self._client.services.delete), svc.id)
+
+        return svc
+
+    def test_service_crud(self):
+        # The harness starts an in-process worker which registers itself with
+        # the conductor; at least one service record should exist.
+        services = self._client.services.list()
+
+        self.assertTrue(
+            len(services) > 0, "Expected at least one registered service")
+
+        # Create.
+        hostname = socket.gethostname()
+        svc = self._create_service(
+            hostname, "foo-binary", "coriolis_worker")
+
+        # Get.
+        fetched = self._client.services.get(svc.id)
+        self.assertEqual(svc.id, fetched.id)
+        self.assertEqual(hostname, fetched.host)
+        self.assertEqual("foo-binary", fetched.binary)
+
+        # List.
+        services = self._client.services.list()
+        ids = [s.id for s in services]
+        self.assertIn(svc.id, ids)
+
+        # Update.
+        updated = self._client.services.update(svc.id, {"enabled": False})
+        self.assertFalse(updated.enabled)
+
+        # Delete.
+        self._client.services.delete(svc.id)
+
+        services = self._client.services.list()
+        ids = [s.id for s in services]
+        self.assertNotIn(svc.id, ids)

coriolis/tests/integration/test_endpoints.py

@@ -23,7 +23,7 @@ class EndpointCapabilitiesTest(base.CoriolisIntegrationTestBase):
         # /dev/null satisfies os.path.exists checks in validate_connection.
         self._src_endpoint = self._create_endpoint(
             name="cap-src",
-            endpoint_type="test-src",
+            endpoint_type=self._exp_platform,
             connection_info={
                 "block_device_path": "/dev/null",
                 "pkey_path": self._harness.ssh_key_path,
@@ -32,7 +32,7 @@ class EndpointCapabilitiesTest(base.CoriolisIntegrationTestBase):
         # Empty devices list passes the destination validate_connection loop.
         self._dst_endpoint = self._create_endpoint(
             name="cap-dest",
-            endpoint_type="test-dest",
+            endpoint_type=self._imp_platform,
             connection_info={
                 "devices": [],
                 "pkey_path": self._harness.ssh_key_path,
@@ -51,7 +51,7 @@ class EndpointCapabilitiesTest(base.CoriolisIntegrationTestBase):
     def test_validate_connection_failure(self):
         bad_endpoint = self._create_endpoint(
             name="cap-bad",
-            endpoint_type="test-src",
+            endpoint_type=self._exp_platform,
             connection_info={
                 "block_device_path": "/dev/coriolis-no-such-device",
                 "pkey_path": self._harness.ssh_key_path,

coriolis/tests/integration/test_smoke.py

@@ -24,7 +24,7 @@ class HarnessSmokeTest(base.CoriolisIntegrationTestBase):
         """Endpoint create / get exercises."""
         endpoint = self._create_endpoint(
             name="smoke-test-endpoint",
-            endpoint_type="test-src",
+            endpoint_type=self._exp_platform,
             description="harness smoke test",
             connection_info={"key": "value"},
         )
@@ -38,13 +38,13 @@ class HarnessSmokeTest(base.CoriolisIntegrationTestBase):
 
         src = self._create_endpoint(
             name="smoke-src",
-            endpoint_type="test-src",
+            endpoint_type=self._exp_platform,
             connection_info={"foo": "lish"},
         )
 
         dst = self._create_endpoint(
             name="smoke-dst",
-            endpoint_type="test-dest",
+            endpoint_type=self._imp_platform,
             connection_info={"bar": "tender"},
         )