Use service_ports_number option to calculate number of used ips
After subnet is created, ips can be assigned also to neutron service ports.
For example portprober agent service cannot be disabled via API like
DHCP, so when cheking used ips these ports should be taken into account
after each subnet creation.
Change-Id: I25dbc6f2717b017c2f51d7608dd9cafaa377550a
Depends-On: I089bc013188079289914d98fd24cbfd3c4d91530
Related-Prod: https://mirantis.jira.com/browse/PRODX-42203
diff --git a/neutron_tempest_plugin/api/test_network_ip_availability.py b/neutron_tempest_plugin/api/test_network_ip_availability.py
index 22d2fc6..8ec268d 100644
--- a/neutron_tempest_plugin/api/test_network_ip_availability.py
+++ b/neutron_tempest_plugin/api/test_network_ip_availability.py
@@ -22,9 +22,12 @@
from tempest.lib import exceptions
from neutron_tempest_plugin.api import base
+from neutron_tempest_plugin import config
from neutron_lib import constants as lib_constants
+CONF = config.CONF
+
# 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
@@ -68,6 +71,20 @@
availability = self._get_availability(network, net_availability)
return availability and availability['used_ips']
+ def _wait_used_ips_after_subnet_create(self, network, initial_used_ips=0):
+
+ def _wait_used_ips_value(network, used_ips):
+ availabilities = self.admin_client.list_network_ip_availabilities()
+ current_used_ips = self._get_used_ips(network, availabilities)
+ return current_used_ips == used_ips
+
+ used_ips = initial_used_ips + CONF.network.service_ports_number
+
+ self.assertTrue(
+ test_utils.call_until_true(
+ _wait_used_ips_value, 60, 5, network, used_ips),
+ msg=f"Current used ips != {used_ips} after subnet create")
+
def _assert_total_and_used_ips(self, expected_used, expected_total,
network, net_availability):
availability = self._get_availability(network, net_availability)
@@ -101,6 +118,7 @@
@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)
+ self._wait_used_ips_after_subnet_create(self.network)
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)
@@ -114,6 +132,7 @@
@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)
+ self._wait_used_ips_after_subnet_create(self.network)
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)
@@ -136,6 +155,7 @@
self.network['id'])
self._assert_total_and_used_ips(0, 0, self.network, net_availability)
subnet = self.create_subnet(self.network, enable_dhcp=False)
+ self._wait_used_ips_after_subnet_create(self.network)
prefix = netaddr.IPNetwork(subnet['cidr']).prefixlen
net_availability = self.admin_client.show_network_ip_availability(
self.network['id'])
@@ -153,6 +173,7 @@
@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)
+ self._wait_used_ips_after_subnet_create(self.network)
port = self.create_port(self.network)
net_availability = self.admin_client.show_network_ip_availability(
self.network['id'])
@@ -185,6 +206,7 @@
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)
+ self._wait_used_ips_after_subnet_create(self.network)
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)