← Back to branch summary

~ubuntu-branches/ubuntu/vivid/neutron/vivid-proposed

« back to all changes in this revision

Viewing changes to neutron/tests/unit/db/metering/test_metering_db.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2015-04-15 13:59:07 UTC
  • mfrom: (1.1.22)
  • Revision ID: package-import@ubuntu.com-20150415135907-z10fr18evag1ozq3
Tags: 1:2015.1~rc1-0ubuntu1
* New upstream milestone release:
  - debian/control: Update dependencies. 
  - debian/patches/disable-udev-tests.patch: Dropped no longer needed.
  - debian/patches/fixup-driver-test-execution.patch: Dropped no longer needed.
  - debian/patches/skip-iptest.patch: Skip failing test
  - debian/neutron-plugin-openvswitch-agent.install: Added neutron-ovsvapp-agent binary.
  - debian/neutron-plugin-cisco.install: Added neutron-cisco-apic-service-agent and 
    neutron-cisco-apic-host-agent

Show diffs side-by-side

added added

removed removed

Lines of Context:
neutron/tests/unit/db/metering/test_metering_db.py
 
1
# Copyright (C) 2013 eNovance SAS <licensing@enovance.com>
 
2
#
 
3
# Licensed under the Apache License, Version 2.0 (the "License"); you may
 
4
# not use this file except in compliance with the License. You may obtain
 
5
# 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, WITHOUT
 
11
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 
12
# License for the specific language governing permissions and limitations
 
13
# under the License.
 
14
 
 
15
import contextlib
 
16
 
 
17
import webob.exc
 
18
 
 
19
from neutron.api import extensions
 
20
from neutron.common import config
 
21
from neutron.common import constants as n_consts
 
22
from neutron import context
 
23
import neutron.extensions
 
24
from neutron.extensions import metering
 
25
from neutron.plugins.common import constants
 
26
from neutron.services.metering import metering_plugin
 
27
from neutron.tests.unit.db import test_db_base_plugin_v2
 
28
 
 
29
DB_METERING_PLUGIN_KLASS = (
 
30
    "neutron.services.metering."
 
31
    "metering_plugin.MeteringPlugin"
 
32
)
 
33
 
 
34
extensions_path = ':'.join(neutron.extensions.__path__)
 
35
 
 
36
 
 
37
class MeteringPluginDbTestCaseMixin(object):
 
38
    def _create_metering_label(self, fmt, name, description, **kwargs):
 
39
        data = {'metering_label': {'name': name,
 
40
                                   'tenant_id': kwargs.get('tenant_id',
 
41
                                                           'test-tenant'),
 
42
                                   'shared': kwargs.get('shared', False),
 
43
                                   'description': description}}
 
44
        req = self.new_create_request('metering-labels', data,
 
45
                                      fmt)
 
46
 
 
47
        if kwargs.get('set_context') and 'tenant_id' in kwargs:
 
48
            # create a specific auth context for this request
 
49
            req.environ['neutron.context'] = (
 
50
                context.Context('', kwargs['tenant_id'],
 
51
                                is_admin=kwargs.get('is_admin', True)))
 
52
 
 
53
        return req.get_response(self.ext_api)
 
54
 
 
55
    def _make_metering_label(self, fmt, name, description, **kwargs):
 
56
        res = self._create_metering_label(fmt, name, description, **kwargs)
 
57
        if res.status_int >= 400:
 
58
            raise webob.exc.HTTPClientError(code=res.status_int)
 
59
        return self.deserialize(fmt, res)
 
60
 
 
61
    def _create_metering_label_rule(self, fmt, metering_label_id, direction,
 
62
                                    remote_ip_prefix, excluded, **kwargs):
 
63
        data = {'metering_label_rule':
 
64
                {'metering_label_id': metering_label_id,
 
65
                 'tenant_id': kwargs.get('tenant_id', 'test-tenant'),
 
66
                 'direction': direction,
 
67
                 'excluded': excluded,
 
68
                 'remote_ip_prefix': remote_ip_prefix}}
 
69
        req = self.new_create_request('metering-label-rules',
 
70
                                      data, fmt)
 
71
 
 
72
        if kwargs.get('set_context') and 'tenant_id' in kwargs:
 
73
            # create a specific auth context for this request
 
74
            req.environ['neutron.context'] = (
 
75
                context.Context('', kwargs['tenant_id']))
 
76
 
 
77
        return req.get_response(self.ext_api)
 
78
 
 
79
    def _make_metering_label_rule(self, fmt, metering_label_id, direction,
 
80
                                  remote_ip_prefix, excluded, **kwargs):
 
81
        res = self._create_metering_label_rule(fmt, metering_label_id,
 
82
                                               direction, remote_ip_prefix,
 
83
                                               excluded, **kwargs)
 
84
        if res.status_int >= 400:
 
85
            raise webob.exc.HTTPClientError(code=res.status_int)
 
