1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
|
# Copyright 2012 OpenStack Foundation
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import netaddr
from tempest import config
from tempest import exceptions
from tempest.lib.common.utils import data_utils
from tempest.lib.common.utils import test_utils
from tempest.lib import exceptions as lib_exc
import tempest.test
CONF = config.CONF
class BaseNetworkTest(tempest.test.BaseTestCase):
"""Base class for the Neutron tests.
Per the Neutron API Guide, API v1.x was removed from the source code tree
(docs.openstack.org/api/openstack-network/2.0/content/Overview-d1e71.html)
Therefore, v2.x of the Neutron API is assumed. It is also assumed that the
following options are defined in the [network] section of etc/tempest.conf:
project_network_cidr with a block of cidr's from which smaller blocks
can be allocated for project networks
project_network_mask_bits with the mask bits to be used to partition
the block defined by project-network_cidr
Finally, it is assumed that the following option is defined in the
[service_available] section of etc/tempest.conf
neutron as True
"""
force_tenant_isolation = False
credentials = ['primary']
# Default to ipv4.
_ip_version = 4
@classmethod
def skip_checks(cls):
super(BaseNetworkTest, cls).skip_checks()
if not CONF.service_available.neutron:
raise cls.skipException("Neutron support is required")
if cls._ip_version == 6 and not CONF.network_feature_enabled.ipv6:
raise cls.skipException("IPv6 Tests are disabled.")
@classmethod
def setup_credentials(cls):
# Create no network resources for these test.
cls.set_network_resources()
super(BaseNetworkTest, cls).setup_credentials()
@classmethod
def setup_clients(cls):
super(BaseNetworkTest, cls).setup_clients()
cls.agents_client = cls.os_primary.network_agents_client
cls.network_extensions_client =\
cls.os_primary.network_extensions_client
cls.networks_client = cls.os_primary.networks_client
cls.routers_client = cls.os_primary.routers_client
cls.subnetpools_client = cls.os_primary.subnetpools_client
cls.subnets_client = cls.os_primary.subnets_client
cls.ports_client = cls.os_primary.ports_client
cls.floating_ips_port_forwarding_client =\
cls.os_primary.floating_ips_port_forwarding_client
cls.quotas_client = cls.os_primary.network_quotas_client
cls.floating_ips_client = cls.os_primary.floating_ips_client
cls.security_groups_client = cls.os_primary.security_groups_client
cls.security_group_rules_client = (
cls.os_primary.security_group_rules_client)
cls.network_versions_client = cls.os_primary.network_versions_client
cls.service_providers_client = cls.os_primary.service_providers_client
cls.tags_client = cls.os_primary.tags_client
cls.log_resource_client = cls.os_primary.log_resource_client
cls.loggable_resource_client = cls.os_primary.loggable_resource_client
@classmethod
def resource_setup(cls):
super(BaseNetworkTest, cls).resource_setup()
cls.ethertype = "IPv" + str(cls._ip_version)
if cls._ip_version == 4:
cls.cidr = netaddr.IPNetwork(CONF.network.project_network_cidr)
cls.mask_bits = CONF.network.project_network_mask_bits
elif cls._ip_version == 6:
cls.cidr = netaddr.IPNetwork(CONF.network.project_network_v6_cidr)
cls.mask_bits = CONF.network.project_network_v6_mask_bits
@classmethod
def create_network(cls, network_name=None, **kwargs):
"""Wrapper utility that returns a test network."""
network_name = network_name or data_utils.rand_name(
cls.__name__ + '-test-network')
body = cls.networks_client.create_network(name=network_name, **kwargs)
network = body['network']
cls.addClassResourceCleanup(test_utils.call_and_ignore_notfound_exc,
cls.networks_client.delete_network,
network['id'])
return network
@classmethod
def create_subnet(cls, network, gateway='', cidr=None, mask_bits=None,
ip_version=None, client=None, **kwargs):
"""Wrapper utility that returns a test subnet."""
# allow tests to use admin client
if not client:
client = cls.subnets_client
# The cidr and mask_bits depend on the ip version.
ip_version = ip_version if ip_version is not None else cls._ip_version
gateway_not_set = gateway == ''
if ip_version == 4:
cidr = cidr or netaddr.IPNetwork(CONF.network.project_network_cidr)
mask_bits = mask_bits or CONF.network.project_network_mask_bits
elif ip_version == 6:
cidr = (cidr or
netaddr.IPNetwork(CONF.network.project_network_v6_cidr))
mask_bits = mask_bits or CONF.network.project_network_v6_mask_bits
# Find a cidr that is not in use yet and create a subnet with it
for subnet_cidr in cidr.subnet(mask_bits):
if gateway_not_set:
gateway_ip = str(netaddr.IPAddress(subnet_cidr) + 1)
else:
gateway_ip = gateway
try:
body = client.create_subnet(
network_id=network['id'],
cidr=str(subnet_cidr),
ip_version=ip_version,
gateway_ip=gateway_ip,
**kwargs)
break
except lib_exc.BadRequest as e:
is_overlapping_cidr = 'overlaps with another subnet' in str(e)
if not is_overlapping_cidr:
raise
else:
message = 'Available CIDR for subnet creation could not be found'
raise exceptions.BuildErrorException(message)
subnet = body['subnet']
cls.addClassResourceCleanup(test_utils.call_and_ignore_notfound_exc,
cls.subnets_client.delete_subnet,
subnet['id'])
return subnet
@classmethod
def create_port(cls, network, **kwargs):
if 'name' not in kwargs:
kwargs['name'] = data_utils.rand_name(cls.__name__)
"""Wrapper utility that returns a test port."""
body = cls.ports_client.create_port(network_id=network['id'],
**kwargs)
port = body['port']
cls.addClassResourceCleanup(test_utils.call_and_ignore_notfound_exc,
cls.ports_client.delete_port, port['id'])
return port
@classmethod
def update_port(cls, port, **kwargs):
"""Wrapper utility that updates a test port."""
body = cls.ports_client.update_port(port['id'],
**kwargs)
return body['port']
@classmethod
def create_router(cls, router_name=None, admin_state_up=False,
external_network_id=None, enable_snat=None,
**kwargs):
router_name = router_name or data_utils.rand_name(
cls.__name__ + "-router")
ext_gw_info = {}
if external_network_id:
ext_gw_info['network_id'] = external_network_id
if enable_snat is not None:
ext_gw_info['enable_snat'] = enable_snat
body = cls.routers_client.create_router(
name=router_name, external_gateway_info=ext_gw_info,
admin_state_up=admin_state_up, **kwargs)
router = body['router']
cls.addClassResourceCleanup(test_utils.call_and_ignore_notfound_exc,
cls.delete_router, router)
return router
@classmethod
def create_floatingip(cls, external_network_id):
"""Wrapper utility that returns a test floating IP."""
body = cls.floating_ips_client.create_floatingip(
floating_network_id=external_network_id)
fip = body['floatingip']
cls.addClassResourceCleanup(test_utils.call_and_ignore_notfound_exc,
cls.floating_ips_client.delete_floatingip,
fip['id'])
return fip
@classmethod
def create_router_interface(cls, router_id, subnet_id):
"""Wrapper utility that returns a router interface."""
interface = cls.routers_client.add_router_interface(
router_id, subnet_id=subnet_id)
return interface
@classmethod
def delete_router(cls, router):
body = cls.ports_client.list_ports(device_id=router['id'])
interfaces = body['ports']
for i in interfaces:
test_utils.call_and_ignore_notfound_exc(
cls.routers_client.remove_router_interface, router['id'],
subnet_id=i['fixed_ips'][0]['subnet_id'])
cls.routers_client.delete_router(router['id'])
class BaseAdminNetworkTest(BaseNetworkTest):
credentials = ['primary', 'admin']
@classmethod
def setup_clients(cls):
super(BaseAdminNetworkTest, cls).setup_clients()
cls.admin_agents_client = cls.os_admin.network_agents_client
cls.admin_networks_client = cls.os_admin.networks_client
cls.admin_routers_client = cls.os_admin.routers_client
cls.admin_subnets_client = cls.os_admin.subnets_client
cls.admin_ports_client = cls.os_admin.ports_client
cls.admin_quotas_client = cls.os_admin.network_quotas_client
cls.admin_floating_ips_client = cls.os_admin.floating_ips_client
cls.admin_metering_labels_client = cls.os_admin.metering_labels_client
cls.admin_metering_label_rules_client = (
cls.os_admin.metering_label_rules_client)
|
