Fixes 503 error on getting minion pool options

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(

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)