"Fossies" - the Fresh Open Source Software Archive

Member "neutron-14.0.3/neutron/tests/unit/db/quota/test_driver.py" (22 Oct 2019, 12989 Bytes) of package /linux/misc/openstack/neutron-14.0.3.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file.

    1 # Copyright (c) 2014 OpenStack Foundation.
    2 #
    3 # Licensed under the Apache License, Version 2.0 (the "License");
    4 # you may not use this file except in compliance with the License.
    5 # You may obtain a copy of the License at
    6 #
    7 #    http://www.apache.org/licenses/LICENSE-2.0
    8 #
    9 # Unless required by applicable law or agreed to in writing, software
   10 # distributed under the License is distributed on an "AS IS" BASIS,
   11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
   12 # implied.
   13 # See the License for the specific language governing permissions and
   14 # limitations under the License.
   15 
   16 from neutron_lib import context
   17 from neutron_lib import exceptions
   18 
   19 from neutron.db import db_base_plugin_v2 as base_plugin
   20 from neutron.db.quota import api as quota_api
   21 from neutron.db.quota import driver
   22 from neutron.objects import quota as quota_obj
   23 from neutron.quota import resource
   24 from neutron.tests import base
   25 from neutron.tests.unit import quota as test_quota
   26 from neutron.tests.unit import testlib_api
   27 
   28 DB_PLUGIN_KLASS = 'neutron.db.db_base_plugin_v2.NeutronDbPluginV2'
   29 
   30 
   31 def _count_resource(context, resource, tenant_id):
   32     """A fake counting function to determine current used counts"""
   33     if resource[-1] == 's':
   34         resource = resource[:-1]
   35     result = quota_obj.QuotaUsage.get_object_dirty_protected(
   36         context, resource=resource)
   37     return 0 if not result else result.in_use
   38 
   39 
   40 class FakePlugin(base_plugin.NeutronDbPluginV2, driver.DbQuotaDriver):
   41     """A fake plugin class containing all DB methods."""
   42 
   43 
   44 class TestResource(object):
   45     """Describe a test resource for quota checking."""
   46 
   47     def __init__(self, name, default, fake_count=0):
   48         self.name = name
   49         self.quota = default
   50         self.fake_count = fake_count
   51 
   52     @property
   53     def default(self):
   54         return self.quota
   55 
   56     def count(self, *args, **kwargs):
   57         return self.fake_count
   58 
   59 
   60 class TestTrackedResource(resource.TrackedResource):
   61     """Describes a test tracked resource for detailed quota checking"""
   62     def __init__(self, name, model_class, flag=None,
   63                  plural_name=None):
   64         super(TestTrackedResource, self).__init__(
   65             name, model_class, flag=flag, plural_name=None)
   66 
   67     @property
   68     def default(self):
   69         return self.flag
   70 
   71 
   72 class TestCountableResource(resource.CountableResource):
   73     """Describes a test countable resource for detailed quota checking"""
   74     def __init__(self, name, count, flag=-1, plural_name=None):
   75         super(TestCountableResource, self).__init__(
   76             name, count, flag=flag, plural_name=None)
   77 
   78     @property
   79     def default(self):
   80         return self.flag
   81 
   82 
   83 PROJECT = 'prj_test'
   84 RESOURCE = 'res_test'
   85 ALT_RESOURCE = 'res_test_meh'
   86 
   87 
   88 class TestDbQuotaDriver(testlib_api.SqlTestCase,
   89                         base.BaseTestCase):
   90     def setUp(self):
   91         super(TestDbQuotaDriver, self).setUp()
   92         self.plugin = FakePlugin()
   93         self.context = context.get_admin_context()
   94         self.setup_coreplugin(core_plugin=DB_PLUGIN_KLASS)
   95 
   96     def test_create_quota_limit(self):
   97         defaults = {RESOURCE: TestResource(RESOURCE, 4)}
   98 
   99         self.plugin.update_quota_limit(self.context, PROJECT, RESOURCE, 2)
  100         quotas = self.plugin.get_tenant_quotas(self.context, defaults, PROJECT)
  101         self.assertEqual(2, quotas[RESOURCE])
  102 
  103     def test_update_quota_limit(self):
  104         defaults = {RESOURCE: TestResource(RESOURCE, 4)}
  105 
  106         self.plugin.update_quota_limit(self.context, PROJECT, RESOURCE, 2)
  107         self.plugin.update_quota_limit(self.context, PROJECT, RESOURCE, 3)
  108         quotas = self.plugin.get_tenant_quotas(self.context, defaults, PROJECT)
  109         self.assertEqual(3, quotas[RESOURCE])
  110 
  111     def test_delete_tenant_quota_restores_default_limit(self):
  112         defaults = {RESOURCE: TestResource(RESOURCE, 4)}
  113 
  114         self.plugin.update_quota_limit(self.context, PROJECT, RESOURCE, 2)
  115         self.plugin.delete_tenant_quota(self.context, PROJECT)
  116         quotas = self.plugin.get_tenant_quotas(self.context, defaults, PROJECT)
  117         self.assertEqual(4, quotas[RESOURCE])
  118 
  119     def test_get_default_quotas(self):
  120         defaults = {RESOURCE: TestResource(RESOURCE, 4)}
  121         user_ctx = context.Context(user_id=PROJECT, tenant_id=PROJECT)
  122         self.plugin.update_quota_limit(self.context, PROJECT, RESOURCE, 2)
  123         quotas = self.plugin.get_default_quotas(user_ctx, defaults, PROJECT)
  124         self.assertEqual(4, quotas[RESOURCE])
  125 
  126     def test_get_tenant_quotas(self):
  127         user_ctx = context.Context(user_id=PROJECT, tenant_id=PROJECT)
  128         self.plugin.update_quota_limit(self.context, PROJECT, RESOURCE, 2)
  129         quotas = self.plugin.get_tenant_quotas(user_ctx, {}, PROJECT)
  130         self.assertEqual(2, quotas[RESOURCE])
  131 
  132     def test_get_tenant_quotas_different_tenant(self):
  133         user_ctx = context.Context(user_id=PROJECT,
  134                                    tenant_id='another_project')
  135         self.plugin.update_quota_limit(self.context, PROJECT, RESOURCE, 2)
  136         # It is appropriate to use assertFalse here as the expected return
  137         # value is an empty dict (the defaults passed in the statement below
  138         # after the request context)
  139         self.assertFalse(self.plugin.get_tenant_quotas(user_ctx, {}, PROJECT))
  140 
  141     def test_get_all_quotas(self):
  142         project_1 = 'prj_test_1'
  143         project_2 = 'prj_test_2'
  144         resource_1 = 'res_test_1'
  145         resource_2 = 'res_test_2'
  146 
  147         resources = {resource_1: TestResource(resource_1, 3),
  148                      resource_2: TestResource(resource_2, 5)}
  149 
  150         self.plugin.update_quota_limit(self.context, project_1, resource_1, 7)
  151         self.plugin.update_quota_limit(self.context, project_2, resource_2, 9)
  152         quotas = self.plugin.get_all_quotas(self.context, resources)
  153 
  154         # Expect two tenants' quotas
  155         self.assertEqual(2, len(quotas))
  156         # But not quotas for the same tenant twice
  157         self.assertNotEqual(quotas[0]['tenant_id'], quotas[1]['tenant_id'])
  158 
  159         # Check the expected limits. The quotas can be in any order.
  160         for quota in quotas:
  161             project = quota['tenant_id']
  162             self.assertIn(project, (project_1, project_2))
  163             if project == project_1:
  164                 expected_limit_r1 = 7
  165                 expected_limit_r2 = 5
  166             if project == project_2:
  167                 expected_limit_r1 = 3
  168                 expected_limit_r2 = 9
  169             self.assertEqual(expected_limit_r1, quota[resource_1])
  170             self.assertEqual(expected_limit_r2, quota[resource_2])
  171 
  172     def test_limit_check(self):
  173         resources = {RESOURCE: TestResource(RESOURCE, 2)}
  174         values = {RESOURCE: 1}
  175 
  176         self.plugin.update_quota_limit(self.context, PROJECT, RESOURCE, 2)
  177         self.plugin.limit_check(self.context, PROJECT, resources, values)
  178 
  179     def test_limit_check_over_quota(self):
  180         resources = {RESOURCE: TestResource(RESOURCE, 2)}
  181         values = {RESOURCE: 3}
  182 
  183         self.plugin.update_quota_limit(self.context, PROJECT, RESOURCE, 2)
  184 
  185         self.assertRaises(exceptions.OverQuota, self.plugin.limit_check,
  186                           context.get_admin_context(), PROJECT, resources,
  187                           values)
  188 
  189     def test_limit_check_equals_to_quota(self):
  190         resources = {RESOURCE: TestResource(RESOURCE, 2)}
  191         values = {RESOURCE: 2}
  192 
  193         self.plugin.update_quota_limit(self.context, PROJECT, RESOURCE, 2)
  194         self.plugin.limit_check(self.context, PROJECT, resources, values)
  195 
  196     def test_limit_check_value_lower_than_zero(self):
  197         resources = {RESOURCE: TestResource(RESOURCE, 2)}
  198         values = {RESOURCE: -1}
  199 
  200         self.plugin.update_quota_limit(self.context, PROJECT, RESOURCE, 2)
  201         self.assertRaises(exceptions.InvalidQuotaValue,
  202                           self.plugin.limit_check, context.get_admin_context(),
  203                           PROJECT, resources, values)
  204 
  205     def _test_make_reservation_success(self, quota_driver,
  206                                        resource_name, deltas):
  207         resources = {resource_name: TestResource(resource_name, 2)}
  208         self.plugin.update_quota_limit(self.context, PROJECT, resource_name, 2)
  209         reservation = quota_driver.make_reservation(
  210             self.context,
  211             self.context.tenant_id,
  212             resources,
  213             deltas,
  214             self.plugin)
  215         self.assertIn(resource_name, reservation.deltas)
  216         self.assertEqual(deltas[resource_name],
  217                          reservation.deltas[resource_name])
  218         self.assertEqual(self.context.tenant_id,
  219                          reservation.tenant_id)
  220 
  221     def test_make_reservation_single_resource(self):
  222         quota_driver = driver.DbQuotaDriver()
  223         self._test_make_reservation_success(
  224             quota_driver, RESOURCE, {RESOURCE: 1})
  225 
  226     def test_make_reservation_fill_quota(self):
  227         quota_driver = driver.DbQuotaDriver()
  228         self._test_make_reservation_success(
  229             quota_driver, RESOURCE, {RESOURCE: 2})
  230 
  231     def test_make_reservation_multiple_resources(self):
  232         quota_driver = driver.DbQuotaDriver()
  233         resources = {RESOURCE: TestResource(RESOURCE, 2),
  234                      ALT_RESOURCE: TestResource(ALT_RESOURCE, 2)}
  235         deltas = {RESOURCE: 1, ALT_RESOURCE: 2}
  236         self.plugin.update_quota_limit(self.context, PROJECT, RESOURCE, 2)
  237         self.plugin.update_quota_limit(self.context, PROJECT, ALT_RESOURCE, 2)
  238         reservation = quota_driver.make_reservation(
  239             self.context,
  240             self.context.tenant_id,
  241             resources,
  242             deltas,
  243             self.plugin)
  244         self.assertIn(RESOURCE, reservation.deltas)
  245         self.assertIn(ALT_RESOURCE, reservation.deltas)
  246         self.assertEqual(1, reservation.deltas[RESOURCE])
  247         self.assertEqual(2, reservation.deltas[ALT_RESOURCE])
  248         self.assertEqual(self.context.tenant_id,
  249                          reservation.tenant_id)
  250 
  251     def test_make_reservation_over_quota_fails(self):
  252         quota_driver = driver.DbQuotaDriver()
  253         resources = {RESOURCE: TestResource(RESOURCE, 2,
  254                                             fake_count=2)}
  255         deltas = {RESOURCE: 1}
  256         self.plugin.update_quota_limit(self.context, PROJECT, RESOURCE, 2)
  257         self.assertRaises(exceptions.OverQuota,
  258                           quota_driver.make_reservation,
  259                           self.context,
  260                           self.context.tenant_id,
  261                           resources,
  262                           deltas,
  263                           self.plugin)
  264 
  265     def test_get_detailed_tenant_quotas_resource(self):
  266         res = {RESOURCE: TestTrackedResource(RESOURCE, test_quota.MehModel)}
  267 
  268         self.plugin.update_quota_limit(self.context, PROJECT, RESOURCE, 6)
  269         quota_driver = driver.DbQuotaDriver()
  270         quota_driver.make_reservation(self.context, PROJECT, res,
  271                                       {RESOURCE: 1}, self.plugin)
  272         quota_api.set_quota_usage(self.context, RESOURCE, PROJECT, 2)
  273         detailed_quota = self.plugin.get_detailed_tenant_quotas(self.context,
  274                                                                 res, PROJECT)
  275         self.assertEqual(6, detailed_quota[RESOURCE]['limit'])
  276         self.assertEqual(2, detailed_quota[RESOURCE]['used'])
  277         self.assertEqual(1, detailed_quota[RESOURCE]['reserved'])
  278 
  279     def test_get_detailed_tenant_quotas_multiple_resource(self):
  280         project_1 = 'prj_test_1'
  281         resource_1 = 'res_test_1'
  282         resource_2 = 'res_test_2'
  283         resources = {resource_1:
  284                      TestTrackedResource(resource_1, test_quota.MehModel),
  285                      resource_2:
  286                      TestCountableResource(resource_2, _count_resource)}
  287 
  288         self.plugin.update_quota_limit(self.context, project_1, resource_1, 6)
  289         self.plugin.update_quota_limit(self.context, project_1, resource_2, 9)
  290         quota_driver = driver.DbQuotaDriver()
  291         quota_driver.make_reservation(self.context, project_1,
  292                                       resources,
  293                                       {resource_1: 1, resource_2: 7},
  294                                       self.plugin)
  295 
  296         quota_api.set_quota_usage(self.context, resource_1, project_1, 2)
  297         quota_api.set_quota_usage(self.context, resource_2, project_1, 3)
  298         detailed_quota = self.plugin.get_detailed_tenant_quotas(self.context,
  299                                                                 resources,
  300                                                                 project_1)
  301 
  302         self.assertEqual(6, detailed_quota[resource_1]['limit'])
  303         self.assertEqual(1, detailed_quota[resource_1]['reserved'])
  304         self.assertEqual(2, detailed_quota[resource_1]['used'])
  305 
  306         self.assertEqual(9, detailed_quota[resource_2]['limit'])
  307         self.assertEqual(7, detailed_quota[resource_2]['reserved'])
  308         self.assertEqual(3, detailed_quota[resource_2]['used'])