Adds endpoints integration tests

Adds `BaseEndpointSourceOptionsProvider` and
`BaseUpdateSourceReplicaProvider`, `BaseEndpointInstancesProvider` to
`TestExportProvider`.

Adds `BaseEndpointDestinationOptionsProvider`,
`BaseEndpointNetworksProvider`, `BaseEndpointStorageProvider`,
`BaseInstanceProvider`, `BaseUpdateDestinationReplicaProvider` to `TestImportProvider`.

Adds integration tests for `validate_connection`, `endpoint_networks.list`,
`endpoint_storage.list`, `endpoint_source_options.list`,
``endpoint_destination_options.list`, `endpoint_instances.list`,
`endpoint_instances.get` via `coriolisclient`.

coriolis/tests/integration/providers/test_provider/exp.py

@@ -16,9 +16,11 @@ import paramiko
 
 from coriolis import events
 from coriolis.providers import backup_writers
-from coriolis.providers.base import BaseEndpointProvider
+from coriolis.providers.base import BaseEndpointInstancesProvider
+from coriolis.providers.base import BaseEndpointSourceOptionsProvider
 from coriolis.providers.base import BaseReplicaExportProvider
 from coriolis.providers.base import BaseReplicaExportValidationProvider
+from coriolis.providers.base import BaseUpdateSourceReplicaProvider
 from coriolis.providers import replicator as replicator_module
 
 CONF = cfg.CONF
@@ -26,7 +28,9 @@ LOG = logging.getLogger(__name__)
 
 
 class TestExportProvider(
-        BaseEndpointProvider,
+        BaseEndpointInstancesProvider,
+        BaseEndpointSourceOptionsProvider,
+        BaseUpdateSourceReplicaProvider,
         BaseReplicaExportProvider,
         BaseReplicaExportValidationProvider):
     """Source-side provider backed by a local `scsi_debug` block device.
@@ -83,6 +87,57 @@ class TestExportProvider(
     def get_source_environment_schema(self):
         return {"type": "object", "properties": {}}
 
+    # BaseEndpointInstancesProvider
+
+    def get_instances(self, ctxt, connection_info, source_environment,
+                      limit=None, last_seen_id=None,
+                      instance_name_pattern=None, refresh=False):
+        return [self._instance_info(connection_info)]
+
+    def get_instance(self, ctxt, connection_info, source_environment,
+                     instance_name):
+        return self._instance_info(connection_info)
+
+    def _instance_info(self, connection_info):
+        device = connection_info.get("device", "")
+        name = os.path.basename(device) if device else "test-instance"
+        return {
+            "id": name,
+            "name": name,
+            "instance_name": name,
+            "num_cpu": 1,
+            "memory_mb": 512,
+            "os_type": "linux",
+            "nested_virtualization": False,
+            "devices": {
+                "disks": [],
+                "nics": [],
+                "cdroms": [],
+                "serial_ports": [],
+                "floppies": [],
+                "controllers": [],
+            },
+        }
+
+    # BaseEndpointSourceOptionsProvider
+
+    def get_source_environment_options(
+            self, ctxt, connection_info, env=None, option_names=None):
+        return [
+            {
+                "name": "source_opt",
+                "values": ["foo", "lish"],
+                "config_default": "foo",
+            },
+        ]
+
+    # BaseUpdateSourceReplicaProvider
+
+    def check_update_source_environment_params(
+            self, ctxt, connection_info, instance_name, volumes_info,
+            old_params, new_params):
+        return volumes_info
+
     def get_os_morphing_tools(self, os_type, osmorphing_info):
         return []
 

coriolis/tests/integration/providers/test_provider/imp.py

@@ -15,9 +15,13 @@ from oslo_log import log as logging
 import paramiko
 
 from coriolis.providers import backup_writers
+from coriolis.providers.base import BaseEndpointDestinationOptionsProvider
+from coriolis.providers.base import BaseEndpointNetworksProvider
 from coriolis.providers.base import BaseEndpointProvider
+from coriolis.providers.base import BaseEndpointStorageProvider
 from coriolis.providers.base import BaseReplicaImportProvider
 from coriolis.providers.base import BaseReplicaImportValidationProvider
+from coriolis.providers.base import BaseUpdateDestinationReplicaProvider
 from coriolis import utils
 
 LOG = logging.getLogger(__name__)
@@ -29,6 +33,10 @@ WRITER_TEST_PORT = 16677
 
 class TestImportProvider(
         BaseEndpointProvider,
+        BaseEndpointDestinationOptionsProvider,
+        BaseEndpointNetworksProvider,
+        BaseEndpointStorageProvider,
+        BaseUpdateDestinationReplicaProvider,
         BaseReplicaImportProvider,
         BaseReplicaImportValidationProvider):
     """Destination-side provider backed by a local `scsi_debug` block device.
@@ -75,6 +83,37 @@ class TestImportProvider(
     def get_target_environment_schema(self):
         return {"type": "object", "properties": {}}
 
+    # BaseEndpointDestinationOptionsProvider
+
+    def get_target_environment_options(
+            self, ctxt, connection_info, env=None, option_names=None):
+        return [
+            {
+                "name": "dest_opt",
+                "values": ["foo", "lish"],
+                "config_default": "foo",
+            },
+        ]
+
+    # BaseEndpointNetworksProvider
+
+    def get_networks(self, ctxt, connection_info, env):
+        return [{"id": "test-net-1", "name": "test-net-1"}]
+
+    # BaseEndpointStorageProvider
+
+    def get_storage(self, ctxt, connection_info, target_environment):
+        return {
+            "storage_backends": [{"id": "test-store", "name": "test-store"}],
+        }
+
+    # BaseUpdateDestinationReplicaProvider
+
+    def check_update_destination_environment_params(
+            self, ctxt, connection_info, export_info, volumes_info,
+            old_params, new_params):
+        return volumes_info
+
     # BaseReplicaImportProvider
 
     def deploy_replica_disks(

coriolis/tests/integration/test_endpoints.py

@@ -0,0 +1,129 @@
+# Copyright 2026 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+"""Integration tests for endpoint capability APIs.
+
+Exercises endpoint-related operations via the Coriolis REST API:
+- validate_connection (success and failure)
+- endpoint update
+- get_networks
+- get_storage (list and default)
+- get_source_environment_options
+- get_target_environment_options
+- endpoint_instances.list and endpoint_instances.get
+"""
+
+from coriolis.tests.integration import base
+
+
+class EndpointCapabilitiesTest(base.CoriolisIntegrationTestBase):
+
+    def setUp(self):
+        super().setUp()
+        # /dev/null satisfies os.path.exists checks in validate_connection.
+        self._src_endpoint = self._create_endpoint(
+            name="cap-src",
+            endpoint_type="test-src",
+            connection_info={
+                "block_device_path": "/dev/null",
+                "pkey_path": base._TEST_SSH_KEY_PATH,
+            },
+        )
+        # Empty devices list passes the destination validate_connection loop.
+        self._dst_endpoint = self._create_endpoint(
+            name="cap-dest",
+            endpoint_type="test-dest",
+            connection_info={
+                "devices": [],
+                "pkey_path": base._TEST_SSH_KEY_PATH,
+            },
+        )
+
+    def test_validate_connection(self):
+        valid, message = self._client.endpoints.validate_connection(
+            self._src_endpoint.id)
+        self.assertTrue(valid, f"source: {message}")
+
+        valid, message = self._client.endpoints.validate_connection(
+            self._dst_endpoint.id)
+        self.assertTrue(valid, f"destination: {message}")
+
+    def test_validate_connection_failure(self):
+        bad_endpoint = self._create_endpoint(
+            name="cap-bad",
+            endpoint_type="test-src",
+            connection_info={
+                "block_device_path": "/dev/coriolis-no-such-device",
+                "pkey_path": base._TEST_SSH_KEY_PATH,
+            },
+        )
+        valid, message = self._client.endpoints.validate_connection(
+            bad_endpoint.id)
+        self.assertFalse(valid)
+        self.assertIsNotNone(message)
+
+    def test_endpoint_update(self):
+        updated = self._client.endpoints.update(
+            self._src_endpoint.id,
+            {"description": "updated description"},
+        )
+        self.assertEqual("updated description", updated.description)
+
+    def test_list_networks(self):
+        networks = self._client.endpoint_networks.list(self._dst_endpoint.id)
+        self.assertTrue(len(networks) > 0, "Expected at least one network")
+        first = networks[0]
+        self.assertIn("id", first._info)
+        self.assertIn("name", first._info)
+
+    def test_list_storage(self):
+        storage = self._client.endpoint_storage.list(self._dst_endpoint.id)
+        self.assertTrue(
+            len(storage) > 0, "Expected at least one storage backend")
+        first = storage[0]
+        self.assertIn("id", first._info)
+
+        # The test provider's get_storage() does not set a config_default so
+        # the value is expected to be None.
+        default = self._client.endpoint_storage.get_default(
+            self._dst_endpoint.id)
+        self.assertIsNone(default)
+
+    def test_list_source_options(self):
+        options = self._client.endpoint_source_options.list(
+            self._src_endpoint.id)
+        self.assertIsInstance(options, list)
+        self.assertTrue(
+            len(options) > 0, "Expected at least one source option")
+
+    def test_list_destination_options(self):
+        options = self._client.endpoint_destination_options.list(
+            self._dst_endpoint.id)
+        self.assertIsInstance(options, list)
+        self.assertTrue(
+            len(options) > 0, "Expected at least one destination option")
+
+    def test_list_instances(self):
+        instances = self._client.endpoint_instances.list(
+            self._src_endpoint.id, env={})
+
+        self.assertIsInstance(instances, list)
+        self.assertTrue(len(instances) > 0, "Expected at least one instance")
+        first = instances[0]
+        self.assertIn("id", first._info)
+        self.assertIn("name", first._info)
+
+        instances = self._client.endpoint_instances.list(
+            self._src_endpoint.id, env={}, name="null")
+        self.assertIsInstance(instances, list)
+
+    def test_get_instance(self):
+        instances = self._client.endpoint_instances.list(
+            self._src_endpoint.id, env={})
+
+        self.assertTrue(len(instances) > 0)
+
+        instance = self._client.endpoint_instances.get(
+            self._src_endpoint.id, instances[0].name, env={})
+
+        self.assertEqual(instances[0].name, instance.name)