Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use get_registered_limits method in tempest

Best Python code snippet using tempest_python

test_limit.py

Source:test_limit.py

...159        self.assertRaises(ValueError,160                          enforcer.calculate_usage,161                          'project', ['a', 123, 'b'])162    @mock.patch.object(limit._EnforcerUtils, "get_registered_limits")163    def test_get_registered_limits(self, mock_get_limits):164        mock_get_limits.return_value = [("a", 1), ("b", 0), ("c", 2)]165        enforcer = limit.Enforcer(lambda: None)166        limits = enforcer.get_registered_limits(["a", "b", "c"])167        mock_get_limits.assert_called_once_with(["a", "b", "c"])168        self.assertEqual(mock_get_limits.return_value, limits)169    @mock.patch.object(limit._EnforcerUtils, "get_project_limits")170    def test_get_project_limits(self, mock_get_limits):171        project_id = uuid.uuid4().hex172        mock_get_limits.return_value = [("a", 1), ("b", 0), ("c", 2)]173        enforcer = limit.Enforcer(lambda: None)174        limits = enforcer.get_project_limits(project_id, ["a", "b", "c"])175        mock_get_limits.assert_called_once_with(project_id, ["a", "b", "c"])176        self.assertEqual(mock_get_limits.return_value, limits)177class TestFlatEnforcer(base.BaseTestCase):178    def setUp(self):179        super(TestFlatEnforcer, self).setUp()180        self.mock_conn = mock.MagicMock()181        limit._SDK_CONNECTION = self.mock_conn182    @mock.patch.object(limit._EnforcerUtils, "get_registered_limits")183    def test_get_registered_limits(self, mock_get_limits):184        mock_get_limits.return_value = [("a", 1), ("b", 0), ("c", 2)]185        enforcer = limit._FlatEnforcer(lambda: None)186        limits = enforcer.get_registered_limits(["a", "b", "c"])187        mock_get_limits.assert_called_once_with(["a", "b", "c"])188        self.assertEqual(mock_get_limits.return_value, limits)189    @mock.patch.object(limit._EnforcerUtils, "get_project_limits")190    def test_get_project_limits(self, mock_get_limits):191        project_id = uuid.uuid4().hex192        mock_get_limits.return_value = [("a", 1), ("b", 0), ("c", 2)]193        enforcer = limit._FlatEnforcer(lambda: None)194        limits = enforcer.get_project_limits(project_id, ["a", "b", "c"])195        mock_get_limits.assert_called_once_with(project_id, ["a", "b", "c"])196        self.assertEqual(mock_get_limits.return_value, limits)197    @mock.patch.object(limit._EnforcerUtils, "get_project_limits")198    def test_enforce(self, mock_get_limits):199        mock_usage = mock.MagicMock()200        project_id = uuid.uuid4().hex201        deltas = {"a": 1, "b": 1}202        mock_get_limits.return_value = [("a", 1), ("b", 2)]203        mock_usage.return_value = {"a": 0, "b": 1}204        enforcer = limit._FlatEnforcer(mock_usage)205        enforcer.enforce(project_id, deltas)206        self.mock_conn.get_endpoint.assert_called_once_with(None)207        mock_get_limits.assert_called_once_with(project_id, ["a", "b"])208        mock_usage.assert_called_once_with(project_id, ["a", "b"])209    @mock.patch.object(limit._EnforcerUtils, "get_project_limits")210    def test_enforce_raises_on_over(self, mock_get_limits):211        mock_usage = mock.MagicMock()212        project_id = uuid.uuid4().hex213        deltas = {"a": 2, "b": 1}214        mock_get_limits.return_value = [("a", 1), ("b", 2)]215        mock_usage.return_value = {"a": 0, "b": 1}216        enforcer = limit._FlatEnforcer(mock_usage)217        e = self.assertRaises(exception.ProjectOverLimit, enforcer.enforce,218                              project_id, deltas)219        expected = ("Project %s is over a limit for "220                    "[Resource a is over limit of 1 due to current usage 0 "221                    "and delta 2]")222        self.assertEqual(expected % project_id, str(e))223        self.assertEqual(project_id, e.project_id)224        self.assertEqual(1, len(e.over_limit_info_list))225        over_a = e.over_limit_info_list[0]226        self.assertEqual("a", over_a.resource_name)227        self.assertEqual(1, over_a.limit)228        self.assertEqual(0, over_a.current_usage)229        self.assertEqual(2, over_a.delta)230    @mock.patch.object(limit._EnforcerUtils, "_get_project_limit")231    @mock.patch.object(limit._EnforcerUtils, "_get_registered_limit")232    def test_enforce_raises_on_missing_limit(self, mock_get_reglimit,233                                             mock_get_limit):234        def mock_usage(*a):235            return {'a': 1, 'b': 1}236        project_id = uuid.uuid4().hex237        deltas = {"a": 0, "b": 0}238        mock_get_reglimit.return_value = None239        mock_get_limit.return_value = None240        enforcer = limit._FlatEnforcer(mock_usage)241        self.assertRaises(exception.ProjectOverLimit, enforcer.enforce,242                          project_id, deltas)243        self.assertRaises(exception.ProjectOverLimit, enforcer.enforce,244                          None, deltas)245class TestEnforcerUtils(base.BaseTestCase):246    def setUp(self):247        super(TestEnforcerUtils, self).setUp()248        self.mock_conn = mock.MagicMock()249        limit._SDK_CONNECTION = self.mock_conn250    def test_get_endpoint(self):251        fake_endpoint = endpoint.Endpoint()252        self.mock_conn.get_endpoint.return_value = fake_endpoint253        utils = limit._EnforcerUtils()254        self.assertEqual(fake_endpoint, utils._endpoint)255        self.mock_conn.get_endpoint.assert_called_once_with(None)256    def test_get_registered_limit_empty(self):257        self.mock_conn.registered_limits.return_value = iter([])258        utils = limit._EnforcerUtils()259        reg_limit = utils._get_registered_limit("foo")260        self.assertIsNone(reg_limit)261    def test_get_registered_limit(self):262        foo = registered_limit.RegisteredLimit()263        foo.resource_name = "foo"264        self.mock_conn.registered_limits.return_value = iter([foo])265        utils = limit._EnforcerUtils()266        reg_limit = utils._get_registered_limit("foo")267        self.assertEqual(foo, reg_limit)268    def test_get_registered_limits(self):269        fake_endpoint = endpoint.Endpoint()270        fake_endpoint.service_id = "service_id"271        fake_endpoint.region_id = "region_id"272        self.mock_conn.get_endpoint.return_value = fake_endpoint273        # a and c have limits, b doesn't have one274        empty_iterator = iter([])275        a = registered_limit.RegisteredLimit()276        a.resource_name = "a"277        a.default_limit = 1278        a_iterator = iter([a])279        c = registered_limit.RegisteredLimit()280        c.resource_name = "c"281        c.default_limit = 2282        c_iterator = iter([c])283        self.mock_conn.registered_limits.side_effect = [a_iterator,284                                                        empty_iterator,285                                                        c_iterator]286        utils = limit._EnforcerUtils()287        limits = utils.get_registered_limits(["a", "b", "c"])288        self.assertEqual([('a', 1), ('b', 0), ('c', 2)], limits)289    def test_get_project_limits(self):290        fake_endpoint = endpoint.Endpoint()291        fake_endpoint.service_id = "service_id"292        fake_endpoint.region_id = "region_id"293        self.mock_conn.get_endpoint.return_value = fake_endpoint294        project_id = uuid.uuid4().hex295        # a is a project limit, b, c and d don't have one296        empty_iterator = iter([])297        a = klimit.Limit()298        a.resource_name = "a"299        a.resource_limit = 1300        a_iterator = iter([a])301        self.mock_conn.limits.side_effect = [a_iterator, empty_iterator,...

