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
|
# Copyright 2016 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.common import utils
from tempest.lib.common.utils import data_utils
from tempest.lib.common.utils import test_utils
from tempest.lib import decorators
from tempest.lib import exceptions
from neutron_tempest_plugin.api import base
from neutron_lib import constants as lib_constants
# 3 IP addresses are taken from every total for IPv4 these are reserved
DEFAULT_IP4_RESERVED = 3
# 2 IP addresses are taken from every total for IPv6 these are reserved
# I assume the reason for having one less than IPv4 is it does not have
# broadcast address
DEFAULT_IP6_RESERVED = 2
DELETE_TIMEOUT = 10
DELETE_SLEEP = 2
class NetworksIpAvailabilityTest(base.BaseAdminNetworkTest):
"""Tests Networks IP Availability
Tests the following operations in the Neutron API using the REST client for
Neutron:
test total and used ips for net create
test total and ips for net after subnet create
test total and used ips for net after subnet and port create
"""
@classmethod
@utils.requires_ext(extension="network-ip-availability", service="network")
def skip_checks(cls):
super(NetworksIpAvailabilityTest, cls).skip_checks()
@staticmethod
def _get_availability(network, net_availability):
if 'network_ip_availabilities' in net_availability:
for availability in net_availability['network_ip_availabilities']:
if availability['network_id'] == network['id']:
return availability
raise exceptions.TempestException('Network IP Availability not '
'found')
else:
return net_availability['network_ip_availability']
def _get_used_ips(self, network, net_availability):
availability = self._get_availability(network, net_availability)
return availability and availability['used_ips']
def _assert_total_and_used_ips(self, expected_used, expected_total,
network, net_availability):
availability = self._get_availability(network, net_availability)
self.assertEqual(expected_total, availability['total_ips'])
self.assertEqual(expected_used, availability['used_ips'])
def calc_total_ips(prefix, ip_version):
# will calculate total ips after removing reserved.
if ip_version == lib_constants.IP_VERSION_4:
total_ips = 2 ** (lib_constants.IPv4_BITS -
prefix) - DEFAULT_IP4_RESERVED
elif ip_version == lib_constants.IP_VERSION_6:
total_ips = 2 ** (lib_constants.IPv6_BITS -
prefix) - DEFAULT_IP6_RESERVED
return total_ips
class NetworksIpAvailabilityIPv4Test(NetworksIpAvailabilityTest):
def setUp(self):
super(NetworksIpAvailabilityIPv4Test, self).setUp()
net_name = data_utils.rand_name('network')
self.network = self.create_network(network_name=net_name)
@decorators.idempotent_id('0f33cc8c-1bf6-47d1-9ce1-010618240599')
def test_list_ip_availability_before_subnet(self):
net_availability = self.admin_client.list_network_ip_availabilities()
self._assert_total_and_used_ips(0, 0, self.network, net_availability)
@decorators.idempotent_id('3aecd3b2-16ed-4b87-a54a-91d7b3c2986b')
def test_list_ip_availability_after_subnet_and_ports(self):
subnet = self.create_subnet(self.network, enable_dhcp=False)
prefix = netaddr.IPNetwork(subnet['cidr']).prefixlen
body = self.admin_client.list_network_ip_availabilities()
used_ips_before_port_create = self._get_used_ips(self.network, body)
self.create_port(self.network)
net_availability = self.admin_client.list_network_ip_availabilities()
self._assert_total_and_used_ips(
used_ips_before_port_create + 1,
calc_total_ips(prefix, self._ip_version),
self.network, net_availability)
@decorators.idempotent_id('9f11254d-757b-492e-b14b-f52144e4ee7b')
def test_list_ip_availability_after_port_delete(self):
self.create_subnet(self.network, enable_dhcp=False)
port = self.create_port(self.network)
net_availability = self.admin_client.list_network_ip_availabilities()
used_ips = self._get_used_ips(self.network, net_availability)
self.client.delete_port(port['id'])
def is_count_ip_availability_valid():
availabilities = self.admin_client.list_network_ip_availabilities()
used_ips_after_port_delete = self._get_used_ips(self.network,
availabilities)
return used_ips - 1 == used_ips_after_port_delete
self.assertTrue(
test_utils.call_until_true(
is_count_ip_availability_valid, DELETE_TIMEOUT, DELETE_SLEEP),
msg="IP address did not become available after port delete")
@decorators.idempotent_id('da1fbed5-b4a9-45b3-bdcb-b1660710d565')
def test_show_ip_availability_after_subnet_and_ports_create(self):
net_availability = self.admin_client.show_network_ip_availability(
self.network['id'])
self._assert_total_and_used_ips(0, 0, self.network, net_availability)
subnet = self.create_subnet(self.network, enable_dhcp=False)
prefix = netaddr.IPNetwork(subnet['cidr']).prefixlen
net_availability = self.admin_client.show_network_ip_availability(
self.network['id'])
used_ips_before_port_create = self._get_used_ips(self.network,
net_availability)
self.create_port(self.network)
net_availability = self.admin_client.show_network_ip_availability(
self.network['id'])
self._assert_total_and_used_ips(
used_ips_before_port_create + 1,
calc_total_ips(prefix, self._ip_version),
self.network,
net_availability)
@decorators.idempotent_id('a4d1e291-c152-4d62-9316-8c9bf1c6aee2')
def test_show_ip_availability_after_port_delete(self):
self.create_subnet(self.network, enable_dhcp=False)
port = self.create_port(self.network)
net_availability = self.admin_client.show_network_ip_availability(
self.network['id'])
used_ips = self._get_used_ips(self.network, net_availability)
self.client.delete_port(port['id'])
def is_count_ip_availability_valid():
availabilities = self.admin_client.show_network_ip_availability(
self.network['id'])
used_ips_after_port_delete = self._get_used_ips(self.network,
availabilities)
return used_ips - 1 == used_ips_after_port_delete
self.assertTrue(
test_utils.call_until_true(
is_count_ip_availability_valid, DELETE_TIMEOUT, DELETE_SLEEP),
msg="IP address did not become available after port delete")
class NetworksIpAvailabilityIPv6Test(NetworksIpAvailabilityIPv4Test):
_ip_version = lib_constants.IP_VERSION_6
@decorators.idempotent_id('0d5a03f2-fdb7-4ec3-b746-734c51d74b69')
def test_list_ipv6_ip_availability_after_subnet_and_ports(self):
subnet = self.create_subnet(self.network, ip_version=self._ip_version,
enable_dhcp=False)
prefix = netaddr.IPNetwork(subnet['cidr']).prefixlen
body = self.admin_client.list_network_ip_availabilities()
used_ips_before_port_create = self._get_used_ips(self.network, body)
self.create_port(self.network)
net_availability = self.admin_client.list_network_ip_availabilities()
self._assert_total_and_used_ips(
used_ips_before_port_create + 1,
calc_total_ips(prefix, self._ip_version),
self.network, net_availability)
|
