Allow disabling quota management

Adjutant enables management of quotas for all supported services
by default. In some cases this may not be desired
(e.g. some regions don't have a full suite of services available).

This change makes it possible to disable quota management in Adjutant,
either per-region or completely.

To disable quota management for one region, while keeping it
enabled in all others:

quota:
  services:
    RegionOne: []
    '*':
      - cinder
      - neutron
      - nova
      - octavia
      - trove

To disable quota management using Adjutant completely,
set the following:

quota:
  services: {}

If new quota update requests are made including a region for which
quota management is disabled, one of the following actions
will be taken:

* If the request also specifies other regions for which
  quota management is enabled, the request is accepted but with
  the disabled regions removed from the request.
* If the request **only** specifies regions for which
  quota management is disabled, `400 Bad Request` will be returned
  with an appropriate error response.

If quota management for a region is disabled while there are existing
pending quota update tasks:

* If the task also updates the quota for other regions,
  the task can be approved as normal. The task will simply ignore the
  disabled region, adding a note to the task log notifying the admin
  that it has done so.
* If the task **only** updates quotas for disabled regions, the task
  will no longer be valid, and will return an error when an admin
  tries to approve them. Such tasks should be cancelled instead.

Change-Id: I1cfe6d1e3c12595966769334bea4c14450124f13

adjutant/actions/v1/resources.py

@@ -313,6 +313,18 @@ class UpdateProjectQuotasAction(BaseAction, QuotaMixin):
             return False
         return True
 
+    def _validate_quota_management_enabled_for_regions(self):
+        # Check that at least one region in the given list has
+        # quota management enabled.
+        default_services = CONF.quota.services.get("*", {})
+        for region in self.regions:
+            if CONF.quota.services.get(region, default_services):
+                return True
+        self.add_note(
+            "Quota management is disabled for all specified regions",
+        )
+        return False
+
     def _set_region_quota(self, region_name, quota_size):
         # Set the quota for an individual region
         quota_config = CONF.quota.sizes.get(quota_size, {})
@@ -360,9 +372,17 @@ class UpdateProjectQuotasAction(BaseAction, QuotaMixin):
         )
 
         for region in self.regions:
-            current_size = quota_manager.get_region_quota_data(
+            current_quota = quota_manager.get_region_quota_data(
                 region, include_usage=False
-            )["current_quota_size"]
+            )
+            # If get_region_quota_data returns None, this region
+            # has quota management disabled.
+            if not current_quota:
+                self.add_note(
+                    f"Quota management is disabled in region: {region}",
+                )
+                continue
+            current_size = current_quota["current_quota_size"]
             region_sizes.append(current_size)
             self.add_note(
                 "Project has size '%s' in region: '%s'" % (current_size, region)
@@ -372,6 +392,12 @@ class UpdateProjectQuotasAction(BaseAction, QuotaMixin):
         preapproved_quotas = []
         smaller_quotas = []
 
+        if not region_sizes:
+            self.add_note(
+                "Quota management is disabled for all specified regions",
+            )
+            return False
+
         # If all region sizes are the same
         if region_sizes.count(region_sizes[0]) == len(region_sizes):
             preapproved_quotas = quota_manager.get_quota_change_options(region_sizes[0])
@@ -409,6 +435,7 @@ class UpdateProjectQuotasAction(BaseAction, QuotaMixin):
                 self._validate_project_id,
                 self._validate_quota_size_exists,
                 self._validate_regions_exist,
+                self._validate_quota_management_enabled_for_regions,
                 self._validate_usage_lower_than_quota,
             ]
         )

adjutant/actions/v1/tests/test_resource_actions.py

@@ -568,6 +568,122 @@ class QuotaActionTests(AdjutantTestCase):
         neutronquota = neutron_cache["RegionOne"]["test_project_id"]["quota"]
         self.assertEqual(neutronquota["network"], 5)
 
