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

@@ -218,9 +218,21 @@ class MinionManagerServerEndpoint(object):
     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 @@ class MinionManagerServerEndpoint(object):
     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 @@ class MinionManagerServerEndpoint(object):
     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 @@ class MinionManagerServerEndpoint(object):
     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(

coriolis/tests/minion_manager/rpc/test_server.py

@@ -333,3 +333,41 @@ class MinionManagerServerEndpointTestCase(test_base.CoriolisBaseTestCase):
                         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)