86
        return self.deserialize(fmt, res)
 
87
 
 
88
    @contextlib.contextmanager
 
89
    def metering_label(self, name='label', description='desc',
 
90
                       fmt=None, **kwargs):
 
91
        if not fmt:
 
92
            fmt = self.fmt
 
93
        metering_label = self._make_metering_label(fmt, name,
 
94
                                                   description, **kwargs)
 
95
        yield metering_label
 
96
 
 
97
    @contextlib.contextmanager
 
98
    def metering_label_rule(self, metering_label_id=None, direction='ingress',
 
99
                            remote_ip_prefix='10.0.0.0/24',
 
100
                            excluded='false', fmt=None):
 
101
        if not fmt:
 
102
            fmt = self.fmt
 
103
        metering_label_rule = self._make_metering_label_rule(fmt,
 
104
                                                             metering_label_id,
 
105
                                                             direction,
 
106
                                                             remote_ip_prefix,
 
107
                                                             excluded)
 
108
        yield metering_label_rule
 
109
 
 
110
 
 
111
class MeteringPluginDbTestCase(
 
112
        test_db_base_plugin_v2.NeutronDbPluginV2TestCase,
 
113
        MeteringPluginDbTestCaseMixin):
 
114
    fmt = 'json'
 
115
 
 
116
    resource_prefix_map = dict(
 
117
        (k.replace('_', '-'), constants.COMMON_PREFIXES[constants.METERING])
 
118
        for k in metering.RESOURCE_ATTRIBUTE_MAP.keys()
 
119
    )
 
120
 
 
121
    def setUp(self, plugin=None):
 
122
        service_plugins = {'metering_plugin_name': DB_METERING_PLUGIN_KLASS}
 
123
 
 
124
        super(MeteringPluginDbTestCase, self).setUp(
 
125
            plugin=plugin,
 
126
            service_plugins=service_plugins
 
127
        )
 
128
 
 
129
        self.plugin = metering_plugin.MeteringPlugin()
 
130
        ext_mgr = extensions.PluginAwareExtensionManager(
 
131
            extensions_path,
 
132
            {constants.METERING: self.plugin}
 
133
        )
 
134
        app = config.load_paste_app('extensions_test_app')
 
135
        self.ext_api = extensions.ExtensionMiddleware(app, ext_mgr=ext_mgr)
 
136
 
 
137
 
 
138
class TestMetering(MeteringPluginDbTestCase):
 
139
    def test_create_metering_label(self):
 
140
        name = 'my label'
 
141
        description = 'my metering label'
 
142
        keys = [('name', name,), ('description', description)]
 
143
        with self.metering_label(name, description) as metering_label:
 
144
            for k, v, in keys:
 
145
                self.assertEqual(metering_label['metering_label'][k], v)
 
146
 
 
147
    def test_create_metering_label_shared(self):
 
148
        name = 'my label'
 
149
        description = 'my metering label'
 
150
        shared = True
 
151
        keys = [('name', name,), ('description', description),
 
152
                ('shared', shared)]
 
153
        with self.metering_label(name, description,
 
154
                                 shared=shared) as metering_label:
 
155
            for k, v, in keys:
 
156
                self.assertEqual(metering_label['metering_label'][k], v)
 
157
 
 
158
    def test_delete_metering_label(self):
 
159
        name = 'my label'
 
160
        description = 'my metering label'
 
161
 
 
162
        with self.metering_label(name, description) as metering_label:
 
163
            metering_label_id = metering_label['metering_label']['id']
 
164
            self._delete('metering-labels', metering_label_id, 204)
 
165
 
 
166
    def test_list_metering_label(self):
 
167
        name = 'my label'
 
168
        description = 'my metering label'
 
169
 
 
170
        with contextlib.nested(
 
171
                self.metering_label(name, description),
 
172
                self.metering_label(name, description)) as metering_label:
 
173
 
 
174
            self._test_list_resources('metering-label', metering_label)
 
175
 
 
176
    def test_create_metering_label_rule(self):
 
177
        name = 'my label'
 
178
        description = 'my metering label'
 
179
 
 
180
        with self.metering_label(name, description) as metering_label:
 
181
            metering_label_id = metering_label['metering_label']['id']
 
182
 
 
183
            direction = 'egress'
 
184
            remote_ip_prefix = '192.168.0.0/24'
 
185
            excluded = True
 
186
 
 
187
            keys = [('metering_label_id', metering_label_id),
 
188
                    ('direction', direction),
 
189
                    ('excluded', excluded),
 
190
                    ('remote_ip_prefix', remote_ip_prefix)]
 
191
            with self.metering_label_rule(metering_label_id,
 
192
                                          direction,
 
193
                                          remote_ip_prefix,
 
194
                                          excluded) as label_rule:
 
195
                for k, v, in keys:
 