limit.py

Source:limit.py

...142        limits = self.model.get_project_limits(project_id, resources_to_check)143        usage = self.model.get_project_usage(project_id, resources_to_check)144        return {resource: ProjectUsage(limit, usage[resource])145                for resource, limit in limits}146    def get_registered_limits(self, resources_to_check):147        return self.model.get_registered_limits(resources_to_check)148    def get_project_limits(self, project_id, resources_to_check):149        return self.model.get_project_limits(project_id,150                                             resources_to_check)151class _FlatEnforcer(object):152    name = 'flat'153    def __init__(self, usage_callback, cache=True):154        self._usage_callback = usage_callback155        self._utils = _EnforcerUtils(cache=cache)156    def get_registered_limits(self, resources_to_check):157        return self._utils.get_registered_limits(resources_to_check)158    def get_project_limits(self, project_id, resources_to_check):159        return self._utils.get_project_limits(project_id, resources_to_check)160    def get_project_usage(self, project_id, resources_to_check):161        return self._usage_callback(project_id, resources_to_check)162    def enforce(self, project_id, deltas):163        resources_to_check = list(deltas.keys())164        # Always check the limits in the same order, for predictable errors165        resources_to_check.sort()166        project_limits = self.get_project_limits(project_id,167                                                 resources_to_check)168        current_usage = self.get_project_usage(project_id,169                                               resources_to_check)170        self._utils.enforce_limits(project_id, project_limits,171                                   current_usage, deltas)172class _StrictTwoLevelEnforcer(object):173    name = 'strict-two-level'174    def __init__(self, usage_callback, cache=True):175        self._usage_callback = usage_callback176    def get_registered_limits(self, resources_to_check):177        raise NotImplementedError()178    def get_project_limits(self, project_id, resources_to_check):179        raise NotImplementedError()180    def get_project_usage(self, project_id, resources_to_check):181        raise NotImplementedError()182    def enforce(self, project_id, deltas):183        raise NotImplementedError()184_MODELS = [_FlatEnforcer, _StrictTwoLevelEnforcer]185class _LimitNotFound(Exception):186    def __init__(self, resource):187        msg = "Can't find the limit for resource %(resource)s" % {188            'resource': resource}189        self.resource = resource190        super(_LimitNotFound, self).__init__(msg)191class _EnforcerUtils(object):192    """Logic common used by multiple enforcers"""193    def __init__(self, cache=True):194        self.connection = _get_keystone_connection()195        self.should_cache = cache196        # {project_id: {resource_name: project_limit}}197        self.plimit_cache = defaultdict(dict)198        # {resource_name: registered_limit}199        self.rlimit_cache = {}200        # get and cache endpoint info201        endpoint_id = CONF.oslo_limit.endpoint_id202        self._endpoint = self.connection.get_endpoint(endpoint_id)203        if not self._endpoint:204            raise ValueError("can't find endpoint for %s" % endpoint_id)205        self._service_id = self._endpoint.service_id206        self._region_id = self._endpoint.region_id207    @staticmethod208    def enforce_limits(project_id, limits, current_usage, deltas):209        """Check that proposed usage is not over given limits210        :param project_id: project being checked or None211        :param limits: list of (resource_name,limit) pairs212        :param current_usage: dict of resource name and current usage213        :param deltas: dict of resource name and proposed additional usage214        :raises exception.ClaimExceedsLimit: raise if over limit215        """216        over_limit_list = []217        for resource_name, limit in limits:218            if resource_name not in current_usage:219                msg = "unable to get current usage for %s" % resource_name220                raise ValueError(msg)221            current = int(current_usage[resource_name])222            delta = int(deltas[resource_name])223            proposed_usage_total = current + delta224            if proposed_usage_total > limit:225                over_limit_list.append(exception.OverLimitInfo(226                    resource_name, limit, current, delta))227        if len(over_limit_list) > 0:228            LOG.debug("hit limit for project: %s", over_limit_list)229            raise exception.ProjectOverLimit(project_id, over_limit_list)230    def get_registered_limits(self, resource_names):231        """Get all the default limits for a given resource name list232        :param resource_names: list of resource_name strings233        :return: list of (resource_name, limit) pairs234        """235        # Using a list to preserve the resource_name order236        registered_limits = []237        for resource_name in resource_names:238            reg_limit = self._get_registered_limit(resource_name)239            if reg_limit:240                limit = reg_limit.default_limit241            else:242                limit = 0243            registered_limits.append((resource_name, limit))244        return registered_limits...

limits_client.py

Source:limits_client.py

...14from oslo_serialization import jsonutils as json15from tempest.lib.common import rest_client16class LimitsClient(rest_client.RestClient):17    api_version = "v3"18    def get_registered_limits(self):19        """Lists all registered limits."""20        resp, body = self.get('registered_limits')21        self.expected_success(200, resp.status)22        return rest_client.ResponseBody(resp, json.loads(body))23    def create_limit(self, region_id, service_id, project_id, resource_name,24                     default_limit, description=None, domain_id=None):25        """Creates a limit in keystone."""26        limit = {27            'service_id': service_id,28            'project_id': project_id,29            'resource_name': resource_name,30            'resource_limit': default_limit,31            'region_id': region_id,32            'description': description or '%s limit for %s' % (...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.