Add network update quota "limit_check" parameter

This new parameter commands the Neutron server to first check the
resource usage before setting the new quota limit. If the resource
usage is below the new limit, the Neutron server will raise an
exception.

Depends-On: https://review.opendev.org/c/openstack/openstacksdk/+/806254
Depends-On: https://review.opendev.org/c/openstack/neutron/+/801470

Partial-Bug: #1936408
Change-Id: Idc1b99492d609eb699d0a6bef6cd760458a774f6

3 files changed, 77 insertions, 0 deletions

diff --git a/openstackclient/common/quota.py b/openstackclient/common/quota.py
index 643cb4e4..677cba03 100644
--- a/openstackclient/common/quota.py
+++ b/openstackclient/common/quota.py
@@ -535,6 +535,12 @@ class SetQuota(common.NetDetectionMixin, command.Command):
             action='store_true',
             help=_('Force quota update (only supported by compute)')
         )
+        parser.add_argument(
+            '--check-limit',
+            action='store_true',
+            help=_('Check quota limit when updating (only supported by '
+                   'network)')
+        )
         return parser
 
     def take_action(self, parsed_args):
@@ -561,6 +567,9 @@ class SetQuota(common.NetDetectionMixin, command.Command):
                 volume_kwargs[k] = value
 
         network_kwargs = {}
+        if parsed_args.check_limit:
+            network_kwargs['check_limit'] = True
+
         if self.app.client_manager.is_network_endpoint_enabled():
             for k, v in NETWORK_QUOTAS.items():
                 value = getattr(parsed_args, k, None)
diff --git a/openstackclient/tests/functional/common/test_quota.py b/openstackclient/tests/functional/common/test_quota.py
index 9c057460..bf67101a 100644
--- a/openstackclient/tests/functional/common/test_quota.py
+++ b/openstackclient/tests/functional/common/test_quota.py
@@ -11,6 +11,9 @@
 #    under the License.
 
 import json
+import uuid
+
+from tempest.lib import exceptions
 
 from openstackclient.tests.functional import base
 
@@ -165,3 +168,25 @@ class QuotaTests(base.TestCase):
         # returned attributes
         self.assertTrue(cmd_output["key-pairs"] >= 0)
         self.assertTrue(cmd_output["snapshots"] >= 0)
+
+    def test_quota_network_set_with_check_limit(self):
+        if not self.haz_network:
+            self.skipTest('No Network service present')
+        if not self.is_extension_enabled('quota-check-limit'):
+            self.skipTest('No "quota-check-limit" extension present')
+
+        self.openstack('quota set --networks 40 ' + self.PROJECT_NAME)
+        cmd_output = json.loads(self.openstack(
+            'quota list -f json --network'
+        ))
+        self.assertIsNotNone(cmd_output)
+        self.assertEqual(40, cmd_output[0]['Networks'])
+
+        # That will ensure we have at least two networks in the system.
+        for _ in range(2):
+            self.openstack('network create --project %s %s' %
+                           (self.PROJECT_NAME, uuid.uuid4().hex))
+
+        self.assertRaises(exceptions.CommandFailed, self.openstack,
+                          'quota set --networks 1 --check-limit ' +
+                          self.PROJECT_NAME)
diff --git a/openstackclient/tests/unit/common/test_quota.py b/openstackclient/tests/unit/common/test_quota.py
index 8771359c..896a63a7 100644
--- a/openstackclient/tests/unit/common/test_quota.py
+++ b/openstackclient/tests/unit/common/test_quota.py
@@ -950,6 +950,49 @@ class TestQuotaSet(TestQuota):
         )
         self.assertIsNone(result)
 
+    def test_quota_set_with_check_limit(self):
+        arglist = [
+            '--subnets', str(network_fakes.QUOTA['subnet']),
+            '--volumes', str(volume_fakes.QUOTA['volumes']),
+            '--cores', str(compute_fakes.core_num),
+            '--check-limit',
+            self.projects[0].name,
+        ]
+        verifylist = [
+            ('subnet', network_fakes.QUOTA['subnet']),
+            ('volumes', volume_fakes.QUOTA['volumes']),
+            ('cores', compute_fakes.core_num),
+            ('check_limit', True),
+            ('project', self.projects[0].name),
+        ]
+        parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+        result = self.cmd.take_action(parsed_args)
+
+        kwargs_compute = {
+            'cores': compute_fakes.core_num,
+        }
+        kwargs_volume = {
+            'volumes': volume_fakes.QUOTA['volumes'],
+        }
+        kwargs_network = {
+            'subnet': network_fakes.QUOTA['subnet'],
+            'check_limit': True,
+        }
+        self.compute_quotas_mock.update.assert_called_once_with(
+            self.projects[0].id,
+            **kwargs_compute
+        )
+        self.volume_quotas_mock.update.assert_called_once_with(
+            self.projects[0].id,
+            **kwargs_volume
+        )
+        self.network_mock.update_quota.assert_called_once_with(
+            self.projects[0].id,
+            **kwargs_network
+        )
+        self.assertIsNone(result)
+
 
 class TestQuotaShow(TestQuota):