← Back to branch summary

~ubuntu-branches/ubuntu/vivid/neutron/vivid-updates

« back to all changes in this revision

Viewing changes to neutron/tests/tempest/api/network/base.py

  • Committer: Package Import Robot
  • Author(s): James Page
  • Date: 2015-03-30 11:17:19 UTC
  • mfrom: (1.1.21)
  • Revision ID: package-import@ubuntu.com-20150330111719-h0gx7233p4jkkgfh
Tags: 1:2015.1~b3-0ubuntu1
https://launchpad.net/bugs/1263539
* New upstream milestone release:
  - d/control: Align version requirements with upstream.
  - d/control: Add new dependency on oslo-log.
  - d/p/*: Rebase.
  - d/control,d/neutron-plugin-hyperv*: Dropped, decomposed into
    separate project upstream.
  - d/control,d/neutron-plugin-openflow*: Dropped, decomposed into
    separate project upstream.
  - d/neutron-common.install: Add neutron-rootwrap-daemon and 
    neutron-keepalived-state-change binaries.
  - d/rules: Ignore neutron-hyperv-agent when installing; only for Windows.
  - d/neutron-plugin-cisco.install: Drop neutron-cisco-cfg-agent as
    decomposed into separate project upstream.
  - d/neutron-plugin-vmware.install: Drop neutron-check-nsx-config and
    neutron-nsx-manage as decomposed into separate project upstream.
  - d/control: Add dependency on python-neutron-fwaas to neutron-l3-agent.
* d/pydist-overrides: Add overrides for oslo packages.
* d/control: Fixup type in package description (LP: #1263539).
* d/p/fixup-driver-test-execution.patch: Cherry pick fix from upstream VCS
  to support unit test exection in out-of-tree vendor drivers.
* d/neutron-common.postinst: Allow general access to /etc/neutron but limit
  access to root/neutron to /etc/neutron/neutron.conf to support execution
  of unit tests in decomposed vendor drivers.
* d/control: Add dependency on python-neutron-fwaas to neutron-l3-agent
  package.

Show diffs side-by-side

added added

removed removed

Lines of Context:
neutron/tests/tempest/api/network/base.py
 
1
# Copyright 2012 OpenStack Foundation
 
2
# All Rights Reserved.
 
3
#
 
4
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
 
5
#    not use this file except in compliance with the License. You may obtain
 
6
#    a copy of the License at
 
7
#
 
8
#         http://www.apache.org/licenses/LICENSE-2.0
 
9
#
 
10
#    Unless required by applicable law or agreed to in writing, software
 
11
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 
12
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 
13
#    License for the specific language governing permissions and limitations
 
14
#    under the License.
 
15
 
 
16
import netaddr
 
17
from oslo_log import log as logging
 
18
from tempest_lib.common.utils import data_utils
 
19
from tempest_lib import exceptions as lib_exc
 
20
 
 
21
from neutron.tests.api.contrib import clients
 
22
from neutron.tests.tempest import config
 
23
from neutron.tests.tempest import exceptions
 
24
import neutron.tests.tempest.test
 
25
 
 
26
CONF = config.CONF
 
27
 
 
28
LOG = logging.getLogger(__name__)
 
29
 
 
30
 
 
31
class BaseNetworkTest(neutron.tests.tempest.test.BaseTestCase):
 
32
 
 
33
    """
 
34
    Base class for the Neutron tests that use the Tempest Neutron REST client
 
35
 
 
36
    Per the Neutron API Guide, API v1.x was removed from the source code tree
 
37
    (docs.openstack.org/api/openstack-network/2.0/content/Overview-d1e71.html)
 
38
    Therefore, v2.x of the Neutron API is assumed. It is also assumed that the
 
39
    following options are defined in the [network] section of etc/tempest.conf:
 
40
 
 
41
        tenant_network_cidr with a block of cidr's from which smaller blocks
 
42
        can be allocated for tenant networks
 
43
 
 
44
        tenant_network_mask_bits with the mask bits to be used to partition the
 
45
        block defined by tenant-network_cidr
 
46
 
 
47
    Finally, it is assumed that the following option is defined in the
 
48
    [service_available] section of etc/tempest.conf
 
49
 
 
50
        neutron as True
 
51
    """
 
52
 
 
53
    force_tenant_isolation = False
 
54
 
 
55
    # Default to ipv4.
 
56
    _ip_version = 4
 
57
 
 
58
    @classmethod
 
59
    def resource_setup(cls):
 
60
        # Create no network resources for these test.
 
61
        cls.set_network_resources()
 
62
        super(BaseNetworkTest, cls).resource_setup()
 
63
        if not CONF.service_available.neutron:
 
64
            raise cls.skipException("Neutron support is required")
 
65
        if cls._ip_version == 6 and not CONF.network_feature_enabled.ipv6:
 
66
            raise cls.skipException("IPv6 Tests are disabled.")
 
67
 
 
68
        os = cls.get_client_manager()
 
69
 
 
70
        cls.network_cfg = CONF.network
 
71
        cls.client = os.network_client
 
72
        cls.networks = []
 
73
        cls.subnets = []
 
74
        cls.ports = []
 
75
        cls.routers = []
 
76
        cls.pools = []
 
77
        cls.vips = []
 
78
        cls.members = []
 
79
        cls.health_monitors = []
 
80
        cls.vpnservices = []
 
81
        cls.ikepolicies = []
 
82
        cls.floating_ips = []
 
83
        cls.metering_labels = []
 
84
        cls.metering_label_rules = []
 
85
        cls.fw_rules = []
 
86
        cls.fw_policies = []
 
87
        cls.ipsecpolicies = []
 
88
        cls.ethertype = "IPv" + str(cls._ip_version)
 
89
 
 
90
    @classmethod
 
91
    def resource_cleanup(cls):
 
92
        if CONF.service_available.neutron:
 
93
            # Clean up ipsec policies
 
94
            for ipsecpolicy in cls.ipsecpolicies:
 
95
                cls._try_delete_resource(cls.client.delete_ipsecpolicy,
 
96
                                         ipsecpolicy['id'])
 
97
            # Clean up firewall policies
 
98
            for fw_policy in cls.fw_policies:
 
99
                cls._try_delete_resource(cls.client.delete_firewall_policy,
 
100
                                         fw_policy['id'])
 
101
            # Clean up firewall rules
 
102
            for fw_rule in cls.fw_rules:
 
103
                cls._try_delete_resource(cls.client.delete_firewall_rule,
 
104
                                         fw_rule['id'])
 
105
            # Clean up ike policies
 
106
            for ikepolicy in cls.ikepolicies:
 
107
                cls._try_delete_resource(cls.client.delete_ikepolicy,
 
108
                                         ikepolicy['id'])
 
109
            # Clean up vpn services
 
110
            for vpnservice in cls.vpnservices:
 
111
                cls._try_delete_resource(cls.client.delete_vpnservice,
 
112
                                         vpnservice['id'])
 
113
            # Clean up floating IPs
 
114
            for floating_ip in cls.floating_ips:
 
115
                cls._try_delete_resource(cls.client.delete_floatingip,
 
116
                                         floating_ip['id'])
 
117
            # Clean up routers
 
118
            for router in cls.routers:
 
119
                cls._try_delete_resource(cls.delete_router,
 
120
                                         router)
 
121
 
 
122
            # Clean up health monitors
 
123
            for health_monitor in cls.health_monitors:
 
124
                cls._try_delete_resource(cls.client.delete_health_monitor,
 
125
                                         health_monitor['id'])
 
126
            # Clean up members
 
127
            for member in cls.members:
 
128
                cls._try_delete_resource(cls.client.delete_member,
 
129
                                         member['id'])
 
130
            # Clean up vips
 
131
            for vip in cls.vips:
 
132
                cls._try_delete_resource(cls.client.delete_vip,
 
133
                                         vip['id'])
 
134
            # Clean up pools
 
135
            for pool in cls.pools:
 
136
                cls._try_delete_resource(cls.client.delete_pool,
 
137
                                         pool['id'])
 
138
            # Clean up metering label rules
 
139
            for metering_label_rule in cls.metering_label_rules:
 
140
                cls._try_delete_resource(
 
141
                    cls.admin_client.delete_metering_label_rule,
 
142
                    metering_label_rule['id'])
 
143
            # Clean up metering labels
 
144
            for metering_label in cls.metering_labels:
 
145
                cls._try_delete_resource(
 
146
                    cls.admin_client.delete_metering_label,
 
147
                    metering_label['id'])
 
148
            # Clean up ports
 
149
            for port in cls.ports:
 
150
                cls._try_delete_resource(cls.client.delete_port,
 
151
                                         port['id'])
 
152
            # Clean up subnets
 
153
            for subnet in cls.subnets:
 
154
                cls._try_delete_resource(cls.client.delete_subnet,
 
155
                                         subnet['id'])
 
156
            # Clean up networks
 
157
            for network in cls.networks:
 
158
                cls._try_delete_resource(cls.client.delete_network,
 
159
                                         network['id'])
 
160
            cls.clear_isolated_creds()
 
161
        super(BaseNetworkTest, cls).resource_cleanup()
 
162
 
 
163
    @classmethod
 
164
    def _try_delete_resource(self, delete_callable, *args, **kwargs):
 
165
        """Cleanup resources in case of test-failure
 
166
 
 
167
        Some resources are explicitly deleted by the test.
 
168
        If the test failed to delete a resource, this method will execute
 
169
        the appropriate delete methods. Otherwise, the method ignores NotFound
 
170
        exceptions thrown for resources that were correctly deleted by the
 
171
        test.
 
172
 
 
173
        :param delete_callable: delete method
 
174
        :param args: arguments for delete method
 
175
        :param kwargs: keyword arguments for delete method
 
176
        """
 
177
        try:
 
178
            delete_callable(*args, **kwargs)
 
179
        # if resource is not found, this means it was deleted in the test
 
180
        except lib_exc.NotFound:
 
181
            pass
 
182
 
 
183
    @classmethod
 
184
    def create_network(cls, network_name=None):
 
185
        """Wrapper utility that returns a test network."""
 
186
        network_name = network_name or data_utils.rand_name('test-network-')
 
187
 
 
188
        body = cls.client.create_network(name=network_name)
 
189
        network = body['network']
 
190
        cls.networks.append(network)
 
191
        return network
 
192
 
 
193
    @classmethod
 
194
    def create_subnet(cls, network, gateway='', cidr=None, mask_bits=None,
 
195
                      ip_version=None, client=None, **kwargs):
 
196
        """Wrapper utility that returns a test subnet."""
 
197
 
 
198
        # allow tests to use admin client
 
199
        if not client:
 
200
            client = cls.client
 
201
 
 
202
        # The cidr and mask_bits depend on the ip version.
 
203
        ip_version = ip_version if ip_version is not None else cls._ip_version
 
204
        gateway_not_set = gateway == ''
 
205
        if ip_version == 4:
 
206
            cidr = cidr or netaddr.IPNetwork(CONF.network.tenant_network_cidr)
 
207
            mask_bits = mask_bits or CONF.network.tenant_network_mask_bits
 
208
        elif ip_version == 6:
 
209
            cidr = (
 
210
                cidr or netaddr.IPNetwork(CONF.network.tenant_network_v6_cidr))
 
211
            mask_bits = mask_bits or CONF.network.tenant_network_v6_mask_bits
 
212
        # Find a cidr that is not in use yet and create a subnet with it
 
213
        for subnet_cidr in cidr.subnet(mask_bits):
 
214
            if gateway_not_set:
 
215
                gateway_ip = str(netaddr.IPAddress(subnet_cidr) + 1)
 
216
            else:
 
217
                gateway_ip = gateway
 
218
            try:
 
219
                body = client.create_subnet(
 
220
                    network_id=network['id'],
 
221
                    cidr=str(subnet_cidr),
 
222
                    ip_version=ip_version,
 
223
                    gateway_ip=gateway_ip,
 
224
                    **kwargs)
 
225
                break
 
226
            except lib_exc.BadRequest as e:
 
227
                is_overlapping_cidr = 'overlaps with another subnet' in str(e)
 
228
                if not is_overlapping_cidr:
 
229
                    raise
 
230
        else:
 
231
            message = 'Available CIDR for subnet creation could not be found'
 
232
            raise exceptions.BuildErrorException(message)
 
233
        subnet = body['subnet']
 
234
        cls.subnets.append(subnet)
 
235
        return subnet
 
236
 
 
237
    @classmethod
 
238
    def create_port(cls, network, **kwargs):
 
239
        """Wrapper utility that returns a test port."""
 
240
        body = cls.client.create_port(network_id=network['id'],
 
241
                                      **kwargs)
 
242
        port = body['port']
 
243
        cls.ports.append(port)
 
244
        return port
 
245
 
 
246
    @classmethod
 
247
    def update_port(cls, port, **kwargs):
 
248
        """Wrapper utility that updates a test port."""
 
249
        body = cls.client.update_port(port['id'],
 
250
                                      **kwargs)
 
251
        return body['port']
 
252
 
 
253
    @classmethod
 
254
    def create_router(cls, router_name=None, admin_state_up=False,
 
255
                      external_network_id=None, enable_snat=None):
 
256
        ext_gw_info = {}
 
257
        if external_network_id:
 
258
            ext_gw_info['network_id'] = external_network_id
 
259
        if enable_snat:
 
260
            ext_gw_info['enable_snat'] = enable_snat
 
261
        body = cls.client.create_router(
 
262
            router_name, external_gateway_info=ext_gw_info,
 
263
            admin_state_up=admin_state_up)
 
264
        router = body['router']
 
265
        cls.routers.append(router)
 
266
        return router
 
267
 
 
268
    @classmethod
 
269
    def create_floatingip(cls, external_network_id):
 
270
        """Wrapper utility that returns a test floating IP."""
 
271
        body = cls.client.create_floatingip(
 
272
            floating_network_id=external_network_id)
 
273
        fip = body['floatingip']
 
274
        cls.floating_ips.append(fip)
 
275
        return fip
 
276
 
 
277
    @classmethod
 
278
    def create_pool(cls, name, lb_method, protocol, subnet):
 
279
        """Wrapper utility that returns a test pool."""
 
280
        body = cls.client.create_pool(
 
281
            name=name,
 
282
            lb_method=lb_method,
 
283
            protocol=protocol,
 
284
            subnet_id=subnet['id'])
 
285
        pool = body['pool']
 
286
        cls.pools.append(pool)
 
287
        return pool
 
288
 
 
289
    @classmethod
 
290
    def update_pool(cls, name):
 
291
        """Wrapper utility that returns a test pool."""
 
292
        body = cls.client.update_pool(name=name)
 
293
        pool = body['pool']
 
294
        return pool
 
295
 
 
296
    @classmethod
 
297
    def create_vip(cls, name, protocol, protocol_port, subnet, pool):
 
298
        """Wrapper utility that returns a test vip."""
 
299
        body = cls.client.create_vip(name=name,
 
300
                                     protocol=protocol,
 
301
                                     protocol_port=protocol_port,
 
302
                                     subnet_id=subnet['id'],
 
303
                                     pool_id=pool['id'])
 
304
        vip = body['vip']
 
305
        cls.vips.append(vip)
 
306
        return vip
 
307
 
 
308
    @classmethod
 
309
    def update_vip(cls, name):
 
310
        body = cls.client.update_vip(name=name)
 
311
        vip = body['vip']
 
312
        return vip
 
313
 
 
314
    @classmethod
 
315
    def create_member(cls, protocol_port, pool, ip_version=None):
 
316
        """Wrapper utility that returns a test member."""
 
317
        ip_version = ip_version if ip_version is not None else cls._ip_version
 
318
        member_address = "fd00::abcd" if ip_version == 6 else "10.0.9.46"
 
319
        body = cls.client.create_member(address=member_address,
 
320
                                        protocol_port=protocol_port,
 
321
                                        pool_id=pool['id'])
 
322
        member = body['member']
 
323
        cls.members.append(member)
 
324
        return member
 
325
 
 
326
    @classmethod
 
327
    def update_member(cls, admin_state_up):
 
328
        body = cls.client.update_member(admin_state_up=admin_state_up)
 
329
        member = body['member']
 
330
        return member
 
331
 
 
332
    @classmethod
 
333
    def create_health_monitor(cls, delay, max_retries, Type, timeout):
 
334
        """Wrapper utility that returns a test health monitor."""
 
335
        body = cls.client.create_health_monitor(delay=delay,
 
336
                                                max_retries=max_retries,
 
337
                                                type=Type,
 
338
                                                timeout=timeout)
 
339
        health_monitor = body['health_monitor']
 
340
        cls.health_monitors.append(health_monitor)
 
341
        return health_monitor
 
342
 
 
343
    @classmethod
 
344
    def update_health_monitor(cls, admin_state_up):
 
345
        body = cls.client.update_vip(admin_state_up=admin_state_up)
 
346
        health_monitor = body['health_monitor']
 
347
        return health_monitor
 
348
 
 
349
    @classmethod
 
350
    def create_router_interface(cls, router_id, subnet_id):
 
351
        """Wrapper utility that returns a router interface."""
 
352
        interface = cls.client.add_router_interface_with_subnet_id(
 
353
            router_id, subnet_id)
 
354
        return interface
 
355
 
 
356
    @classmethod
 
357
    def create_vpnservice(cls, subnet_id, router_id):
 
358
        """Wrapper utility that returns a test vpn service."""
 
359
        body = cls.client.create_vpnservice(
 
360
            subnet_id=subnet_id, router_id=router_id, admin_state_up=True,
 
361
            name=data_utils.rand_name("vpnservice-"))
 
362
        vpnservice = body['vpnservice']
 
363
        cls.vpnservices.append(vpnservice)
 
364
        return vpnservice
 
365
 
 
366
    @classmethod
 
367
    def create_ikepolicy(cls, name):
 
368
        """Wrapper utility that returns a test ike policy."""
 
369
        body = cls.client.create_ikepolicy(name=name)
 
370
        ikepolicy = body['ikepolicy']
 
371
        cls.ikepolicies.append(ikepolicy)
 
372
        return ikepolicy
 
373
 
 
374
    @classmethod
 
375
    def create_firewall_rule(cls, action, protocol):
 
376
        """Wrapper utility that returns a test firewall rule."""
 
377
        body = cls.client.create_firewall_rule(
 
378
            name=data_utils.rand_name("fw-rule"),
 
379
            action=action,
 
380
            protocol=protocol)
 
381
        fw_rule = body['firewall_rule']
 
382
        cls.fw_rules.append(fw_rule)
 
383
        return fw_rule
 
384
 
 
385
    @classmethod
 
386
    def create_firewall_policy(cls):
 
387
        """Wrapper utility that returns a test firewall policy."""
 
388
        body = cls.client.create_firewall_policy(
 
389
            name=data_utils.rand_name("fw-policy"))
 
390
        fw_policy = body['firewall_policy']
 
391
        cls.fw_policies.append(fw_policy)
 
392
        return fw_policy
 
393
 
 
394
    @classmethod
 
395
    def delete_router(cls, router):
 
396
        body = cls.client.list_router_interfaces(router['id'])
 
397
        interfaces = body['ports']
 
398
        for i in interfaces:
 
399
            try:
 
400
                cls.client.remove_router_interface_with_subnet_id(
 
401
                    router['id'], i['fixed_ips'][0]['subnet_id'])
 
402
            except lib_exc.NotFound:
 
403
                pass
 
404
        cls.client.delete_router(router['id'])
 
405
 
 
406
    @classmethod
 
407
    def create_ipsecpolicy(cls, name):
 
408
        """Wrapper utility that returns a test ipsec policy."""
 
409
        body = cls.client.create_ipsecpolicy(name=name)
 
410
        ipsecpolicy = body['ipsecpolicy']
 
411
        cls.ipsecpolicies.append(ipsecpolicy)
 
412
        return ipsecpolicy
 
413
 
 
414
 
 
415
class BaseAdminNetworkTest(BaseNetworkTest):
 
416
 
 
417
    @classmethod
 
418
    def resource_setup(cls):
 
419
        super(BaseAdminNetworkTest, cls).resource_setup()
 
420
 
 
421
        try:
 
422
            creds = cls.isolated_creds.get_admin_creds()
 
423
            cls.os_adm = clients.Manager(credentials=creds)
 
424
        except NotImplementedError:
 
425
            msg = ("Missing Administrative Network API credentials "
 
426
                   "in configuration.")
 
427
            raise cls.skipException(msg)
 
428
        cls.admin_client = cls.os_adm.network_client
 
429
 
 
430
    @classmethod
 
431
    def create_metering_label(cls, name, description):
 
432
        """Wrapper utility that returns a test metering label."""
 
433
        body = cls.admin_client.create_metering_label(
 
434
            description=description,
 
435
            name=data_utils.rand_name("metering-label"))
 
436
        metering_label = body['metering_label']
 
437
        cls.metering_labels.append(metering_label)
 
438
        return metering_label
 
439
 
 
440
    @classmethod
 
441
    def create_metering_label_rule(cls, remote_ip_prefix, direction,
 
442
                                   metering_label_id):
 
443
        """Wrapper utility that returns a test metering label rule."""
 
444
        body = cls.admin_client.create_metering_label_rule(
 
445
            remote_ip_prefix=remote_ip_prefix, direction=direction,
 
446
            metering_label_id=metering_label_id)
 
447
        metering_label_rule = body['metering_label_rule']
 
448
        cls.metering_label_rules.append(metering_label_rule)
 
449
        return metering_label_rule