+    @conf_utils.modify_conf(
+        CONF,
+        operations={
+            "adjutant.quota.services": [
+                {
+                    "operation": "override",
+                    "value": {
+                        "RegionOne": [],
+                        "*": ["cinder", "neutron", "nova"],
+                    },
+                },
+            ],
+        },
+    )
+    def test_update_quota_fail_disabled_region(self):
+        """
+        Check that a quota update for a region for which quota management
+        is disabled is not valid, or performed.
+        """
+        project = mock.Mock()
+        project.id = "test_project_id"
+        project.name = "test_project"
+        project.domain = "default"
+        project.roles = {}
+
+        user = mock.Mock()
+        user.id = "user_id"
+        user.name = "test@example.com"
+        user.email = "test@example.com"
+        user.domain = "default"
+        user.password = "test_password"
+
+        setup_identity_cache(projects=[project], users=[user])
+        setup_mock_caches("RegionOne", "test_project_id")
+
+        # Test sending to only a single region
+        task = Task.objects.create(keystone_user={"roles": ["admin"]})
+
+        data = {
+            "project_id": "test_project_id",
+            "size": "large",
+            "regions": ["RegionOne"],
+            "user_id": user.id,
+        }
+
+        action = UpdateProjectQuotasAction(data, task=task, order=1)
+
+        action.prepare()
+        self.assertEqual(action.valid, False)
+
+        action.approve()
+        self.assertEqual(action.valid, False)
+
+        # check the quotas were updated
+        cinderquota = cinder_cache["RegionOne"]["test_project_id"]["quota"]
+        self.assertEqual(cinderquota["gigabytes"], 5000)
+        novaquota = nova_cache["RegionOne"]["test_project_id"]["quota"]
+        self.assertEqual(novaquota["ram"], 65536)
+        neutronquota = neutron_cache["RegionOne"]["test_project_id"]["quota"]
+        self.assertEqual(neutronquota["network"], 3)
+
+    @conf_utils.modify_conf(
+        CONF,
+        operations={
+            "adjutant.quota.services": [
+                {"operation": "override", "value": {}},
+            ],
+        },
+    )
+    def test_update_quota_fail_disabled(self):
+        """
+        Check that a quota update tasks are not valid or performed
+        when quota management is disabled completely.
+        """
+        project = mock.Mock()
+        project.id = "test_project_id"
+        project.name = "test_project"
+        project.domain = "default"
+        project.roles = {}
+
+        user = mock.Mock()
+        user.id = "user_id"
+        user.name = "test@example.com"
+        user.email = "test@example.com"
+        user.domain = "default"
+        user.password = "test_password"
+
+        setup_identity_cache(projects=[project], users=[user])
+        setup_mock_caches("RegionOne", "test_project_id")
+
+        # Test sending to only a single region
+        task = Task.objects.create(keystone_user={"roles": ["admin"]})
+
+        data = {
+            "project_id": "test_project_id",
+            "size": "large",
+            "regions": ["RegionOne"],
+            "user_id": user.id,
+        }
+
+        action = UpdateProjectQuotasAction(data, task=task, order=1)
+
+        action.prepare()
+        self.assertEqual(action.valid, False)
+
+        action.approve()
+        self.assertEqual(action.valid, False)
+
+        # check the quotas were updated
+        cinderquota = cinder_cache["RegionOne"]["test_project_id"]["quota"]
+        self.assertEqual(cinderquota["gigabytes"], 5000)
+        novaquota = nova_cache["RegionOne"]["test_project_id"]["quota"]
+        self.assertEqual(novaquota["ram"], 65536)
+        neutronquota = neutron_cache["RegionOne"]["test_project_id"]["quota"]
+        self.assertEqual(neutronquota["network"], 3)
+
     def test_update_quota_multi_region(self):
         """
         Sets a new quota on all services of a project in multiple regions
@@ -622,6 +738,140 @@ class QuotaActionTests(AdjutantTestCase):
         neutronquota = neutron_cache["RegionTwo"]["test_project_id"]["quota"]
         self.assertEqual(neutronquota["network"], 10)
 
+    @conf_utils.modify_conf(
+        CONF,
+        operations={
+            "adjutant.quota.services": [
+                {
+                    "operation": "override",
+                    "value": {
+                        "RegionTwo": [],
+                        "*": ["cinder", "neutron", "nova"],
+                    },
+                },
+            ],
+        },
+    )
+    def test_update_quota_multi_region_one_disabled(self):
+        """
+        Check that when a request to update multiple regions at once
+        and one of the regions have quota management disabled,
+        only the enabled regions have their quotas updated.
+        """
+        project = mock.Mock()
+        project.id = "test_project_id"
+        project.name = "test_project"
+        project.domain = "default"
+        project.roles = {}
+
+        user = mock.Mock()
+        user.id = "user_id"
+        user.name = "test@example.com"
+        user.email = "test@example.com"
+        user.domain = "default"
+        user.password = "test_password"
+
+        setup_identity_cache(projects=[project], users=[user])
+        setup_mock_caches("RegionOne", project.id)
+        setup_mock_caches("RegionTwo", project.id)
+
+        task = Task.objects.create(keystone_user={"roles": ["admin"]})
+
+        data = {
+            "project_id": "test_project_id",
+            "size": "large",
+            "domain_id": "default",
+            "regions": ["RegionOne", "RegionTwo"],
+            "user_id": "user_id",
+        }
+
+        action = UpdateProjectQuotasAction(data, task=task, order=1)
+
+        action.prepare()
+        self.assertEqual(action.valid, True)
+
+        action.approve()
+        self.assertEqual(action.valid, True)
+
+        # check the quotas were updated
+        cinderquota = cinder_cache["RegionOne"]["test_project_id"]["quota"]
+        self.assertEqual(cinderquota["gigabytes"], 50000)
+        novaquota = nova_cache["RegionOne"]["test_project_id"]["quota"]
+        self.assertEqual(novaquota["ram"], 655360)
+        neutronquota = neutron_cache["RegionOne"]["test_project_id"]["quota"]
+        self.assertEqual(neutronquota["network"], 10)
+
+        cinderquota = cinder_cache["RegionTwo"]["test_project_id"]["quota"]
+        self.assertEqual(cinderquota["gigabytes"], 5000)
+        novaquota = nova_cache["RegionTwo"]["test_project_id"]["quota"]
+        self.assertEqual(novaquota["ram"], 65536)
+        neutronquota = neutron_cache["RegionTwo"]["test_project_id"]["quota"]
+        self.assertEqual(neutronquota["network"], 3)
+
+    @conf_utils.modify_conf(
+        CONF,
+        operations={
+            "adjutant.quota.services": [
+                {"operation": "override", "value": {}},
+            ],
+        },
+    )
+    def test_update_quota_multi_region_disabled(self):
+        """
+        Check that if a task to update quotas for multiple regions at once
+        is initiated but quota management is disabled, no regions' quotas
+        are updated.
+        """
+        project = mock.Mock()
+        project.id = "test_project_id"
+        project.name = "test_project"
+        project.domain = "default"
+        project.roles = {}
+
+        user = mock.Mock()
+        user.id = "user_id"
+        user.name = "test@example.com"
+        user.email = "test@example.com"
+        user.domain = "default"
+        user.password = "test_password"
+
+        setup_identity_cache(projects=[project], users=[user])
+        setup_mock_caches("RegionOne", project.id)
+        setup_mock_caches("RegionTwo", project.id)
+
+        task = Task.objects.create(keystone_user={"roles": ["admin"]})
+
+        data = {
+            "project_id": "test_project_id",
+            "size": "large",
+            "domain_id": "default",
+            "regions": ["RegionOne", "RegionTwo"],
+            "user_id": "user_id",
+        }
+
+        action = UpdateProjectQuotasAction(data, task=task, order=1)
+
+        action.prepare()
+        self.assertEqual(action.valid, False)
+
+        action.approve()
+        self.assertEqual(action.valid, False)
+
+        # check the quotas were updated
+        cinderquota = cinder_cache["RegionOne"]["test_project_id"]["quota"]
+        self.assertEqual(cinderquota["gigabytes"], 5000)
+        novaquota = nova_cache["RegionOne"]["test_project_id"]["quota"]
+        self.assertEqual(novaquota["ram"], 65536)
+        neutronquota = neutron_cache["RegionOne"]["test_project_id"]["quota"]
+        self.assertEqual(neutronquota["network"], 3)
+
+        cinderquota = cinder_cache["RegionTwo"]["test_project_id"]["quota"]
+        self.assertEqual(cinderquota["gigabytes"], 5000)
+        novaquota = nova_cache["RegionTwo"]["test_project_id"]["quota"]
+        self.assertEqual(novaquota["ram"], 65536)
+        neutronquota = neutron_cache["RegionTwo"]["test_project_id"]["quota"]
+        self.assertEqual(neutronquota["network"], 3)
+
     @conf_utils.modify_conf(
         CONF,
         operations={

adjutant/api/v1/openstack.py

@@ -467,9 +467,9 @@ class UpdateProjectQuotas(BaseDelegateAPI):
         quota_manager = QuotaManager(self.project_id)
         for region in regions:
             if self.check_region_exists(region):
-                region_quotas.append(
+                quota = quota_manager.get_region_quota_data(region, include_usage)
-                    quota_manager.get_region_quota_data(region, include_usage)
+                if quota:
-                )
+                    region_quotas.append(quota)
             else:
                 return Response({"ERROR": ["Region: %s is not valid" % region]}, 400)
 
@@ -489,15 +489,67 @@ class UpdateProjectQuotas(BaseDelegateAPI):
         request.data["project_id"] = request.keystone_user["project_id"]
         self.project_id = request.keystone_user["project_id"]
 
-        regions = request.data.get("regions", None)
+        # Fetch the currently existing regions.
+        active_regions = {
+            region.id: region for region in user_store.IdentityManager().list_regions()
+        }
 
+        # Get the regions for which quota updates should be applied,
+        # parsing the list to get the unique region names while
+        # preserving list order.
+        # If no regions are specified in the request, update all regions.
+        _target_regions = request.data.get("regions", [])
+        target_regions = list(
+            (
+                dict.fromkeys(_target_regions) if _target_regions else active_regions
+            ).keys()
+        )
+
+        # Create a mapping to check whether or not quota management
+        # is enabled on a per-region basis.
+        quota_enabled_for_region = {
+            region: bool(services) for region, services in CONF.quota.services.items()
+        }
+
+        # Filter out regions for which quota management is disabled.
+        # This checks for region-specific settings first, and if
+        # there isn't one, checks the settings for '*' (all regions).
+        regions = [
+            region
+            for region in target_regions
+            if quota_enabled_for_region.get(
+                region,
+                quota_enabled_for_region.get("*", False),
+            )
+        ]
+        request.data["regions"] = regions
+
+        # Check that any of the specified regions can
+        # have their quota updated.
         if not regions:
-            id_manager = user_store.IdentityManager()
+            return Response(
-            regions = [region.id for region in id_manager.list_regions()]
+                {
-            request.data["regions"] = regions
+                    "errors": [
+                        "Unable to update the quotas for the given regions",
+                    ],
+                },
+                status=400,
+            )
 
-        self.logger.info("(%s) - New UpdateProjectQuotas request." % timezone.now())
+        # Check that the specified regions actually exist.
+        not_regions = []
+        for region in regions:
+            if region not in active_regions:
+                not_regions.append(region)
+        if not_regions:
+            return Response(
+                {"errors": [f"Invalid regions: {', '.join(not_regions)}"]},
+                status=400,
+            )
 
+        self.logger.info(
+            "(%s) - New UpdateProjectQuotas request.",
+            timezone.now(),
+        )
         self.task_manager.create_from_request(self.task_type, request)
-
         return Response({"notes": ["task created"]}, status=202)

adjutant/api/v1/tests/test_api_openstack.py

@@ -455,6 +455,334 @@ class QuotaAPITests(AdjutantAPITestCase):
             instance = CONF.quota.sizes.get(size)["trove"]["instances"]
             self.assertEqual(trove_quota["instances"], instance)
 
+    def test_quota_show(self):
+        """Check fetching the current quota sizes for available regions."""
+
+        project = fake_clients.FakeProject(
+            name="test_project",
+            id="test_project_id",
+        )
+
+        user = fake_clients.FakeUser(
+            name="test@example.com", password="123", email="test@example.com"
+        )
+
+        setup_identity_cache(projects=[project], users=[user])
+
+        admin_headers = {
+            "project_name": "test_project",
+            "project_id": project.id,
+            "roles": "project_admin,member,project_mod",
+            "username": "test@example.com",
+            "user_id": "user_id",
+            "authenticated": True,
+        }
+
+        url = "/v1/openstack/quotas/"
+
+        data = {}
+
+        response = self.client.get(
+            url,
+            data,
+            headers=admin_headers,
+            format="json",
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(
+            {
+                res["region"]: res["current_quota_size"]
+                for res in response.data["regions"]
+            },
+            {"RegionOne": "small", "RegionTwo": "small"},
+        )
+
+    def test_quota_show_explicit_single(self):
+        """Check the quota size for a single region explicitly."""
+
+        project = fake_clients.FakeProject(
+            name="test_project",
+            id="test_project_id",
+        )
+
+        user = fake_clients.FakeUser(
+            name="test@example.com", password="123", email="test@example.com"
+        )
+
+        setup_identity_cache(projects=[project], users=[user])
+
+        admin_headers = {
+            "project_name": "test_project",
+            "project_id": project.id,
+            "roles": "project_admin,member,project_mod",
+            "username": "test@example.com",
+            "user_id": "user_id",
+            "authenticated": True,
+        }
+
+        url = "/v1/openstack/quotas/"
+
+        data = {"regions": "RegionOne"}
+
+        response = self.client.get(
+            url,
+            data,
+            headers=admin_headers,
+            format="json",
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(
+            {
+                res["region"]: res["current_quota_size"]
+                for res in response.data["regions"]
+            },
+            {"RegionOne": "small"},
+        )
+
+    def test_quota_show_explicit_multiple(self):
+        """Check the quota size for multiple regions explicitly."""
+
+        project = fake_clients.FakeProject(
+            name="test_project",
+            id="test_project_id",
+        )
+
+        user = fake_clients.FakeUser(
+            name="test@example.com", password="123", email="test@example.com"
+        )
+
+        setup_identity_cache(projects=[project], users=[user])
+
+        admin_headers = {
+            "project_name": "test_project",
+            "project_id": project.id,
+            "roles": "project_admin,member,project_mod",
+            "username": "test@example.com",
+            "user_id": "user_id",
+            "authenticated": True,
+        }
+
+        url = "/v1/openstack/quotas/"
+
+        data = {"regions": "RegionOne,RegionTwo"}
+
+        response = self.client.get(
+            url,
+            data,
+            headers=admin_headers,
+            format="json",
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(
+            {
+                res["region"]: res["current_quota_size"]
+                for res in response.data["regions"]
+            },
+            {"RegionOne": "small", "RegionTwo": "small"},
+        )
+
+    @conf_utils.modify_conf(
+        CONF,
+        operations={
+            "adjutant.quota.services": [
+                {
+                    "operation": "override",
+                    "value": {
+                        "RegionTwo": [],
+                        "*": ["cinder", "neutron", "nova"],
+                    },
+                },
+            ],
+        },
+    )
+    def test_quota_show_region_disabled(self):
+        """Check that if a request for showing the quota size of a region
+        for which quota management is disabled, an OK response is returned
+        with no regions listed.
+        """
+
+        project = fake_clients.FakeProject(
+            name="test_project",
+            id="test_project_id",
+        )
+
+        user = fake_clients.FakeUser(
+            name="test@example.com", password="123", email="test@example.com"
+        )
+
+        setup_identity_cache(projects=[project], users=[user])
+
+        admin_headers = {
+            "project_name": "test_project",
+            "project_id": project.id,
+            "roles": "project_admin,member,project_mod",
+            "username": "test@example.com",
+            "user_id": "user_id",
+            "authenticated": True,
+        }
+
+        url = "/v1/openstack/quotas/"
+
+        data = {"regions": "RegionTwo"}
+
+        response = self.client.get(
+            url,
+            data,
+            headers=admin_headers,
+            format="json",
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.data["regions"], [])
+
+    @conf_utils.modify_conf(
+        CONF,
+        operations={
+            "adjutant.quota.services": [
+                {
+                    "operation": "override",
+                    "value": {
+                        "RegionTwo": [],
+                        "*": ["cinder", "neutron", "nova"],
+                    },
+                },
+            ],
+        },
+    )
+    def test_quota_show_one_region_disabled(self):
+        """Check that if a request for showing quota sizes for multiple
+        regions are made, and one of those regions have quota management
+        disabled, that only quotas for enabled regions are returned.
+        """
+
+        project = fake_clients.FakeProject(
+            name="test_project",
+            id="test_project_id",
+        )
+
+        user = fake_clients.FakeUser(
+            name="test@example.com", password="123", email="test@example.com"
+        )
+
+        setup_identity_cache(projects=[project], users=[user])
+
+        admin_headers = {
+            "project_name": "test_project",
+            "project_id": project.id,
+            "roles": "project_admin,member,project_mod",
+            "username": "test@example.com",
+            "user_id": "user_id",
+            "authenticated": True,
+        }
+
+        url = "/v1/openstack/quotas/"
+
+        data = {"regions": "RegionOne,RegionTwo"}
+
+        response = self.client.get(
+            url,
+            data,
+            headers=admin_headers,
+            format="json",
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(
+            {
+                res["region"]: res["current_quota_size"]
+                for res in response.data["regions"]
+            },
+            {"RegionOne": "small"},
+        )
+
+    @conf_utils.modify_conf(
+        CONF,
+        operations={
+            "adjutant.quota.services": [
+                {"operation": "override", "value": {}},
+            ],
+        },
+    )
+    def test_quota_show_disabled(self):
+        """Check that no quota sizes for regions are returned if
+        quota management is disabled entirely.
+        """
+
+        project = fake_clients.FakeProject(
+            name="test_project",
+            id="test_project_id",
+        )
+
+        user = fake_clients.FakeUser(
+            name="test@example.com", password="123", email="test@example.com"
+        )
+
+        setup_identity_cache(projects=[project], users=[user])
+
+        admin_headers = {
+            "project_name": "test_project",
+            "project_id": project.id,
+            "roles": "project_admin,member,project_mod",
+            "username": "test@example.com",
+            "user_id": "user_id",
+            "authenticated": True,
+        }
+
+        url = "/v1/openstack/quotas/"
+
+        data = {}
+
+        response = self.client.get(
+            url,
+            data,
+            headers=admin_headers,
+            format="json",
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.data["regions"], [])
+
+    def test_quota_show_invalid_region(self):
+        """Check that if a request for showing the quota size of an
+        invalid region is made, an error response is returned.
+        """
+
+        project = fake_clients.FakeProject(
+            name="test_project",
+            id="test_project_id",
+        )
+
+        user = fake_clients.FakeUser(
+            name="test@example.com", password="123", email="test@example.com"
+        )
+
+        setup_identity_cache(projects=[project], users=[user])
+
+        admin_headers = {
+            "project_name": "test_project",
+            "project_id": project.id,
+            "roles": "project_admin,member,project_mod",
+            "username": "test@example.com",
+            "user_id": "user_id",
+            "authenticated": True,
+        }
+
+        url = "/v1/openstack/quotas/"
+
+        data = {"regions": "RegionThree"}
+
+        response = self.client.get(
+            url,
+            data,
+            headers=admin_headers,
+            format="json",
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
     def test_update_quota_no_history(self):
         """Update the quota size of a project with no history"""
 
@@ -780,6 +1108,165 @@ class QuotaAPITests(AdjutantAPITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertEqual(response.data["regions"][0]["current_quota_size"], "small")
 
+    @conf_utils.modify_conf(
+        CONF,
+        operations={
+            "adjutant.quota.services": [
+                {
+                    "operation": "override",
+                    "value": {
+                        "RegionOne": [],
+                        "*": ["cinder", "neutron", "nova"],
+                    },
+                },
+            ],
+        },
+    )
+    def test_update_quota_disabled_region(self):
+        """Check that if a quota update request is made for a disabled region,
+        the request is denied and the quota is not changed.
+        """
+
+        project = fake_clients.FakeProject(
+            name="test_project",
+            id="test_project_id",
+        )
+
+        user = fake_clients.FakeUser(
+            name="test@example.com", password="123", email="test@example.com"
+        )
+
+        setup_identity_cache(projects=[project], users=[user])
+
+        admin_headers = {
+            "project_name": "test_project",
+            "project_id": project.id,
+            "roles": "project_admin,member,project_mod",
+            "username": "test@example.com",
+            "user_id": "user_id",
+            "authenticated": True,
+        }
+
+        url = "/v1/openstack/quotas/"
+
+        data = {"size": "medium", "regions": ["RegionOne"]}
+
+        response = self.client.post(
+            url,
+            data,
+            headers=admin_headers,
+            format="json",
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+        self.check_quota_cache("RegionOne", project.id, "small")
+
+    @conf_utils.modify_conf(
+        CONF,
+        operations={
+            "adjutant.quota.services": [
+                {
+                    "operation": "override",
+                    "value": {
+                        "RegionTwo": [],
+                        "*": ["cinder", "neutron", "nova"],
+                    },
+                },
+            ],
+        },
+    )
+    def test_update_quota_one_region_disabled(self):
+        """Check that if a quota update request is made for multiple regions
+        and one of them has quota management disabled, only the enabled
+        regions have their quota updated.
+        """
+
+        project = fake_clients.FakeProject(
+            name="test_project",
+            id="test_project_id",
+        )
+
+        user = fake_clients.FakeUser(
+            name="test@example.com", password="123", email="test@example.com"
+        )
+
+        setup_identity_cache(projects=[project], users=[user])
+
+        admin_headers = {
+            "project_name": "test_project",
+            "project_id": project.id,
+            "roles": "project_admin,member,project_mod",
+            "username": "test@example.com",
+            "user_id": "user_id",
+            "authenticated": True,
+        }
+
+        url = "/v1/openstack/quotas/"
+
+        data = {"size": "medium", "regions": ["RegionOne", "RegionTwo"]}
+
+        response = self.client.post(
+            url,
+            data,
+            headers=admin_headers,
+            format="json",
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
+
+        self.check_quota_cache("RegionOne", project.id, "medium")
+        self.check_quota_cache("RegionTwo", project.id, "small")
+
+    @conf_utils.modify_conf(
+        CONF,
+        operations={
+            "adjutant.quota.services": [
+                {"operation": "override", "value": {}},
+            ],
+        },
+    )
+    def test_update_quota_disabled(self):
+        """Check that quota update requests return error responses and
+        updates are not performed if quota management is disabled entirely.
+        """
+
+        project = fake_clients.FakeProject(
+            name="test_project",
+            id="test_project_id",
+        )
+
+        user = fake_clients.FakeUser(
+            name="test@example.com", password="123", email="test@example.com"
+        )
+
+        setup_identity_cache(projects=[project], users=[user])
+
+        admin_headers = {
+            "project_name": "test_project",
+            "project_id": project.id,
+            "roles": "project_admin,member,project_mod",
+            "username": "test@example.com",
+            "user_id": "user_id",
+            "authenticated": True,
+        }
+
+        url = "/v1/openstack/quotas/"
+
+        data = {"size": "medium", "regions": ["RegionOne", "RegionTwo"]}
+
+        response = self.client.post(
+            url,
+            data,
+            headers=admin_headers,
+            format="json",
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+        self.check_quota_cache("RegionOne", project.id, "small")
+        self.check_quota_cache("RegionTwo", project.id, "small")
+
     @conf_utils.modify_conf(
         CONF,
         operations={

adjutant/common/quota.py

@@ -203,17 +203,15 @@ class QuotaManager(object):
         # TODO(amelia): Try to find out which endpoints are available and get
         # the non enabled ones out of the list
 
-        self.default_helpers = dict(self._quota_updaters)
+        self.default_helpers = {}
         self.helpers = {}
 
         quota_services = dict(CONF.quota.services)
 
-        all_regions = quota_services.pop("*", None)
+        all_regions = quota_services.pop("*", [])
-        if all_regions:
+        for service in all_regions:
-            self.default_helpers = {}
+            if service in self._quota_updaters:
-            for service in all_regions:
+                self.default_helpers[service] = self._quota_updaters[service]
-                if service in self._quota_updaters:
-                    self.default_helpers[service] = self._quota_updaters[service]
 
         for region, services in quota_services.items():
             self.helpers[region] = {}
@@ -314,6 +312,12 @@ class QuotaManager(object):
         return quota_list[:list_position]
 
     def get_region_quota_data(self, region_id, include_usage=True):
+        # NOTE(callumdickinson): If the region has no services
+        # for which quotas should be managed, return None so the caller
+        # can handle this case properly.
+        if not self.helpers.get(region_id, self.default_helpers):
+            return None
+
         current_quota = self.get_current_region_quota(region_id)
         current_quota_size = self.get_quota_size(current_quota)
         change_options = self.get_quota_change_options(current_quota_size)
@@ -340,13 +344,18 @@ class QuotaManager(object):
         return current_usage
 
     def set_region_quota(self, region_id, quota_dict):
+        region_helpers = self.helpers.get(region_id, self.default_helpers)
+        if not region_helpers:
+            return [
+                (
+                    "WARNING: Quota management disabled in region "
+                    f"{region_id}, skipping."
+                ),
+            ]
         notes = []
         for service_name, values in quota_dict.items():
-            updater_class = self.helpers.get(region_id, self.default_helpers).get(
+            updater_class = region_helpers.get(service_name)
-                service_name
-            )
             if not updater_class:
-                notes.append("No quota updater found for %s. Ignoring" % service_name)
                 continue
 
             service_helper = updater_class(region_id, self.project_id)

releasenotes/notes/disable-quota-management-feddbaab2c304758.yaml

@@ -0,0 +1,11 @@
+---
+features:
+  - |
+    Quota management can now be disabled for specific regions by setting
+    ``quota.services.<region_name>`` to ``[]`` in the configuration.
+    This allows additional flexibility in what services are deployed to
+    which region.
+  - |
+    Quota management can now be disabled entirely in Adjutant by setting
+    ``quota.services`` to ``{}`` in the configuration, if quota management
+    is not required in deployments.