196
                    self.assertEqual(label_rule['metering_label_rule'][k], v)
 
197
 
 
198
    def test_delete_metering_label_rule(self):
 
199
        name = 'my label'
 
200
        description = 'my metering label'
 
201
 
 
202
        with self.metering_label(name, description) as metering_label:
 
203
            metering_label_id = metering_label['metering_label']['id']
 
204
 
 
205
            direction = 'egress'
 
206
            remote_ip_prefix = '192.168.0.0/24'
 
207
            excluded = True
 
208
 
 
209
            with self.metering_label_rule(metering_label_id,
 
210
                                          direction,
 
211
                                          remote_ip_prefix,
 
212
                                          excluded) as label_rule:
 
213
                rule_id = label_rule['metering_label_rule']['id']
 
214
                self._delete('metering-label-rules', rule_id, 204)
 
215
 
 
216
    def test_list_metering_label_rule(self):
 
217
        name = 'my label'
 
218
        description = 'my metering label'
 
219
 
 
220
        with self.metering_label(name, description) as metering_label:
 
221
            metering_label_id = metering_label['metering_label']['id']
 
222
 
 
223
            direction = 'egress'
 
224
            remote_ip_prefix = '192.168.0.0/24'
 
225
            excluded = True
 
226
 
 
227
            with contextlib.nested(
 
228
                self.metering_label_rule(metering_label_id,
 
229
                                         direction,
 
230
                                         remote_ip_prefix,
 
231
                                         excluded),
 
232
                self.metering_label_rule(metering_label_id,
 
233
                                         'ingress',
 
234
                                         remote_ip_prefix,
 
235
                                         excluded)) as metering_label_rule:
 
236
 
 
237
                self._test_list_resources('metering-label-rule',
 
238
                                          metering_label_rule)
 
239
 
 
240
    def test_create_metering_label_rules(self):
 
241
        name = 'my label'
 
242
        description = 'my metering label'
 
243
 
 
244
        with self.metering_label(name, description) as metering_label:
 
245
            metering_label_id = metering_label['metering_label']['id']
 
246
 
 
247
            direction = 'egress'
 
248
            remote_ip_prefix = '192.168.0.0/24'
 
249
            excluded = True
 
250
 
 
251
            with contextlib.nested(
 
252
                self.metering_label_rule(metering_label_id,
 
253
                                         direction,
 
254
                                         remote_ip_prefix,
 
255
                                         excluded),
 
256
                self.metering_label_rule(metering_label_id,
 
257
                                         direction,
 
258
                                         n_consts.IPv4_ANY,
 
259
                                         False)) as metering_label_rule:
 
260
 
 
261
                self._test_list_resources('metering-label-rule',
 
262
                                          metering_label_rule)
 
263
 
 
264
    def test_create_overlap_metering_label_rules(self):
 
265
        name = 'my label'
 
266
        description = 'my metering label'
 
267
 
 
268
        with self.metering_label(name, description) as metering_label:
 
269
            metering_label_id = metering_label['metering_label']['id']
 
270
 
 
271
            direction = 'egress'
 
272
            remote_ip_prefix1 = '192.168.0.0/24'
 
273
            remote_ip_prefix2 = '192.168.0.0/16'
 
274
            excluded = True
 
275
 
 
276
            with self.metering_label_rule(metering_label_id,
 
277
                                          direction,
 
278
                                          remote_ip_prefix1,
 
279
                                          excluded):
 
280
                res = self._create_metering_label_rule(self.fmt,
 
281
                                                       metering_label_id,
 
282
                                                       direction,
 
283
                                                       remote_ip_prefix2,
 
284
                                                       excluded)
 
285
                self.assertEqual(webob.exc.HTTPConflict.code, res.status_int)
 
286
 
 
287
    def test_create_metering_label_rule_two_labels(self):
 
288
        name1 = 'my label 1'
 
289
        name2 = 'my label 2'
 
290
        description = 'my metering label'
 
291
 
 
292
        with self.metering_label(name1, description) as metering_label1:
 
293
            metering_label_id1 = metering_label1['metering_label']['id']
 
294
 
 
295
            with self.metering_label(name2, description) as metering_label2:
 
296
                metering_label_id2 = metering_label2['metering_label']['id']
 
297
 
 
298
                direction = 'egress'
 
299
                remote_ip_prefix = '192.168.0.0/24'
 
300
                excluded = True
 
301
 
 
302
                with contextlib.nested(
 
303
                    self.metering_label_rule(metering_label_id1,
 
304
                                             direction,
 
305
                                             remote_ip_prefix,
 
306
                                             excluded),
 
307
                    self.metering_label_rule(metering_label_id2,
 
308
                                             direction,
 
309
                                             remote_ip_prefix,
 
310
                                             excluded)) as metering_label_rule:
 
311
 
 
312
                    self._test_list_resources('metering-label-rule',
 
313
                                              metering_label_rule)