From 5f578caa911436bed8a55b65f1d1ab8154aa9107 Mon Sep 17 00:00:00 2001
From: Claudiu Belu <cbelu@cloudbasesolutions.com>
Date: Fri, 14 Aug 2026 15:39:26 +0000
Subject: [PATCH] Fixes 503 error on getting minion pool options

If the endpoint has no minion pool support, the Coriolis API errors out
with 503 error. We now instead error out with 405.
---
 coriolis/minion_manager/rpc/server.py         | 20 ++++++++++
 .../tests/minion_manager/rpc/test_server.py   | 38 +++++++++++++++++++
 2 files changed, 58 insertions(+)

diff --git a/coriolis/minion_manager/rpc/server.py b/coriolis/minion_manager/rpc/server.py
index af7caf387..1d3ca12c6 100644
--- a/coriolis/minion_manager/rpc/server.py
+++ b/coriolis/minion_manager/rpc/server.py
@@ -218,9 +218,21 @@ def _rpc_minion_manager_client(self):
     def get_diagnostics(self, ctxt):
         return utils.get_diagnostics_info()
 
+    def _check_minion_pool_capability(
+            self, ctxt, platform_name, provider_type):
+        # Raises 405 error if the platform does not support minion pools.
+        provider_types = self._rpc_conductor_client.get_available_providers(
+            ctxt).get(platform_name, {}).get('types', [])
+        if provider_type not in provider_types:
+            raise exception.NotSupportedOperation(
+                operation="minion pools not supported for platform '%s'" %
+                platform_name)
+
     def get_endpoint_source_minion_pool_options(
             self, ctxt, endpoint_id, env, option_names):
         endpoint = self._rpc_conductor_client.get_endpoint(ctxt, endpoint_id)
+        self._check_minion_pool_capability(
+            ctxt, endpoint['type'], constants.PROVIDER_TYPE_SOURCE_MINION_POOL)
 
         worker_service = (
             self._rpc_scheduler_client.get_worker_service_for_specs(
@@ -240,6 +252,9 @@ def get_endpoint_source_minion_pool_options(
     def get_endpoint_destination_minion_pool_options(
             self, ctxt, endpoint_id, env, option_names):
         endpoint = self._rpc_conductor_client.get_endpoint(ctxt, endpoint_id)
+        self._check_minion_pool_capability(
+            ctxt, endpoint['type'],
+            constants.PROVIDER_TYPE_DESTINATION_MINION_POOL)
 
         worker_service = (
             self._rpc_scheduler_client.get_worker_service_for_specs(
@@ -258,6 +273,8 @@ def get_endpoint_destination_minion_pool_options(
     def validate_endpoint_source_minion_pool_options(
             self, ctxt, endpoint_id, pool_environment):
         endpoint = self._rpc_conductor_client.get_endpoint(ctxt, endpoint_id)
+        self._check_minion_pool_capability(
+            ctxt, endpoint['type'], constants.PROVIDER_TYPE_SOURCE_MINION_POOL)
 
         worker_service = (
             self._rpc_scheduler_client.get_worker_service_for_specs(
@@ -275,6 +292,9 @@ def validate_endpoint_source_minion_pool_options(
     def validate_endpoint_destination_minion_pool_options(
             self, ctxt, endpoint_id, pool_environment):
         endpoint = self._rpc_conductor_client.get_endpoint(ctxt, endpoint_id)
+        self._check_minion_pool_capability(
+            ctxt, endpoint['type'],
+            constants.PROVIDER_TYPE_DESTINATION_MINION_POOL)
 
         worker_service = (
             self._rpc_scheduler_client.get_worker_service_for_specs(
diff --git a/coriolis/tests/minion_manager/rpc/test_server.py b/coriolis/tests/minion_manager/rpc/test_server.py
index 7e5054516..9b3465c4e 100644
--- a/coriolis/tests/minion_manager/rpc/test_server.py
+++ b/coriolis/tests/minion_manager/rpc/test_server.py
@@ -333,3 +333,41 @@ def test_get_minion_pool_refresh_flow(
                         not in mock_set_minion_machine_allocation_status\
                         .mock_calls, f"Unexpected call to {method}, " \
                         f"args: {args}"
+
+    @ddt.data(
+        {
+            "available_types": [
+                constants.PROVIDER_TYPE_DESTINATION_MINION_POOL],
+            "expect_exception": False,
+        },
+        {
+            "available_types": [constants.PROVIDER_TYPE_OS_MORPHING],
+            "expect_exception": True,
+        },
+    )
+    @ddt.unpack
+    def test__check_minion_pool_capability(
+            self, available_types, expect_exception):
+        available_providers = {}
+        if available_types is not None:
+            available_providers = {"foo": {"types": available_types}}
+
+        mock_conductor_client = mock.MagicMock()
+        mock_conductor_client.get_available_providers.return_value = (
+            available_providers)
+        self.server._conductor_client_instance = mock_conductor_client
+
+        if expect_exception:
+            self.assertRaises(
+                exception.NotSupportedOperation,
+                self.server._check_minion_pool_capability,
+                mock.sentinel.context, "foo",
+                constants.PROVIDER_TYPE_DESTINATION_MINION_POOL)
+            return
+
+        self.server._check_minion_pool_capability(
+            mock.sentinel.context, "foo",
+            constants.PROVIDER_TYPE_DESTINATION_MINION_POOL)
+
+        mock_conductor_client.get_available_providers.assert_called_once_with(
+            mock.sentinel.context)