Split out Neutron ports client
Splitting out a ports client for Neutron.
Partially implements blueprint consistent-service-method-names
Change-Id: I1248cc6132f4a2e40ad13f6177c7ecda834db57d
diff --git a/tempest/api/network/test_ports.py b/tempest/api/network/test_ports.py
index 6f58075..d0ed1f2 100644
--- a/tempest/api/network/test_ports.py
+++ b/tempest/api/network/test_ports.py
@@ -45,8 +45,8 @@
cls.port = cls.create_port(cls.network)
def _delete_port(self, port_id):
- self.client.delete_port(port_id)
- body = self.client.list_ports()
+ self.ports_client.delete_port(port_id)
+ body = self.ports_client.list_ports()
ports_list = body['ports']
self.assertFalse(port_id in [n['id'] for n in ports_list])
@@ -54,16 +54,16 @@
@test.idempotent_id('c72c1c0c-2193-4aca-aaa4-b1442640f51c')
def test_create_update_delete_port(self):
# Verify port creation
- body = self.client.create_port(network_id=self.network['id'])
+ body = self.ports_client.create_port(network_id=self.network['id'])
port = body['port']
# Schedule port deletion with verification upon test completion
self.addCleanup(self._delete_port, port['id'])
self.assertTrue(port['admin_state_up'])
# Verify port update
new_name = "New_Port"
- body = self.client.update_port(port['id'],
- name=new_name,
- admin_state_up=False)
+ body = self.ports_client.update_port(port['id'],
+ name=new_name,
+ admin_state_up=False)
updated_port = body['port']
self.assertEqual(updated_port['name'], new_name)
self.assertFalse(updated_port['admin_state_up'])
@@ -107,8 +107,8 @@
'end': str(address + 6)}]}
subnet = self.create_subnet(network, **allocation_pools)
self.addCleanup(self.subnets_client.delete_subnet, subnet['id'])
- body = self.client.create_port(network_id=net_id)
- self.addCleanup(self.client.delete_port, body['port']['id'])
+ body = self.ports_client.create_port(network_id=net_id)
+ self.addCleanup(self.ports_client.delete_port, body['port']['id'])
port = body['port']
ip_address = port['fixed_ips'][0]['ip_address']
start_ip_address = allocation_pools['allocation_pools'][0]['start']
@@ -120,7 +120,7 @@
@test.idempotent_id('c9a685bd-e83f-499c-939f-9f7863ca259f')
def test_show_port(self):
# Verify the details of port
- body = self.client.show_port(self.port['id'])
+ body = self.ports_client.show_port(self.port['id'])
port = body['port']
self.assertIn('id', port)
# TODO(Santosh)- This is a temporary workaround to compare create_port
@@ -134,8 +134,8 @@
def test_show_port_fields(self):
# Verify specific fields of a port
fields = ['id', 'mac_address']
- body = self.client.show_port(self.port['id'],
- fields=fields)
+ body = self.ports_client.show_port(self.port['id'],
+ fields=fields)
port = body['port']
self.assertEqual(sorted(port.keys()), sorted(fields))
for field_name in fields:
@@ -145,7 +145,7 @@
@test.idempotent_id('cf95b358-3e92-4a29-a148-52445e1ac50e')
def test_list_ports(self):
# Verify the port exists in the list of all ports
- body = self.client.list_ports()
+ body = self.ports_client.list_ports()
ports = [port['id'] for port in body['ports']
if port['id'] == self.port['id']]
self.assertNotEmpty(ports, "Created port not found in the list")
@@ -157,14 +157,14 @@
subnet = self.create_subnet(network)
self.addCleanup(self.subnets_client.delete_subnet, subnet['id'])
# Create two ports
- port_1 = self.client.create_port(network_id=network['id'])
- self.addCleanup(self.client.delete_port, port_1['port']['id'])
- port_2 = self.client.create_port(network_id=network['id'])
- self.addCleanup(self.client.delete_port, port_2['port']['id'])
+ port_1 = self.ports_client.create_port(network_id=network['id'])
+ self.addCleanup(self.ports_client.delete_port, port_1['port']['id'])
+ port_2 = self.ports_client.create_port(network_id=network['id'])
+ self.addCleanup(self.ports_client.delete_port, port_2['port']['id'])
# List ports filtered by fixed_ips
port_1_fixed_ip = port_1['port']['fixed_ips'][0]['ip_address']
fixed_ips = 'ip_address=' + port_1_fixed_ip
- port_list = self.client.list_ports(fixed_ips=fixed_ips)
+ port_list = self.ports_client.list_ports(fixed_ips=fixed_ips)
# Check that we got the desired port
ports = port_list['ports']
tenant_ids = set([port['tenant_id'] for port in ports])
@@ -190,14 +190,14 @@
self.addCleanup(self.subnets_client.delete_subnet, subnet['id'])
router = self.create_router(data_utils.rand_name('router-'))
self.addCleanup(self.client.delete_router, router['id'])
- port = self.client.create_port(network_id=network['id'])
+ port = self.ports_client.create_port(network_id=network['id'])
# Add router interface to port created above
self.client.add_router_interface_with_port_id(
router['id'], port['port']['id'])
self.addCleanup(self.client.remove_router_interface_with_port_id,
router['id'], port['port']['id'])
# List ports filtered by router_id
- port_list = self.client.list_ports(device_id=router['id'])
+ port_list = self.ports_client.list_ports(device_id=router['id'])
ports = port_list['ports']
self.assertEqual(len(ports), 1)
self.assertEqual(ports[0]['id'], port['port']['id'])
@@ -207,7 +207,7 @@
def test_list_ports_fields(self):
# Verify specific fields of ports
fields = ['id', 'mac_address']
- body = self.client.list_ports(fields=fields)
+ body = self.ports_client.list_ports(fields=fields)
ports = body['ports']
self.assertNotEmpty(ports, "Port list returned is empty")
# Asserting the fields returned are correct
@@ -231,7 +231,7 @@
# Create a port with multiple IP addresses
port = self.create_port(network,
fixed_ips=fixed_ips)
- self.addCleanup(self.client.delete_port, port['id'])
+ self.addCleanup(self.ports_client.delete_port, port['id'])
self.assertEqual(2, len(port['fixed_ips']))
check_fixed_ips = [subnet_1['id'], subnet_2['id']]
for item in port['fixed_ips']:
@@ -269,8 +269,8 @@
"network_id": self.network['id'],
"admin_state_up": True,
"fixed_ips": fixed_ip_1}
- body = self.client.create_port(**post_body)
- self.addCleanup(self.client.delete_port, body['port']['id'])
+ body = self.ports_client.create_port(**post_body)
+ self.addCleanup(self.ports_client.delete_port, body['port']['id'])
port = body['port']
# Update the port with security groups
@@ -280,7 +280,7 @@
"admin_state_up": False,
"fixed_ips": fixed_ip_2,
"security_groups": security_groups_list}
- body = self.client.update_port(port['id'], **update_body)
+ body = self.ports_client.update_port(port['id'], **update_body)
port_show = body['port']
# Verify the security groups and other attributes updated to port
exclude_keys = set(port_show).symmetric_difference(update_body)
@@ -308,16 +308,16 @@
@test.idempotent_id('13e95171-6cbd-489c-9d7c-3f9c58215c18')
def test_create_show_delete_port_user_defined_mac(self):
# Create a port for a legal mac
- body = self.client.create_port(network_id=self.network['id'])
+ body = self.ports_client.create_port(network_id=self.network['id'])
old_port = body['port']
free_mac_address = old_port['mac_address']
- self.client.delete_port(old_port['id'])
+ self.ports_client.delete_port(old_port['id'])
# Create a new port with user defined mac
- body = self.client.create_port(network_id=self.network['id'],
- mac_address=free_mac_address)
- self.addCleanup(self.client.delete_port, body['port']['id'])
+ body = self.ports_client.create_port(network_id=self.network['id'],
+ mac_address=free_mac_address)
+ self.addCleanup(self.ports_client.delete_port, body['port']['id'])
port = body['port']
- body = self.client.show_port(port['id'])
+ body = self.ports_client.show_port(port['id'])
show_port = body['port']
self.assertEqual(free_mac_address,
show_port['mac_address'])
@@ -330,7 +330,7 @@
subnet = self.create_subnet(network)
self.addCleanup(self.subnets_client.delete_subnet, subnet['id'])
port = self.create_port(network, security_groups=[])
- self.addCleanup(self.client.delete_port, port['id'])
+ self.addCleanup(self.ports_client.delete_port, port['id'])
self.assertIsNotNone(port['security_groups'])
self.assertEmpty(port['security_groups'])
@@ -352,9 +352,9 @@
def test_create_port_binding_ext_attr(self):
post_body = {"network_id": self.network['id'],
"binding:host_id": self.host_id}
- body = self.admin_client.create_port(**post_body)
+ body = self.admin_ports_client.create_port(**post_body)
port = body['port']
- self.addCleanup(self.admin_client.delete_port, port['id'])
+ self.addCleanup(self.admin_ports_client.delete_port, port['id'])
host_id = port['binding:host_id']
self.assertIsNotNone(host_id)
self.assertEqual(self.host_id, host_id)
@@ -362,11 +362,11 @@
@test.idempotent_id('6f6c412c-711f-444d-8502-0ac30fbf5dd5')
def test_update_port_binding_ext_attr(self):
post_body = {"network_id": self.network['id']}
- body = self.admin_client.create_port(**post_body)
+ body = self.admin_ports_client.create_port(**post_body)
port = body['port']
- self.addCleanup(self.admin_client.delete_port, port['id'])
+ self.addCleanup(self.admin_ports_client.delete_port, port['id'])
update_body = {"binding:host_id": self.host_id}
- body = self.admin_client.update_port(port['id'], **update_body)
+ body = self.admin_ports_client.update_port(port['id'], **update_body)
updated_port = body['port']
host_id = updated_port['binding:host_id']
self.assertIsNotNone(host_id)
@@ -376,18 +376,18 @@
def test_list_ports_binding_ext_attr(self):
# Create a new port
post_body = {"network_id": self.network['id']}
- body = self.admin_client.create_port(**post_body)
+ body = self.admin_ports_client.create_port(**post_body)
port = body['port']
- self.addCleanup(self.admin_client.delete_port, port['id'])
+ self.addCleanup(self.admin_ports_client.delete_port, port['id'])
# Update the port's binding attributes so that is now 'bound'
# to a host
update_body = {"binding:host_id": self.host_id}
- self.admin_client.update_port(port['id'], **update_body)
+ self.admin_ports_client.update_port(port['id'], **update_body)
# List all ports, ensure new port is part of list and its binding
# attributes are set and accurate
- body = self.admin_client.list_ports()
+ body = self.admin_ports_client.list_ports()
ports_list = body['ports']
pids_list = [p['id'] for p in ports_list]
self.assertIn(port['id'], pids_list)
@@ -399,10 +399,11 @@
@test.idempotent_id('b54ac0ff-35fc-4c79-9ca3-c7dbd4ea4f13')
def test_show_port_binding_ext_attr(self):
- body = self.admin_client.create_port(network_id=self.network['id'])
+ body = self.admin_ports_client.create_port(
+ network_id=self.network['id'])
port = body['port']
- self.addCleanup(self.admin_client.delete_port, port['id'])
- body = self.admin_client.show_port(port['id'])
+ self.addCleanup(self.admin_ports_client.delete_port, port['id'])
+ body = self.admin_ports_client.show_port(port['id'])
show_port = body['port']
self.assertEqual(port['binding:host_id'],
show_port['binding:host_id'])