1# Licensed under the Apache License, Version 2.0 (the "License"); you may
2# not use this file except in compliance with the License. You may obtain
3# a copy of the License at
4#
5#      http://www.apache.org/licenses/LICENSE-2.0
6#
7# Unless required by applicable law or agreed to in writing, software
8# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
9# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
10# License for the specific language governing permissions and limitations
11# under the License.
12
13import copy
14from unittest import mock
15
16from keystoneauth1 import adapter
17
18from openstack.compute.v2 import limits
19from openstack.tests.unit import base
20
21ABSOLUTE_LIMITS = {
22    "maxImageMeta": 128,
23    "maxSecurityGroupRules": 20,
24    "maxSecurityGroups": 10,
25    "maxServerMeta": 128,
26    "maxTotalCores": 20,
27    "maxTotalFloatingIps": 10,
28    "maxTotalInstances": 10,
29    "maxTotalKeypairs": 100,
30    "maxTotalRAMSize": 51200,
31    "maxServerGroups": 10,
32    "maxServerGroupMembers": 10,
33    "totalFloatingIpsUsed": 1,
34    "totalSecurityGroupsUsed": 2,
35    "totalRAMUsed": 4,
36    "totalInstancesUsed": 5,
37    "totalServerGroupsUsed": 6,
38    "totalCoresUsed": 7
39}
40
41RATE_LIMIT = {
42    "limit": [
43        {
44            "next-available": "2012-11-27T17:22:18Z",
45            "remaining": 120,
46            "unit": "MINUTE",
47            "value": 120,
48            "verb": "POST"
49        },
50    ],
51    "regex": ".*",
52    "uri": "*"
53}
54
55LIMITS_BODY = {
56    "limits": {
57        "absolute": ABSOLUTE_LIMITS,
58        "rate": [RATE_LIMIT]
59    }
60}
61
62
63class TestAbsoluteLimits(base.TestCase):
64
65    def test_basic(self):
66        sot = limits.AbsoluteLimits()
67        self.assertIsNone(sot.resource_key)
68        self.assertIsNone(sot.resources_key)
69        self.assertEqual("", sot.base_path)
70        self.assertFalse(sot.allow_create)
71        self.assertFalse(sot.allow_fetch)
72        self.assertFalse(sot.allow_commit)
73        self.assertFalse(sot.allow_delete)
74        self.assertFalse(sot.allow_list)
75
76    def test_make_it(self):
77        sot = limits.AbsoluteLimits(**ABSOLUTE_LIMITS)
78        self.assertEqual(ABSOLUTE_LIMITS["maxImageMeta"], sot.image_meta)
79        self.assertEqual(ABSOLUTE_LIMITS["maxSecurityGroupRules"],
80                         sot.security_group_rules)
81        self.assertEqual(ABSOLUTE_LIMITS["maxSecurityGroups"],
82                         sot.security_groups)
83        self.assertEqual(ABSOLUTE_LIMITS["maxServerMeta"], sot.server_meta)
84        self.assertEqual(ABSOLUTE_LIMITS["maxTotalCores"], sot.total_cores)
85        self.assertEqual(ABSOLUTE_LIMITS["maxTotalFloatingIps"],
86                         sot.floating_ips)
87        self.assertEqual(ABSOLUTE_LIMITS["maxTotalInstances"],
88                         sot.instances)
89        self.assertEqual(ABSOLUTE_LIMITS["maxTotalKeypairs"],
90                         sot.keypairs)
91        self.assertEqual(ABSOLUTE_LIMITS["maxTotalRAMSize"],
92                         sot.total_ram)
93        self.assertEqual(ABSOLUTE_LIMITS["maxServerGroups"], sot.server_groups)
94        self.assertEqual(ABSOLUTE_LIMITS["maxServerGroupMembers"],
95                         sot.server_group_members)
96        self.assertEqual(ABSOLUTE_LIMITS["totalFloatingIpsUsed"],
97                         sot.floating_ips_used)
98        self.assertEqual(ABSOLUTE_LIMITS["totalSecurityGroupsUsed"],
99                         sot.security_groups_used)
100        self.assertEqual(ABSOLUTE_LIMITS["totalRAMUsed"], sot.total_ram_used)
101        self.assertEqual(ABSOLUTE_LIMITS["totalInstancesUsed"],
102                         sot.instances_used)
103        self.assertEqual(ABSOLUTE_LIMITS["totalServerGroupsUsed"],
104                         sot.server_groups_used)
105        self.assertEqual(ABSOLUTE_LIMITS["totalCoresUsed"],
106                         sot.total_cores_used)
107
108
109class TestRateLimit(base.TestCase):
110
111    def test_basic(self):
112        sot = limits.RateLimit()
113        self.assertIsNone(sot.resource_key)
114        self.assertIsNone(sot.resources_key)
115        self.assertEqual("", sot.base_path)
116        self.assertIsNone(sot.service)
117        self.assertFalse(sot.allow_create)
118        self.assertFalse(sot.allow_fetch)
119        self.assertFalse(sot.allow_commit)
120        self.assertFalse(sot.allow_delete)
121        self.assertFalse(sot.allow_list)
122
123    def test_make_it(self):
124        sot = limits.RateLimit(**RATE_LIMIT)
125        self.assertEqual(RATE_LIMIT["regex"], sot.regex)
126        self.assertEqual(RATE_LIMIT["uri"], sot.uri)
127        self.assertEqual(RATE_LIMIT["limit"], sot.limits)
128
129
130class TestLimits(base.TestCase):
131
132    def test_basic(self):
133        sot = limits.Limits()
134        self.assertEqual("limits", sot.resource_key)
135        self.assertEqual("/limits", sot.base_path)
136        self.assertTrue(sot.allow_fetch)
137        self.assertFalse(sot.allow_create)
138        self.assertFalse(sot.allow_commit)
139        self.assertFalse(sot.allow_delete)
140        self.assertFalse(sot.allow_list)
141
142    def test_get(self):
143        sess = mock.Mock(spec=adapter.Adapter)
144        sess.default_microversion = None
145        resp = mock.Mock()
146        sess.get.return_value = resp
147        resp.json.return_value = copy.deepcopy(LIMITS_BODY)
148        resp.headers = {}
149        resp.status_code = 200
150
151        sot = limits.Limits().fetch(sess)
152
153        self.assertEqual(ABSOLUTE_LIMITS["maxImageMeta"],
154                         sot.absolute.image_meta)
155        self.assertEqual(ABSOLUTE_LIMITS["maxSecurityGroupRules"],
156                         sot.absolute.security_group_rules)
157        self.assertEqual(ABSOLUTE_LIMITS["maxSecurityGroups"],
158                         sot.absolute.security_groups)
159        self.assertEqual(ABSOLUTE_LIMITS["maxServerMeta"],
160                         sot.absolute.server_meta)
161        self.assertEqual(ABSOLUTE_LIMITS["maxTotalCores"],
162                         sot.absolute.total_cores)
163        self.assertEqual(ABSOLUTE_LIMITS["maxTotalFloatingIps"],
164                         sot.absolute.floating_ips)
165        self.assertEqual(ABSOLUTE_LIMITS["maxTotalInstances"],
166                         sot.absolute.instances)
167        self.assertEqual(ABSOLUTE_LIMITS["maxTotalKeypairs"],
168                         sot.absolute.keypairs)
169        self.assertEqual(ABSOLUTE_LIMITS["maxTotalRAMSize"],
170                         sot.absolute.total_ram)
171        self.assertEqual(ABSOLUTE_LIMITS["maxServerGroups"],
172                         sot.absolute.server_groups)
173        self.assertEqual(ABSOLUTE_LIMITS["maxServerGroupMembers"],
174                         sot.absolute.server_group_members)
175        self.assertEqual(ABSOLUTE_LIMITS["totalFloatingIpsUsed"],
176                         sot.absolute.floating_ips_used)
177        self.assertEqual(ABSOLUTE_LIMITS["totalSecurityGroupsUsed"],
178                         sot.absolute.security_groups_used)
179        self.assertEqual(ABSOLUTE_LIMITS["totalRAMUsed"],
180                         sot.absolute.total_ram_used)
181        self.assertEqual(ABSOLUTE_LIMITS["totalInstancesUsed"],
182                         sot.absolute.instances_used)
183        self.assertEqual(ABSOLUTE_LIMITS["totalServerGroupsUsed"],
184                         sot.absolute.server_groups_used)
185        self.assertEqual(ABSOLUTE_LIMITS["totalCoresUsed"],
186                         sot.absolute.total_cores_used)
187
188        self.assertEqual(RATE_LIMIT["uri"], sot.rate[0].uri)
189        self.assertEqual(RATE_LIMIT["regex"], sot.rate[0].regex)
190        self.assertEqual(RATE_LIMIT["limit"], sot.rate[0].limits)
191
192        dsot = sot.to_dict()
193
194        self.assertIsInstance(dsot['rate'][0], dict)
195        self.assertIsInstance(dsot['absolute'], dict)
196        self.assertEqual(RATE_LIMIT["uri"], dsot['rate'][0]['uri'])
197        self.assertEqual(
198            ABSOLUTE_LIMITS["totalSecurityGroupsUsed"],
199            dsot['absolute']['security_groups_used'])
200