openstack/tempest
Code Issues Proposed changes

Merge "Change neutron client methods to return one value and update tests"

Browse Source
This commit is contained in:
Jenkins
2014-12-19 01:37:53 +00:00
committed by Gerrit Code Review
parent 16b6826b94 34e881201d
commit c7b08af36f
40 changed files with 530 additions and 536 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
tempest/api/compute/servers/test_create_server.py
View File

@@ -130,23 +130,23 @@ class ServersTestJSON(base.BaseV2ComputeTest):
# Verify that the networks order given at the server creation is
# preserved within the server.
name_net1 = data_utils.rand_name(self.__class__.__name__)
_, net1 = self.network_client.create_network(name=name_net1)
net1 = self.network_client.create_network(name=name_net1)
self.addCleanup(self.network_client.delete_network,
net1['network']['id'])
name_net2 = data_utils.rand_name(self.__class__.__name__)
_, net2 = self.network_client.create_network(name=name_net2)
net2 = self.network_client.create_network(name=name_net2)
self.addCleanup(self.network_client.delete_network,
net2['network']['id'])
_, subnet1 = self.network_client.create_subnet(
subnet1 = self.network_client.create_subnet(
network_id=net1['network']['id'],
cidr='19.80.0.0/24',
ip_version=4)
self.addCleanup(self.network_client.delete_subnet,
subnet1['subnet']['id'])
_, subnet2 = self.network_client.create_subnet(
subnet2 = self.network_client.create_subnet(
network_id=net2['network']['id'],
cidr='19.86.0.0/24',
ip_version=4)

16
tempest/api/network/admin/test_agent_management.py
View File

@@ -26,13 +26,13 @@ class AgentManagementTestJSON(base.BaseAdminNetworkTest):
if not test.is_extension_enabled('agent', 'network'):
msg = "agent extension not enabled."
raise cls.skipException(msg)
resp, body = cls.admin_client.list_agents()
body = cls.admin_client.list_agents()
agents = body['agents']
cls.agent = agents[0]
@test.attr(type='smoke')
def test_list_agent(self):
_, body = self.admin_client.list_agents()
body = self.admin_client.list_agents()
agents = body['agents']
# Hearthbeats must be excluded from comparison
self.agent.pop('heartbeat_timestamp', None)
@@ -44,12 +44,12 @@ class AgentManagementTestJSON(base.BaseAdminNetworkTest):
@test.attr(type=['smoke'])
def test_list_agents_non_admin(self):
_, body = self.client.list_agents()
body = self.client.list_agents()
self.assertEqual(len(body["agents"]), 0)
@test.attr(type='smoke')
def test_show_agent(self):
_, body = self.admin_client.show_agent(self.agent['id'])
body = self.admin_client.show_agent(self.agent['id'])
agent = body['agent']
self.assertEqual(agent['id'], self.agent['id'])
@@ -59,8 +59,8 @@ class AgentManagementTestJSON(base.BaseAdminNetworkTest):
# Try to update the 'admin_state_up' to the original
# one to avoid the negative effect.
agent_status = {'admin_state_up': origin_status}
_, body = self.admin_client.update_agent(agent_id=self.agent['id'],
agent_info=agent_status)
body = self.admin_client.update_agent(agent_id=self.agent['id'],
agent_info=agent_status)
updated_status = body['agent']['admin_state_up']
self.assertEqual(origin_status, updated_status)
@@ -69,8 +69,8 @@ class AgentManagementTestJSON(base.BaseAdminNetworkTest):
self.useFixture(fixtures.LockFixture('agent_description'))
description = 'description for update agent.'
agent_description = {'description': description}
_, body = self.admin_client.update_agent(agent_id=self.agent['id'],
agent_info=agent_description)
body = self.admin_client.update_agent(agent_id=self.agent['id'],
agent_info=agent_description)
self.addCleanup(self._restore_agent)
updated_description = body['agent']['description']
self.assertEqual(updated_description, description)

14
tempest/api/network/admin/test_dhcp_agent_scheduler.py
View File

@@ -34,12 +34,12 @@ class DHCPAgentSchedulersTestJSON(base.BaseAdminNetworkTest):
@test.attr(type='smoke')
def test_list_dhcp_agent_hosting_network(self):
_, body = self.admin_client.list_dhcp_agent_hosting_network(
self.admin_client.list_dhcp_agent_hosting_network(
self.network['id'])
@test.attr(type='smoke')
def test_list_networks_hosted_by_one_dhcp(self):
resp, body = self.admin_client.list_dhcp_agent_hosting_network(
body = self.admin_client.list_dhcp_agent_hosting_network(
self.network['id'])
agents = body['agents']
self.assertIsNotNone(agents)
@@ -49,7 +49,7 @@ class DHCPAgentSchedulersTestJSON(base.BaseAdminNetworkTest):
def _check_network_in_dhcp_agent(self, network_id, agent):
network_ids = []
_, body = self.admin_client.list_networks_hosted_by_one_dhcp_agent(
body = self.admin_client.list_networks_hosted_by_one_dhcp_agent(
agent['id'])
networks = body['networks']
for network in networks:
@@ -63,7 +63,7 @@ class DHCPAgentSchedulersTestJSON(base.BaseAdminNetworkTest):
self.ports.remove(self.port)
agent = dict()
agent['agent_type'] = None
resp, body = self.admin_client.list_agents()
body = self.admin_client.list_agents()
agents = body['agents']
for a in agents:
if a['agent_type'] == 'DHCP agent':
@@ -82,14 +82,14 @@ class DHCPAgentSchedulersTestJSON(base.BaseAdminNetworkTest):
self._remove_network_from_dhcp_agent(network_id, agent)
def _remove_network_from_dhcp_agent(self, network_id, agent):
_, body = self.admin_client.remove_network_from_dhcp_agent(
self.admin_client.remove_network_from_dhcp_agent(
agent_id=agent['id'],
network_id=network_id)
self.assertFalse(self._check_network_in_dhcp_agent(
network_id, agent))
def _add_dhcp_agent_to_network(self, network_id, agent):
_, body = self.admin_client.add_dhcp_agent_to_network(agent['id'],
network_id)
self.admin_client.add_dhcp_agent_to_network(agent['id'],
network_id)
self.assertTrue(self._check_network_in_dhcp_agent(
network_id, agent))

22
tempest/api/network/admin/test_external_network_extension.py
View File

@@ -26,7 +26,7 @@ class ExternalNetworksTestJSON(base.BaseAdminNetworkTest):
post_body = {'name': data_utils.rand_name('network-')}
if external:
post_body['router:external'] = external
_, body = self.admin_client.create_network(**post_body)
body = self.admin_client.create_network(**post_body)
network = body['network']
self.addCleanup(self.admin_client.delete_network, network['id'])
return network
@@ -45,8 +45,8 @@ class ExternalNetworksTestJSON(base.BaseAdminNetworkTest):
network = self._create_network(external=False)
self.assertFalse(network.get('router:external', False))
update_body = {'router:external': True}
_, body = self.admin_client.update_network(network['id'],
**update_body)
body = self.admin_client.update_network(network['id'],
**update_body)
updated_network = body['network']
# Verify that router:external parameter was updated
self.assertTrue(updated_network['router:external'])
@@ -57,7 +57,7 @@ class ExternalNetworksTestJSON(base.BaseAdminNetworkTest):
# List networks as a normal user and confirm the external
# network extension attribute is returned for those networks
# that were created as external
_, body = self.client.list_networks()
body = self.client.list_networks()
networks_list = [net['id'] for net in body['networks']]
self.assertIn(external_network['id'], networks_list)
self.assertIn(self.network['id'], networks_list)
@@ -72,12 +72,12 @@ class ExternalNetworksTestJSON(base.BaseAdminNetworkTest):
external_network = self._create_network()
# Show an external network as a normal user and confirm the
# external network extension attribute is returned.
_, body = self.client.show_network(external_network['id'])
body = self.client.show_network(external_network['id'])
show_ext_net = body['network']
self.assertEqual(external_network['name'], show_ext_net['name'])
self.assertEqual(external_network['id'], show_ext_net['id'])
self.assertTrue(show_ext_net['router:external'])
_, body = self.client.show_network(self.network['id'])
body = self.client.show_network(self.network['id'])
show_net = body['network']
# Verify with show that router:external is False for network
self.assertEqual(self.network['name'], show_net['name'])
@@ -91,29 +91,29 @@ class ExternalNetworksTestJSON(base.BaseAdminNetworkTest):
"""
# Set cls.client to admin to use base.create_subnet()
client = self.admin_client
_, body = client.create_network(**{'router:external': True})
body = client.create_network(**{'router:external': True})
external_network = body['network']
self.addCleanup(self._try_delete_resource,
client.delete_network,
external_network['id'])
subnet = self.create_subnet(external_network, client=client)
_, body = client.create_floatingip(
body = client.create_floatingip(
floating_network_id=external_network['id'])
created_floating_ip = body['floatingip']
self.addCleanup(self._try_delete_resource,
client.delete_floatingip,
created_floating_ip['id'])
_, floatingip_list = client.list_floatingips(
floatingip_list = client.list_floatingips(
network=external_network['id'])
self.assertIn(created_floating_ip['id'],
(f['id'] for f in floatingip_list['floatingips']))
client.delete_network(external_network['id'])
# Verifies floating ip is deleted
_, floatingip_list = client.list_floatingips()
floatingip_list = client.list_floatingips()
self.assertNotIn(created_floating_ip['id'],
(f['id'] for f in floatingip_list['floatingips']))
# Verifies subnet is deleted
_, subnet_list = client.list_subnets()
subnet_list = client.list_subnets()
self.assertNotIn(subnet['id'],
(s['id'] for s in subnet_list))
# Removes subnet from the cleanup list

8
tempest/api/network/admin/test_floating_ips_admin_actions.py
View File

@@ -36,18 +36,18 @@ class FloatingIPAdminTestJSON(base.BaseAdminNetworkTest):
@test.attr(type='smoke')
def test_list_floating_ips_from_admin_and_nonadmin(self):
# Create floating ip from admin user
_, floating_ip_admin = self.admin_client.create_floatingip(
floating_ip_admin = self.admin_client.create_floatingip(
floating_network_id=self.ext_net_id)
self.addCleanup(self.admin_client.delete_floatingip,
floating_ip_admin['floatingip']['id'])
# Create floating ip from alt user
_, body = self.alt_client.create_floatingip(
body = self.alt_client.create_floatingip(
floating_network_id=self.ext_net_id)
floating_ip_alt = body['floatingip']
self.addCleanup(self.alt_client.delete_floatingip,
floating_ip_alt['id'])
# List floating ips from admin
_, body = self.admin_client.list_floatingips()
body = self.admin_client.list_floatingips()
floating_ip_ids_admin = [f['id'] for f in body['floatingips']]
# Check that admin sees all floating ips
self.assertIn(self.floating_ip['id'], floating_ip_ids_admin)
@@ -55,7 +55,7 @@ class FloatingIPAdminTestJSON(base.BaseAdminNetworkTest):
floating_ip_ids_admin)
self.assertIn(floating_ip_alt['id'], floating_ip_ids_admin)
# List floating ips from nonadmin
resp, body = self.client.list_floatingips()
body = self.client.list_floatingips()
floating_ip_ids = [f['id'] for f in body['floatingips']]
# Check that nonadmin user doesn't see floating ip created from admin
# and floating ip that is created in another tenant (alt user)

12
tempest/api/network/admin/test_l3_agent_scheduler.py
View File

@@ -40,7 +40,7 @@ class L3AgentSchedulerTestJSON(base.BaseAdminNetworkTest):
msg = "L3 Agent Scheduler Extension not enabled."
raise cls.skipException(msg)
# Trying to get agent details for L3 Agent
resp, body = cls.admin_client.list_agents()
body = cls.admin_client.list_agents()
agents = body['agents']
for agent in agents:
if agent['agent_type'] == 'L3 agent':
@@ -52,18 +52,18 @@ class L3AgentSchedulerTestJSON(base.BaseAdminNetworkTest):
@test.attr(type='smoke')
def test_list_routers_on_l3_agent(self):
_, body = self.admin_client.list_routers_on_l3_agent(self.agent['id'])
self.admin_client.list_routers_on_l3_agent(self.agent['id'])
@test.attr(type='smoke')
def test_add_list_remove_router_on_l3_agent(self):
l3_agent_ids = list()
name = data_utils.rand_name('router1-')
resp, router = self.client.create_router(name)
router = self.client.create_router(name)
self.addCleanup(self.client.delete_router, router['router']['id'])
_, body = self.admin_client.add_router_to_l3_agent(
self.admin_client.add_router_to_l3_agent(
self.agent['id'],
router['router']['id'])
_, body = self.admin_client.list_l3_agents_hosting_router(
body = self.admin_client.list_l3_agents_hosting_router(
router['router']['id'])
for agent in body['agents']:
l3_agent_ids.append(agent['id'])
@@ -71,7 +71,7 @@ class L3AgentSchedulerTestJSON(base.BaseAdminNetworkTest):
self.assertEqual('L3 agent', agent['agent_type'])
self.assertIn(self.agent['id'], l3_agent_ids)
del l3_agent_ids[:]
_, body = self.admin_client.remove_router_from_l3_agent(
body = self.admin_client.remove_router_from_l3_agent(
self.agent['id'],
router['router']['id'])
# NOTE(afazekas): The deletion not asserted, because neutron

6
tempest/api/network/admin/test_lbaas_agent_scheduler.py
View File

@@ -49,13 +49,13 @@ class LBaaSAgentSchedulerTestJSON(base.BaseAdminNetworkTest):
@test.attr(type='smoke')
def test_list_pools_on_lbaas_agent(self):
found = False
_, body = self.admin_client.list_agents(
body = self.admin_client.list_agents(
agent_type="Loadbalancer agent")
agents = body['agents']
for a in agents:
msg = 'Load Balancer agent expected'
self.assertEqual(a['agent_type'], 'Loadbalancer agent', msg)
_, body = (
body = (
self.admin_client.list_pools_hosted_by_one_lbaas_agent(
a['id']))
pools = body['pools']
@@ -66,6 +66,6 @@ class LBaaSAgentSchedulerTestJSON(base.BaseAdminNetworkTest):
@test.attr(type='smoke')
def test_show_lbaas_agent_hosting_pool(self):
_, body = self.admin_client.show_lbaas_agent_hosting_pool(
body = self.admin_client.show_lbaas_agent_hosting_pool(
self.pool['id'])
self.assertEqual('Loadbalancer agent', body['agent']['agent_type'])

30
tempest/api/network/admin/test_load_balancer_admin_actions.py
View File

@@ -46,7 +46,7 @@ class LoadBalancerAdminTestJSON(base.BaseAdminNetworkTest):
@test.attr(type='smoke')
def test_create_vip_as_admin_for_another_tenant(self):
name = data_utils.rand_name('vip-')
_, body = self.admin_client.create_pool(
body = self.admin_client.create_pool(
name=data_utils.rand_name('pool-'),
lb_method="ROUND_ROBIN",
protocol="HTTP",
@@ -54,24 +54,24 @@ class LoadBalancerAdminTestJSON(base.BaseAdminNetworkTest):
tenant_id=self.tenant_id)
pool = body['pool']
self.addCleanup(self.admin_client.delete_pool, pool['id'])
_, body = self.admin_client.create_vip(name=name,
protocol="HTTP",
protocol_port=80,
subnet_id=self.subnet['id'],
pool_id=pool['id'],
tenant_id=self.tenant_id)
body = self.admin_client.create_vip(name=name,
protocol="HTTP",
protocol_port=80,
subnet_id=self.subnet['id'],
pool_id=pool['id'],
tenant_id=self.tenant_id)
vip = body['vip']
self.addCleanup(self.admin_client.delete_vip, vip['id'])
self.assertIsNotNone(vip['id'])
self.assertEqual(self.tenant_id, vip['tenant_id'])
_, body = self.client.show_vip(vip['id'])
body = self.client.show_vip(vip['id'])
show_vip = body['vip']
self.assertEqual(vip['id'], show_vip['id'])
self.assertEqual(vip['name'], show_vip['name'])
@test.attr(type='smoke')
def test_create_health_monitor_as_admin_for_another_tenant(self):
_, body = (
body = (
self.admin_client.create_health_monitor(delay=4,
max_retries=3,
type="TCP",
@@ -82,13 +82,13 @@ class LoadBalancerAdminTestJSON(base.BaseAdminNetworkTest):
health_monitor['id'])
self.assertIsNotNone(health_monitor['id'])
self.assertEqual(self.tenant_id, health_monitor['tenant_id'])
_, body = self.client.show_health_monitor(health_monitor['id'])
body = self.client.show_health_monitor(health_monitor['id'])
show_health_monitor = body['health_monitor']
self.assertEqual(health_monitor['id'], show_health_monitor['id'])
@test.attr(type='smoke')
def test_create_pool_from_admin_user_other_tenant(self):
_, body = self.admin_client.create_pool(
body = self.admin_client.create_pool(
name=data_utils.rand_name('pool-'),
lb_method="ROUND_ROBIN",
protocol="HTTP",
@@ -101,10 +101,10 @@ class LoadBalancerAdminTestJSON(base.BaseAdminNetworkTest):
@test.attr(type='smoke')
def test_create_member_from_admin_user_other_tenant(self):
_, body = self.admin_client.create_member(address="10.0.9.47",
protocol_port=80,
pool_id=self.pool['id'],
tenant_id=self.tenant_id)
body = self.admin_client.create_member(address="10.0.9.47",
protocol_port=80,
pool_id=self.pool['id'],
tenant_id=self.tenant_id)
member = body['member']
self.addCleanup(self.admin_client.delete_member, member['id'])
self.assertIsNotNone(member['id'])

12
tempest/api/network/admin/test_quotas.py
View File

@@ -57,14 +57,14 @@ class QuotasTest(base.BaseAdminNetworkTest):
self.addCleanup(self.identity_admin_client.delete_tenant, tenant_id)
# Change quotas for tenant
_, quota_set = self.admin_client.update_quotas(tenant_id,
**new_quotas)
quota_set = self.admin_client.update_quotas(tenant_id,
**new_quotas)
self.addCleanup(self.admin_client.reset_quotas, tenant_id)
for key, value in new_quotas.iteritems():
self.assertEqual(value, quota_set[key])
# Confirm our tenant is listed among tenants with non default quotas
_, non_default_quotas = self.admin_client.list_quotas()
non_default_quotas = self.admin_client.list_quotas()
found = False
for qs in non_default_quotas['quotas']:
if qs['tenant_id'] == tenant_id:
@@ -72,14 +72,14 @@ class QuotasTest(base.BaseAdminNetworkTest):
self.assertTrue(found)
# Confirm from API quotas were changed as requested for tenant
_, quota_set = self.admin_client.show_quotas(tenant_id)
quota_set = self.admin_client.show_quotas(tenant_id)
quota_set = quota_set['quota']
for key, value in new_quotas.iteritems():
self.assertEqual(value, quota_set[key])
# Reset quotas to default and confirm
_, body = self.admin_client.reset_quotas(tenant_id)
_, non_default_quotas = self.admin_client.list_quotas()
self.admin_client.reset_quotas(tenant_id)
non_default_quotas = self.admin_client.list_quotas()
for q in non_default_quotas['quotas']:
self.assertNotEqual(tenant_id, q['tenant_id'])

68
tempest/api/network/base.py
View File

@@ -185,7 +185,7 @@ class BaseNetworkTest(tempest.test.BaseTestCase):
"""Wrapper utility that returns a test network."""
network_name = network_name or data_utils.rand_name('test-network-')
resp, body = cls.client.create_network(name=network_name)
body = cls.client.create_network(name=network_name)
network = body['network']
cls.networks.append(network)
return network
@@ -216,7 +216,7 @@ class BaseNetworkTest(tempest.test.BaseTestCase):
else:
gateway_ip = gateway
try:
resp, body = client.create_subnet(
body = client.create_subnet(
network_id=network['id'],
cidr=str(subnet_cidr),
ip_version=ip_version,
@@ -237,8 +237,8 @@ class BaseNetworkTest(tempest.test.BaseTestCase):
@classmethod
def create_port(cls, network, **kwargs):
"""Wrapper utility that returns a test port."""
resp, body = cls.client.create_port(network_id=network['id'],
**kwargs)
body = cls.client.create_port(network_id=network['id'],
**kwargs)
port = body['port']
cls.ports.append(port)
return port
@@ -246,8 +246,8 @@ class BaseNetworkTest(tempest.test.BaseTestCase):
@classmethod
def update_port(cls, port, **kwargs):
"""Wrapper utility that updates a test port."""
resp, body = cls.client.update_port(port['id'],
**kwargs)
body = cls.client.update_port(port['id'],
**kwargs)
return body['port']
@classmethod
@@ -258,7 +258,7 @@ class BaseNetworkTest(tempest.test.BaseTestCase):
ext_gw_info['network_id'] = external_network_id
if enable_snat:
ext_gw_info['enable_snat'] = enable_snat
resp, body = cls.client.create_router(
body = cls.client.create_router(
router_name, external_gateway_info=ext_gw_info,
admin_state_up=admin_state_up)
router = body['router']
@@ -268,7 +268,7 @@ class BaseNetworkTest(tempest.test.BaseTestCase):
@classmethod
def create_floatingip(cls, external_network_id):
"""Wrapper utility that returns a test floating IP."""
resp, body = cls.client.create_floatingip(
body = cls.client.create_floatingip(
floating_network_id=external_network_id)
fip = body['floatingip']
cls.floating_ips.append(fip)
@@ -277,7 +277,7 @@ class BaseNetworkTest(tempest.test.BaseTestCase):
@classmethod
def create_pool(cls, name, lb_method, protocol, subnet):
"""Wrapper utility that returns a test pool."""
resp, body = cls.client.create_pool(
body = cls.client.create_pool(
name=name,
lb_method=lb_method,
protocol=protocol,
@@ -289,25 +289,25 @@ class BaseNetworkTest(tempest.test.BaseTestCase):
@classmethod
def update_pool(cls, name):
"""Wrapper utility that returns a test pool."""
resp, body = cls.client.update_pool(name=name)
body = cls.client.update_pool(name=name)
pool = body['pool']
return pool
@classmethod
def create_vip(cls, name, protocol, protocol_port, subnet, pool):
"""Wrapper utility that returns a test vip."""
resp, body = cls.client.create_vip(name=name,
protocol=protocol,
protocol_port=protocol_port,
subnet_id=subnet['id'],
pool_id=pool['id'])
body = cls.client.create_vip(name=name,
protocol=protocol,
protocol_port=protocol_port,
subnet_id=subnet['id'],
pool_id=pool['id'])
vip = body['vip']
cls.vips.append(vip)
return vip
@classmethod
def update_vip(cls, name):
resp, body = cls.client.update_vip(name=name)
body = cls.client.update_vip(name=name)
vip = body['vip']
return vip
@@ -316,47 +316,47 @@ class BaseNetworkTest(tempest.test.BaseTestCase):
"""Wrapper utility that returns a test member."""
ip_version = ip_version if ip_version is not None else cls._ip_version
member_address = "fd00::abcd" if ip_version == 6 else "10.0.9.46"
resp, body = cls.client.create_member(address=member_address,
protocol_port=protocol_port,
pool_id=pool['id'])
body = cls.client.create_member(address=member_address,
protocol_port=protocol_port,
pool_id=pool['id'])
member = body['member']
cls.members.append(member)
return member
@classmethod
def update_member(cls, admin_state_up):
resp, body = cls.client.update_member(admin_state_up=admin_state_up)
body = cls.client.update_member(admin_state_up=admin_state_up)
member = body['member']
return member
@classmethod
def create_health_monitor(cls, delay, max_retries, Type, timeout):
"""Wrapper utility that returns a test health monitor."""
resp, body = cls.client.create_health_monitor(delay=delay,
max_retries=max_retries,
type=Type,
timeout=timeout)
body = cls.client.create_health_monitor(delay=delay,
max_retries=max_retries,
type=Type,
timeout=timeout)
health_monitor = body['health_monitor']
cls.health_monitors.append(health_monitor)
return health_monitor
@classmethod
def update_health_monitor(cls, admin_state_up):
resp, body = cls.client.update_vip(admin_state_up=admin_state_up)
body = cls.client.update_vip(admin_state_up=admin_state_up)
health_monitor = body['health_monitor']
return health_monitor
@classmethod
def create_router_interface(cls, router_id, subnet_id):
"""Wrapper utility that returns a router interface."""
resp, interface = cls.client.add_router_interface_with_subnet_id(
interface = cls.client.add_router_interface_with_subnet_id(
router_id, subnet_id)
return interface
@classmethod
def create_vpnservice(cls, subnet_id, router_id):
"""Wrapper utility that returns a test vpn service."""
resp, body = cls.client.create_vpnservice(
body = cls.client.create_vpnservice(
subnet_id=subnet_id, router_id=router_id, admin_state_up=True,
name=data_utils.rand_name("vpnservice-"))
vpnservice = body['vpnservice']
@@ -366,7 +366,7 @@ class BaseNetworkTest(tempest.test.BaseTestCase):
@classmethod
def create_ikepolicy(cls, name):
"""Wrapper utility that returns a test ike policy."""
resp, body = cls.client.create_ikepolicy(name=name)
body = cls.client.create_ikepolicy(name=name)
ikepolicy = body['ikepolicy']
cls.ikepolicies.append(ikepolicy)
return ikepolicy
@@ -374,7 +374,7 @@ class BaseNetworkTest(tempest.test.BaseTestCase):
@classmethod
def create_firewall_rule(cls, action, protocol):
"""Wrapper utility that returns a test firewall rule."""
resp, body = cls.client.create_firewall_rule(
body = cls.client.create_firewall_rule(
name=data_utils.rand_name("fw-rule"),
action=action,
protocol=protocol)
@@ -385,7 +385,7 @@ class BaseNetworkTest(tempest.test.BaseTestCase):
@classmethod
def create_firewall_policy(cls):
"""Wrapper utility that returns a test firewall policy."""
resp, body = cls.client.create_firewall_policy(
body = cls.client.create_firewall_policy(
name=data_utils.rand_name("fw-policy"))
fw_policy = body['firewall_policy']
cls.fw_policies.append(fw_policy)
@@ -393,7 +393,7 @@ class BaseNetworkTest(tempest.test.BaseTestCase):
@classmethod
def delete_router(cls, router):
resp, body = cls.client.list_router_interfaces(router['id'])
body = cls.client.list_router_interfaces(router['id'])
interfaces = body['ports']
for i in interfaces:
cls.client.remove_router_interface_with_subnet_id(
@@ -403,7 +403,7 @@ class BaseNetworkTest(tempest.test.BaseTestCase):
@classmethod
def create_ipsecpolicy(cls, name):
"""Wrapper utility that returns a test ipsec policy."""
_, body = cls.client.create_ipsecpolicy(name=name)
body = cls.client.create_ipsecpolicy(name=name)
ipsecpolicy = body['ipsecpolicy']
cls.ipsecpolicies.append(ipsecpolicy)
return ipsecpolicy
@@ -428,7 +428,7 @@ class BaseAdminNetworkTest(BaseNetworkTest):
@classmethod
def create_metering_label(cls, name, description):
"""Wrapper utility that returns a test metering label."""
resp, body = cls.admin_client.create_metering_label(
body = cls.admin_client.create_metering_label(
description=description,
name=data_utils.rand_name("metering-label"))
metering_label = body['metering_label']
@@ -439,7 +439,7 @@ class BaseAdminNetworkTest(BaseNetworkTest):
def create_metering_label_rule(cls, remote_ip_prefix, direction,
metering_label_id):
"""Wrapper utility that returns a test metering label rule."""
resp, body = cls.admin_client.create_metering_label_rule(
body = cls.admin_client.create_metering_label_rule(
remote_ip_prefix=remote_ip_prefix, direction=direction,
metering_label_id=metering_label_id)
metering_label_rule = body['metering_label_rule']

10
tempest/api/network/base_routers.py
View File

@@ -29,14 +29,14 @@ class BaseRouterTest(base.BaseAdminNetworkTest):
self.client.delete_router(router_id)
# Asserting that the router is not found in the list
# after deletion
_, list_body = self.client.list_routers()
list_body = self.client.list_routers()
routers_list = list()
for router in list_body['routers']:
routers_list.append(router['id'])
self.assertNotIn(router_id, routers_list)
def _add_router_interface_with_subnet_id(self, router_id, subnet_id):
_, interface = self.client.add_router_interface_with_subnet_id(
interface = self.client.add_router_interface_with_subnet_id(
router_id, subnet_id)
self.addCleanup(self._remove_router_interface_with_subnet_id,
router_id, subnet_id)
@@ -44,11 +44,11 @@ class BaseRouterTest(base.BaseAdminNetworkTest):
return interface
def _remove_router_interface_with_subnet_id(self, router_id, subnet_id):
_, body = self.client.remove_router_interface_with_subnet_id(
body = self.client.remove_router_interface_with_subnet_id(
router_id, subnet_id)
self.assertEqual(subnet_id, body['subnet_id'])
def _remove_router_interface_with_port_id(self, router_id, port_id):
_, body = self.client.remove_router_interface_with_port_id(router_id,
port_id)
body = self.client.remove_router_interface_with_port_id(router_id,
port_id)
self.assertEqual(port_id, body['port_id'])

6
tempest/api/network/base_security_groups.py
View File

@@ -26,7 +26,7 @@ class BaseSecGroupTest(base.BaseNetworkTest):
def _create_security_group(self):
# Create a security group
name = data_utils.rand_name('secgroup-')
_, group_create_body = self.client.create_security_group(name=name)
group_create_body = self.client.create_security_group(name=name)
self.addCleanup(self._delete_security_group,
group_create_body['security_group']['id'])
self.assertEqual(group_create_body['security_group']['name'], name)
@@ -36,7 +36,7 @@ class BaseSecGroupTest(base.BaseNetworkTest):
self.client.delete_security_group(secgroup_id)
# Asserting that the security group is not found in the list
# after deletion
_, list_body = self.client.list_security_groups()
list_body = self.client.list_security_groups()
secgroup_list = list()
for secgroup in list_body['security_groups']:
secgroup_list.append(secgroup['id'])
@@ -46,7 +46,7 @@ class BaseSecGroupTest(base.BaseNetworkTest):
self.client.delete_security_group_rule(rule_id)
# Asserting that the security group is not found in the list
# after deletion
_, list_body = self.client.list_security_group_rules()
list_body = self.client.list_security_group_rules()
rules_list = list()
for rule in list_body['security_group_rules']:
rules_list.append(rule['id'])

10
tempest/api/network/test_allowed_address_pair.py
View File

@@ -58,14 +58,14 @@ class AllowedAddressPairTestJSON(base.BaseNetworkTest):
# Create port with allowed address pair attribute
allowed_address_pairs = [{'ip_address': self.ip_address,
'mac_address': self.mac_address}]
_, body = self.client.create_port(
body = self.client.create_port(
network_id=self.network['id'],
allowed_address_pairs=allowed_address_pairs)
port_id = body['port']['id']
self.addCleanup(self.client.delete_port, port_id)
# Confirm port was created with allowed address pair attribute
_, body = self.client.list_ports()
body = self.client.list_ports()
ports = body['ports']
port = [p for p in ports if p['id'] == port_id]
msg = 'Created port not found in list of ports returned by Neutron'
@@ -75,7 +75,7 @@ class AllowedAddressPairTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def _update_port_with_address(self, address, mac_address=None, **kwargs):
# Create a port without allowed address pair
_, body = self.client.create_port(network_id=self.network['id'])
body = self.client.create_port(network_id=self.network['id'])
port_id = body['port']['id']
self.addCleanup(self.client.delete_port, port_id)
if mac_address is None:
@@ -86,7 +86,7 @@ class AllowedAddressPairTestJSON(base.BaseNetworkTest):
'mac_address': mac_address}]
if kwargs:
allowed_address_pairs.append(kwargs['allowed_address_pairs'])
_, body = self.client.update_port(
body = self.client.update_port(
port_id, allowed_address_pairs=allowed_address_pairs)
allowed_address_pair = body['port']['allowed_address_pairs']
self.assertEqual(allowed_address_pair, allowed_address_pairs)
@@ -105,7 +105,7 @@ class AllowedAddressPairTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_update_port_with_multiple_ip_mac_address_pair(self):
# Create an ip _address and mac_address through port create
_, resp = self.client.create_port(network_id=self.network['id'])
resp = self.client.create_port(network_id=self.network['id'])
newportid = resp['port']['id']
self.addCleanup(self.client.delete_port, newportid)
ipaddress = resp['port']['fixed_ips'][0]['ip_address']

10
tempest/api/network/test_dhcp_ipv6.py
View File

@@ -57,7 +57,7 @@ class NetworksTestDHCPv6(base.BaseNetworkTest):
del things_list[index]
def _clean_network(self):
resp, body = self.client.list_ports()
body = self.client.list_ports()
ports = body['ports']
for port in ports:
if (port['device_owner'] == 'network:router_interface'
@@ -69,13 +69,13 @@ class NetworksTestDHCPv6(base.BaseNetworkTest):
if port['id'] in [p['id'] for p in self.ports]:
self.client.delete_port(port['id'])
self._remove_from_list_by_index(self.ports, port)
resp, body = self.client.list_subnets()
body = self.client.list_subnets()
subnets = body['subnets']
for subnet in subnets:
if subnet['id'] in [s['id'] for s in self.subnets]:
self.client.delete_subnet(subnet['id'])
self._remove_from_list_by_index(self.subnets, subnet)
resp, body = self.client.list_routers()
body = self.client.list_routers()
routers = body['routers']
for router in routers:
if router['id'] in [r['id'] for r in self.routers]:
@@ -197,7 +197,7 @@ class NetworksTestDHCPv6(base.BaseNetworkTest):
subnet_slaac]
self.client.delete_port(port['id'])
self.ports.pop()
_, body = self.client.list_ports()
body = self.client.list_ports()
ports_id_list = [i['id'] for i in body['ports']]
self.assertNotIn(port['id'], ports_id_list)
self._clean_network()
@@ -361,7 +361,7 @@ class NetworksTestDHCPv6(base.BaseNetworkTest):
admin_state_up=True)
port = self.create_router_interface(router['id'],
subnet['id'])
_, body = self.client.show_port(port['port_id'])
body = self.client.show_port(port['port_id'])
return subnet, body['port']
def test_dhcp_stateful_router(self):

4
tempest/api/network/test_extensions.py
View File

@@ -47,14 +47,14 @@ class ExtensionsTestJSON(base.BaseNetworkTest):
expected_alias = [ext for ext in expected_alias if
test.is_extension_enabled(ext, 'network')]
actual_alias = list()
_, extensions = self.client.list_extensions()
extensions = self.client.list_extensions()
list_extensions = extensions['extensions']
# Show and verify the details of the available extensions
for ext in list_extensions:
ext_name = ext['name']
ext_alias = ext['alias']
actual_alias.append(ext['alias'])
_, ext_details = self.client.show_extension(ext_alias)
ext_details = self.client.show_extension(ext_alias)
ext_details = ext_details['extension']
self.assertIsNotNone(ext_details)

8
tempest/api/network/test_extra_dhcp_options.py
View File

@@ -57,14 +57,14 @@ class ExtraDHCPOptionsTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_create_list_port_with_extra_dhcp_options(self):
# Create a port with Extra DHCP Options
_, body = self.client.create_port(
body = self.client.create_port(
network_id=self.network['id'],
extra_dhcp_opts=self.extra_dhcp_opts)
port_id = body['port']['id']
self.addCleanup(self.client.delete_port, port_id)
# Confirm port created has Extra DHCP Options
_, body = self.client.list_ports()
body = self.client.list_ports()
ports = body['ports']
port = [p for p in ports if p['id'] == port_id]
self.assertTrue(port)
@@ -74,12 +74,12 @@ class ExtraDHCPOptionsTestJSON(base.BaseNetworkTest):
def test_update_show_port_with_extra_dhcp_options(self):
# Update port with extra dhcp options
name = data_utils.rand_name('new-port-name')
_, body = self.client.update_port(
body = self.client.update_port(
self.port['id'],
name=name,
extra_dhcp_opts=self.extra_dhcp_opts)
# Confirm extra dhcp options were added to the port
_, body = self.client.show_port(self.port['id'])
body = self.client.show_port(self.port['id'])
self._confirm_extra_dhcp_options(body['port'], self.extra_dhcp_opts)
def _confirm_extra_dhcp_options(self, port, extra_dhcp_opts):

38
tempest/api/network/test_floating_ips.py
View File

@@ -67,7 +67,7 @@ class FloatingIPTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_create_list_show_update_delete_floating_ip(self):
# Creates a floating IP
_, body = self.client.create_floatingip(
body = self.client.create_floatingip(
floating_network_id=self.ext_net_id,
port_id=self.ports[0]['id'])
created_floating_ip = body['floatingip']
@@ -82,7 +82,7 @@ class FloatingIPTestJSON(base.BaseNetworkTest):
self.assertIn(created_floating_ip['fixed_ip_address'],
[ip['ip_address'] for ip in self.ports[0]['fixed_ips']])
# Verifies the details of a floating_ip
_, floating_ip = self.client.show_floatingip(created_floating_ip['id'])
floating_ip = self.client.show_floatingip(created_floating_ip['id'])
shown_floating_ip = floating_ip['floatingip']
self.assertEqual(shown_floating_ip['id'], created_floating_ip['id'])
self.assertEqual(shown_floating_ip['floating_network_id'],
@@ -94,13 +94,13 @@ class FloatingIPTestJSON(base.BaseNetworkTest):
self.assertEqual(shown_floating_ip['port_id'], self.ports[0]['id'])
# Verify the floating ip exists in the list of all floating_ips
_, floating_ips = self.client.list_floatingips()
floating_ips = self.client.list_floatingips()
floatingip_id_list = list()
for f in floating_ips['floatingips']:
floatingip_id_list.append(f['id'])
self.assertIn(created_floating_ip['id'], floatingip_id_list)
# Associate floating IP to the other port
_, floating_ip = self.client.update_floatingip(
floating_ip = self.client.update_floatingip(
created_floating_ip['id'],
port_id=self.ports[1]['id'])
updated_floating_ip = floating_ip['floatingip']
@@ -110,7 +110,7 @@ class FloatingIPTestJSON(base.BaseNetworkTest):
self.assertEqual(updated_floating_ip['router_id'], self.router['id'])
# Disassociate floating IP from the port
_, floating_ip = self.client.update_floatingip(
floating_ip = self.client.update_floatingip(
created_floating_ip['id'],
port_id=None)
updated_floating_ip = floating_ip['floatingip']
@@ -121,21 +121,21 @@ class FloatingIPTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_floating_ip_delete_port(self):
# Create a floating IP
_, body = self.client.create_floatingip(
body = self.client.create_floatingip(
floating_network_id=self.ext_net_id)
created_floating_ip = body['floatingip']
self.addCleanup(self.client.delete_floatingip,
created_floating_ip['id'])
# Create a port
resp, port = self.client.create_port(network_id=self.network['id'])
port = self.client.create_port(network_id=self.network['id'])
created_port = port['port']
_, floating_ip = self.client.update_floatingip(
floating_ip = self.client.update_floatingip(
created_floating_ip['id'],
port_id=created_port['id'])
# Delete port
self.client.delete_port(created_port['id'])
# Verifies the details of the floating_ip
_, floating_ip = self.client.show_floatingip(created_floating_ip['id'])
floating_ip = self.client.show_floatingip(created_floating_ip['id'])
shown_floating_ip = floating_ip['floatingip']
# Confirm the fields are back to None
self.assertEqual(shown_floating_ip['id'], created_floating_ip['id'])
@@ -146,7 +146,7 @@ class FloatingIPTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_floating_ip_update_different_router(self):
# Associate a floating IP to a port on a router
_, body = self.client.create_floatingip(
body = self.client.create_floatingip(
floating_network_id=self.ext_net_id,
port_id=self.ports[1]['id'])
created_floating_ip = body['floatingip']
@@ -160,7 +160,7 @@ class FloatingIPTestJSON(base.BaseNetworkTest):
self.create_router_interface(router2['id'], subnet2['id'])
port_other_router = self.create_port(network2)
# Associate floating IP to the other port on another router
_, floating_ip = self.client.update_floatingip(
floating_ip = self.client.update_floatingip(
created_floating_ip['id'],
port_id=port_other_router['id'])
updated_floating_ip = floating_ip['floatingip']
@@ -171,7 +171,7 @@ class FloatingIPTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_create_floating_ip_specifying_a_fixed_ip_address(self):
_, body = self.client.create_floatingip(
body = self.client.create_floatingip(
floating_network_id=self.ext_net_id,
port_id=self.ports[1]['id'],
fixed_ip_address=self.ports[1]['fixed_ips'][0]['ip_address'])
@@ -181,7 +181,7 @@ class FloatingIPTestJSON(base.BaseNetworkTest):
self.assertIsNotNone(created_floating_ip['id'])
self.assertEqual(created_floating_ip['fixed_ip_address'],
self.ports[1]['fixed_ips'][0]['ip_address'])
_, floating_ip = self.client.update_floatingip(
floating_ip = self.client.update_floatingip(
created_floating_ip['id'],
port_id=None)
self.assertIsNone(floating_ip['floatingip']['port_id'])
@@ -193,12 +193,12 @@ class FloatingIPTestJSON(base.BaseNetworkTest):
list_ips = [str(ip) for ip in ips[-3:-1]]
fixed_ips = [{'ip_address': list_ips[0]}, {'ip_address': list_ips[1]}]
# Create port
_, body = self.client.create_port(network_id=self.network['id'],
fixed_ips=fixed_ips)
body = self.client.create_port(network_id=self.network['id'],
fixed_ips=fixed_ips)
port = body['port']
self.addCleanup(self.client.delete_port, port['id'])
# Create floating ip
_, body = self.client.create_floatingip(
body = self.client.create_floatingip(
floating_network_id=self.ext_net_id,
port_id=port['id'],
fixed_ip_address=list_ips[0])
@@ -207,9 +207,9 @@ class FloatingIPTestJSON(base.BaseNetworkTest):
self.assertIsNotNone(floating_ip['id'])
self.assertEqual(floating_ip['fixed_ip_address'], list_ips[0])
# Update floating ip
_, body = self.client.update_floatingip(floating_ip['id'],
port_id=port['id'],
fixed_ip_address=list_ips[1])
body = self.client.update_floatingip(floating_ip['id'],
port_id=port['id'],
fixed_ip_address=list_ips[1])
update_floating_ip = body['floatingip']
self.assertEqual(update_floating_ip['fixed_ip_address'],
list_ips[1])

52
tempest/api/network/test_fwaas_extensions.py
View File

@@ -88,7 +88,7 @@ class FWaaSExtensionTestJSON(base.BaseNetworkTest):
target_states = ('ACTIVE', 'CREATED')
def _wait():
_, firewall = self.client.show_firewall(fw_id)
firewall = self.client.show_firewall(fw_id)
firewall = firewall['firewall']
return firewall['status'] in target_states
@@ -100,7 +100,7 @@ class FWaaSExtensionTestJSON(base.BaseNetworkTest):
def test_list_firewall_rules(self):
# List firewall rules
_, fw_rules = self.client.list_firewall_rules()
fw_rules = self.client.list_firewall_rules()
fw_rules = fw_rules['firewall_rules']
self.assertIn((self.fw_rule['id'],
self.fw_rule['name'],
@@ -117,32 +117,32 @@ class FWaaSExtensionTestJSON(base.BaseNetworkTest):
def test_create_update_delete_firewall_rule(self):
# Create firewall rule
_, body = self.client.create_firewall_rule(
body = self.client.create_firewall_rule(
name=data_utils.rand_name("fw-rule"),
action="allow",
protocol="tcp")
fw_rule_id = body['firewall_rule']['id']
# Update firewall rule
_, body = self.client.update_firewall_rule(fw_rule_id,
shared=True)
body = self.client.update_firewall_rule(fw_rule_id,
shared=True)
self.assertTrue(body["firewall_rule"]['shared'])
# Delete firewall rule
self.client.delete_firewall_rule(fw_rule_id)
# Confirm deletion
resp, fw_rules = self.client.list_firewall_rules()
fw_rules = self.client.list_firewall_rules()
self.assertNotIn(fw_rule_id,
[m['id'] for m in fw_rules['firewall_rules']])
def test_show_firewall_rule(self):
# show a created firewall rule
_, fw_rule = self.client.show_firewall_rule(self.fw_rule['id'])
fw_rule = self.client.show_firewall_rule(self.fw_rule['id'])
for key, value in fw_rule['firewall_rule'].iteritems():
self.assertEqual(self.fw_rule[key], value)
def test_list_firewall_policies(self):
_, fw_policies = self.client.list_firewall_policies()
fw_policies = self.client.list_firewall_policies()
fw_policies = fw_policies['firewall_policies']
self.assertIn((self.fw_policy['id'],
self.fw_policy['name'],
@@ -153,15 +153,15 @@ class FWaaSExtensionTestJSON(base.BaseNetworkTest):
def test_create_update_delete_firewall_policy(self):
# Create firewall policy
_, body = self.client.create_firewall_policy(
body = self.client.create_firewall_policy(
name=data_utils.rand_name("fw-policy"))
fw_policy_id = body['firewall_policy']['id']
self.addCleanup(self._try_delete_policy, fw_policy_id)
# Update firewall policy
_, body = self.client.update_firewall_policy(fw_policy_id,
shared=True,
name="updated_policy")
body = self.client.update_firewall_policy(fw_policy_id,
shared=True,
name="updated_policy")
updated_fw_policy = body["firewall_policy"]
self.assertTrue(updated_fw_policy['shared'])
self.assertEqual("updated_policy", updated_fw_policy['name'])
@@ -169,13 +169,13 @@ class FWaaSExtensionTestJSON(base.BaseNetworkTest):
# Delete firewall policy
self.client.delete_firewall_policy(fw_policy_id)
# Confirm deletion
resp, fw_policies = self.client.list_firewall_policies()
fw_policies = self.client.list_firewall_policies()
fw_policies = fw_policies['firewall_policies']
self.assertNotIn(fw_policy_id, [m['id'] for m in fw_policies])
def test_show_firewall_policy(self):
# show a created firewall policy
_, fw_policy = self.client.show_firewall_policy(self.fw_policy['id'])
fw_policy = self.client.show_firewall_policy(self.fw_policy['id'])
fw_policy = fw_policy['firewall_policy']
for key, value in fw_policy.iteritems():
self.assertEqual(self.fw_policy[key], value)
@@ -191,7 +191,7 @@ class FWaaSExtensionTestJSON(base.BaseNetworkTest):
router['id'], subnet['id'])
# Create firewall
_, body = self.client.create_firewall(
body = self.client.create_firewall(
name=data_utils.rand_name("firewall"),
firewall_policy_id=self.fw_policy['id'])
created_firewall = body['firewall']
@@ -202,7 +202,7 @@ class FWaaSExtensionTestJSON(base.BaseNetworkTest):
self._wait_until_ready(firewall_id)
# show a created firewall
_, firewall = self.client.show_firewall(firewall_id)
firewall = self.client.show_firewall(firewall_id)
firewall = firewall['firewall']
for key, value in firewall.iteritems():
@@ -211,7 +211,7 @@ class FWaaSExtensionTestJSON(base.BaseNetworkTest):
self.assertEqual(created_firewall[key], value)
# list firewall
_, firewalls = self.client.list_firewalls()
firewalls = self.client.list_firewalls()
firewalls = firewalls['firewalls']
self.assertIn((created_firewall['id'],
created_firewall['name'],
@@ -226,14 +226,14 @@ class FWaaSExtensionTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_firewall_rule_insertion_position_removal_rule_from_policy(self):
# Create firewall rule
resp, body = self.client.create_firewall_rule(
body = self.client.create_firewall_rule(
name=data_utils.rand_name("fw-rule"),
action="allow",
protocol="tcp")
fw_rule_id1 = body['firewall_rule']['id']
self.addCleanup(self._try_delete_rule, fw_rule_id1)
# Create firewall policy
_, body = self.client.create_firewall_policy(
body = self.client.create_firewall_policy(
name=data_utils.rand_name("fw-policy"))
fw_policy_id = body['firewall_policy']['id']
self.addCleanup(self._try_delete_policy, fw_policy_id)
@@ -245,7 +245,7 @@ class FWaaSExtensionTestJSON(base.BaseNetworkTest):
# Verify insertion of rule in policy
self.assertIn(fw_rule_id1, self._get_list_fw_rule_ids(fw_policy_id))
# Create another firewall rule
_, body = self.client.create_firewall_rule(
body = self.client.create_firewall_rule(
name=data_utils.rand_name("fw-rule"),
action="allow",
protocol="icmp")
@@ -257,7 +257,7 @@ class FWaaSExtensionTestJSON(base.BaseNetworkTest):
fw_policy_id, fw_rule_id2, fw_rule_id1, '')
# Verify the posiition of rule after insertion
_, fw_rule = self.client.show_firewall_rule(
fw_rule = self.client.show_firewall_rule(
fw_rule_id2)
self.assertEqual(int(fw_rule['firewall_rule']['position']), 2)
@@ -268,7 +268,7 @@ class FWaaSExtensionTestJSON(base.BaseNetworkTest):
self.client.insert_firewall_rule_in_policy(
fw_policy_id, fw_rule_id2, '', fw_rule_id1)
# Verify the posiition of rule after insertion
_, fw_rule = self.client.show_firewall_rule(
fw_rule = self.client.show_firewall_rule(
fw_rule_id2)
self.assertEqual(int(fw_rule['firewall_rule']['position']), 1)
# Remove rule from the firewall policy
@@ -285,21 +285,21 @@ class FWaaSExtensionTestJSON(base.BaseNetworkTest):
self.assertNotIn(fw_rule_id1, self._get_list_fw_rule_ids(fw_policy_id))
def _get_list_fw_rule_ids(self, fw_policy_id):
_, fw_policy = self.client.show_firewall_policy(
fw_policy = self.client.show_firewall_policy(
fw_policy_id)
return [ruleid for ruleid in fw_policy['firewall_policy']
['firewall_rules']]
def test_update_firewall_policy_audited_attribute(self):
# Create firewall rule
_, body = self.client.create_firewall_rule(
body = self.client.create_firewall_rule(
name=data_utils.rand_name("fw-rule"),
action="allow",
protocol="icmp")
fw_rule_id = body['firewall_rule']['id']
self.addCleanup(self._try_delete_rule, fw_rule_id)
# Create firewall policy
_, body = self.client.create_firewall_policy(
body = self.client.create_firewall_policy(
name=data_utils.rand_name('fw-policy'))
fw_policy_id = body['firewall_policy']['id']
self.addCleanup(self._try_delete_policy, fw_policy_id)
@@ -310,6 +310,6 @@ class FWaaSExtensionTestJSON(base.BaseNetworkTest):
# Insert Firewall rule to firewall policy
self.client.insert_firewall_rule_in_policy(
fw_policy_id, fw_rule_id, '', '')
_, body = self.client.show_firewall_policy(
body = self.client.show_firewall_policy(
fw_policy_id)
self.assertFalse(body['firewall_policy']['audited'])

172
tempest/api/network/test_load_balancer.py
View File

@@ -68,31 +68,31 @@ class LoadBalancerTestJSON(base.BaseNetworkTest):
delete_obj = getattr(self.client, 'delete_' + obj_name)
list_objs = getattr(self.client, 'list_' + obj_name + 's')
_, body = create_obj(**kwargs)
body = create_obj(**kwargs)
obj = body[obj_name]
self.addCleanup(delete_obj, obj['id'])
for key, value in obj.iteritems():
# It is not relevant to filter by all arguments. That is why
# there is a list of attr to except
if key not in attr_exceptions:
_, body = list_objs(**{key: value})
body = list_objs(**{key: value})
objs = [v[key] for v in body[obj_name + 's']]
self.assertIn(value, objs)
@test.attr(type='smoke')
def test_list_vips(self):
# Verify the vIP exists in the list of all vIPs
_, body = self.client.list_vips()
body = self.client.list_vips()
vips = body['vips']
self.assertIn(self.vip['id'], [v['id'] for v in vips])
@test.attr(type='smoke')
def test_list_vips_with_filter(self):
name = data_utils.rand_name('vip-')
_, body = self.client.create_pool(name=data_utils.rand_name("pool-"),
lb_method="ROUND_ROBIN",
protocol="HTTPS",
subnet_id=self.subnet['id'])
body = self.client.create_pool(name=data_utils.rand_name("pool-"),
lb_method="ROUND_ROBIN",
protocol="HTTPS",
subnet_id=self.subnet['id'])
pool = body['pool']
self.addCleanup(self.client.delete_pool, pool['id'])
attr_exceptions = ['status', 'session_persistence',
@@ -108,22 +108,22 @@ class LoadBalancerTestJSON(base.BaseNetworkTest):
# Creates a vip
name = data_utils.rand_name('vip-')
address = self.subnet['allocation_pools'][0]['end']
resp, body = self.client.create_pool(
body = self.client.create_pool(
name=data_utils.rand_name("pool-"),
lb_method='ROUND_ROBIN',
protocol='HTTP',
subnet_id=self.subnet['id'])
pool = body['pool']
_, body = self.client.create_vip(name=name,
protocol="HTTP",
protocol_port=80,
subnet_id=self.subnet['id'],
pool_id=pool['id'],
address=address)
body = self.client.create_vip(name=name,
protocol="HTTP",
protocol_port=80,
subnet_id=self.subnet['id'],
pool_id=pool['id'],
address=address)
vip = body['vip']
vip_id = vip['id']
# Confirm VIP's address correctness with a show
_, body = self.client.show_vip(vip_id)
body = self.client.show_vip(vip_id)
vip = body['vip']
self.assertEqual(address, vip['address'])
# Verification of vip update
@@ -132,12 +132,12 @@ class LoadBalancerTestJSON(base.BaseNetworkTest):
persistence_type = "HTTP_COOKIE"
update_data = {"session_persistence": {
"type": persistence_type}}
_, body = self.client.update_vip(vip_id,
name=new_name,
description=new_description,
connection_limit=10,
admin_state_up=False,
**update_data)
body = self.client.update_vip(vip_id,
name=new_name,
description=new_description,
connection_limit=10,
admin_state_up=False,
**update_data)
updated_vip = body['vip']
self.assertEqual(new_name, updated_vip['name'])
self.assertEqual(new_description, updated_vip['description'])
@@ -149,10 +149,10 @@ class LoadBalancerTestJSON(base.BaseNetworkTest):
self.client.wait_for_resource_deletion('vip', vip['id'])
# Verification of pool update
new_name = "New_pool"
_, body = self.client.update_pool(pool['id'],
name=new_name,
description="new_description",
lb_method='LEAST_CONNECTIONS')
body = self.client.update_pool(pool['id'],
name=new_name,
description="new_description",
lb_method='LEAST_CONNECTIONS')
updated_pool = body['pool']
self.assertEqual(new_name, updated_pool['name'])
self.assertEqual('new_description', updated_pool['description'])
@@ -162,7 +162,7 @@ class LoadBalancerTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_show_vip(self):
# Verifies the details of a vip
_, body = self.client.show_vip(self.vip['id'])
body = self.client.show_vip(self.vip['id'])
vip = body['vip']
for key, value in vip.iteritems():
# 'status' should not be confirmed in api tests
@@ -172,14 +172,14 @@ class LoadBalancerTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_show_pool(self):
# Here we need to new pool without any dependence with vips
_, body = self.client.create_pool(name=data_utils.rand_name("pool-"),
lb_method='ROUND_ROBIN',
protocol='HTTP',
subnet_id=self.subnet['id'])
body = self.client.create_pool(name=data_utils.rand_name("pool-"),
lb_method='ROUND_ROBIN',
protocol='HTTP',
subnet_id=self.subnet['id'])
pool = body['pool']
self.addCleanup(self.client.delete_pool, pool['id'])
# Verifies the details of a pool
_, body = self.client.show_pool(pool['id'])
body = self.client.show_pool(pool['id'])
shown_pool = body['pool']
for key, value in pool.iteritems():
# 'status' should not be confirmed in api tests
@@ -189,7 +189,7 @@ class LoadBalancerTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_list_pools(self):
# Verify the pool exists in the list of all pools
_, body = self.client.list_pools()
body = self.client.list_pools()
pools = body['pools']
self.assertIn(self.pool['id'], [p['id'] for p in pools])
@@ -207,7 +207,7 @@ class LoadBalancerTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_list_members(self):
# Verify the member exists in the list of all members
_, body = self.client.list_members()
body = self.client.list_members()
members = body['members']
self.assertIn(self.member['id'], [m['id'] for m in members])
@@ -222,13 +222,13 @@ class LoadBalancerTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_create_update_delete_member(self):
# Creates a member
_, body = self.client.create_member(address=self.member_address,
protocol_port=80,
pool_id=self.pool['id'])
body = self.client.create_member(address=self.member_address,
protocol_port=80,
pool_id=self.pool['id'])
member = body['member']
# Verification of member update
_, body = self.client.update_member(member['id'],
admin_state_up=False)
body = self.client.update_member(member['id'],
admin_state_up=False)
updated_member = body['member']
self.assertFalse(updated_member['admin_state_up'])
# Verification of member delete
@@ -237,7 +237,7 @@ class LoadBalancerTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_show_member(self):
# Verifies the details of a member
_, body = self.client.show_member(self.member['id'])
body = self.client.show_member(self.member['id'])
member = body['member']
for key, value in member.iteritems():
# 'status' should not be confirmed in api tests
@@ -247,7 +247,7 @@ class LoadBalancerTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_list_health_monitors(self):
# Verify the health monitor exists in the list of all health monitors
_, body = self.client.list_health_monitors()
body = self.client.list_health_monitors()
health_monitors = body['health_monitors']
self.assertIn(self.health_monitor['id'],
[h['id'] for h in health_monitors])
@@ -262,27 +262,27 @@ class LoadBalancerTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_create_update_delete_health_monitor(self):
# Creates a health_monitor
_, body = self.client.create_health_monitor(delay=4,
max_retries=3,
type="TCP",
timeout=1)
body = self.client.create_health_monitor(delay=4,
max_retries=3,
type="TCP",
timeout=1)
health_monitor = body['health_monitor']
# Verification of health_monitor update
_, body = (self.client.update_health_monitor
(health_monitor['id'],
admin_state_up=False))
body = (self.client.update_health_monitor
(health_monitor['id'],
admin_state_up=False))
updated_health_monitor = body['health_monitor']
self.assertFalse(updated_health_monitor['admin_state_up'])
# Verification of health_monitor delete
_, body = self.client.delete_health_monitor(health_monitor['id'])
body = self.client.delete_health_monitor(health_monitor['id'])
@test.attr(type='smoke')
def test_create_health_monitor_http_type(self):
hm_type = "HTTP"
_, body = self.client.create_health_monitor(delay=4,
max_retries=3,
type=hm_type,
timeout=1)
body = self.client.create_health_monitor(delay=4,
max_retries=3,
type=hm_type,
timeout=1)
health_monitor = body['health_monitor']
self.addCleanup(self.client.delete_health_monitor,
health_monitor['id'])
@@ -290,18 +290,18 @@ class LoadBalancerTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_update_health_monitor_http_method(self):
_, body = self.client.create_health_monitor(delay=4,
max_retries=3,
type="HTTP",
timeout=1)
body = self.client.create_health_monitor(delay=4,
max_retries=3,
type="HTTP",
timeout=1)
health_monitor = body['health_monitor']
self.addCleanup(self.client.delete_health_monitor,
health_monitor['id'])
_, body = (self.client.update_health_monitor
(health_monitor['id'],
http_method="POST",
url_path="/home/user",
expected_codes="290"))
body = (self.client.update_health_monitor
(health_monitor['id'],
http_method="POST",
url_path="/home/user",
expected_codes="290"))
updated_health_monitor = body['health_monitor']
self.assertEqual("POST", updated_health_monitor['http_method'])
self.assertEqual("/home/user", updated_health_monitor['url_path'])
@@ -310,7 +310,7 @@ class LoadBalancerTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_show_health_monitor(self):
# Verifies the details of a health_monitor
_, body = self.client.show_health_monitor(self.health_monitor['id'])
body = self.client.show_health_monitor(self.health_monitor['id'])
health_monitor = body['health_monitor']
for key, value in health_monitor.iteritems():
# 'status' should not be confirmed in api tests
@@ -320,12 +320,12 @@ class LoadBalancerTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_associate_disassociate_health_monitor_with_pool(self):
# Verify that a health monitor can be associated with a pool
_, body = (self.client.associate_health_monitor_with_pool
(self.health_monitor['id'], self.pool['id']))
resp, body = self.client.show_health_monitor(
self.client.associate_health_monitor_with_pool(
self.health_monitor['id'], self.pool['id'])
body = self.client.show_health_monitor(
self.health_monitor['id'])
health_monitor = body['health_monitor']
resp, body = self.client.show_pool(self.pool['id'])
body = self.client.show_pool(self.pool['id'])
pool = body['pool']
self.assertIn(pool['id'],
[p['pool_id'] for p in health_monitor['pools']])
@@ -333,9 +333,9 @@ class LoadBalancerTestJSON(base.BaseNetworkTest):
# Verify that a health monitor can be disassociated from a pool
(self.client.disassociate_health_monitor_with_pool
(self.health_monitor['id'], self.pool['id']))
_, body = self.client.show_pool(self.pool['id'])
body = self.client.show_pool(self.pool['id'])
pool = body['pool']
resp, body = self.client.show_health_monitor(
body = self.client.show_health_monitor(
self.health_monitor['id'])
health_monitor = body['health_monitor']
self.assertNotIn(health_monitor['id'], pool['health_monitors'])
@@ -345,7 +345,7 @@ class LoadBalancerTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_get_lb_pool_stats(self):
# Verify the details of pool stats
_, body = self.client.list_lb_pool_stats(self.pool['id'])
body = self.client.list_lb_pool_stats(self.pool['id'])
stats = body['stats']
self.assertIn("bytes_in", stats)
self.assertIn("total_connections", stats)
@@ -358,10 +358,10 @@ class LoadBalancerTestJSON(base.BaseNetworkTest):
(self.health_monitor['id'], self.pool['id']))
self.client.update_health_monitor(
self.health_monitor['id'], admin_state_up=False)
_, body = self.client.show_pool(self.pool['id'])
body = self.client.show_pool(self.pool['id'])
health_monitors = body['pool']['health_monitors']
for health_monitor_id in health_monitors:
_, body = self.client.show_health_monitor(health_monitor_id)
body = self.client.show_health_monitor(health_monitor_id)
self.assertFalse(body['health_monitor']['admin_state_up'])
(self.client.disassociate_health_monitor_with_pool
(self.health_monitor['id'], self.pool['id']))
@@ -370,25 +370,25 @@ class LoadBalancerTestJSON(base.BaseNetworkTest):
def test_update_admin_state_up_of_pool(self):
self.client.update_pool(self.pool['id'],
admin_state_up=False)
_, body = self.client.show_pool(self.pool['id'])
body = self.client.show_pool(self.pool['id'])
pool = body['pool']
self.assertFalse(pool['admin_state_up'])
@test.attr(type='smoke')
def test_show_vip_associated_with_pool(self):
_, body = self.client.show_pool(self.pool['id'])
body = self.client.show_pool(self.pool['id'])
pool = body['pool']
_, body = self.client.show_vip(pool['vip_id'])
body = self.client.show_vip(pool['vip_id'])
vip = body['vip']
self.assertEqual(self.vip['name'], vip['name'])
self.assertEqual(self.vip['id'], vip['id'])
@test.attr(type='smoke')
def test_show_members_associated_with_pool(self):
_, body = self.client.show_pool(self.pool['id'])
body = self.client.show_pool(self.pool['id'])
members = body['pool']['members']
for member_id in members:
_, body = self.client.show_member(member_id)
body = self.client.show_member(member_id)
self.assertIsNotNone(body['member']['status'])
self.assertEqual(member_id, body['member']['id'])
self.assertIsNotNone(body['member']['admin_state_up'])
@@ -396,28 +396,28 @@ class LoadBalancerTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_update_pool_related_to_member(self):
# Create new pool
_, body = self.client.create_pool(name=data_utils.rand_name("pool-"),
lb_method='ROUND_ROBIN',
protocol='HTTP',
subnet_id=self.subnet['id'])
body = self.client.create_pool(name=data_utils.rand_name("pool-"),
lb_method='ROUND_ROBIN',
protocol='HTTP',
subnet_id=self.subnet['id'])
new_pool = body['pool']
self.addCleanup(self.client.delete_pool, new_pool['id'])
# Update member with new pool's id
_, body = self.client.update_member(self.member['id'],
pool_id=new_pool['id'])
body = self.client.update_member(self.member['id'],
pool_id=new_pool['id'])
# Confirm with show that pool_id change
resp, body = self.client.show_member(self.member['id'])
body = self.client.show_member(self.member['id'])
member = body['member']
self.assertEqual(member['pool_id'], new_pool['id'])
# Update member with old pool id, this is needed for clean up
_, body = self.client.update_member(self.member['id'],
pool_id=self.pool['id'])
body = self.client.update_member(self.member['id'],
pool_id=self.pool['id'])
@test.attr(type='smoke')
def test_update_member_weight(self):
self.client.update_member(self.member['id'],
weight=2)
_, body = self.client.show_member(self.member['id'])
body = self.client.show_member(self.member['id'])
member = body['member']
self.assertEqual(2, member['weight'])

44
tempest/api/network/test_metering_extensions.py
View File

@@ -52,25 +52,24 @@ class MeteringTestJSON(base.BaseAdminNetworkTest):
def _delete_metering_label(self, metering_label_id):
# Deletes a label and verifies if it is deleted or not
_, body = self.admin_client.delete_metering_label(metering_label_id)
self.admin_client.delete_metering_label(metering_label_id)
# Asserting that the label is not found in list after deletion
resp, labels = (self.admin_client.list_metering_labels(
id=metering_label_id))
labels = self.admin_client.list_metering_labels(id=metering_label_id)
self.assertEqual(len(labels['metering_labels']), 0)
def _delete_metering_label_rule(self, metering_label_rule_id):
# Deletes a rule and verifies if it is deleted or not
_, body = (self.admin_client.delete_metering_label_rule(
metering_label_rule_id))
self.admin_client.delete_metering_label_rule(
metering_label_rule_id)
# Asserting that the rule is not found in list after deletion
resp, rules = (self.admin_client.list_metering_label_rules(
id=metering_label_rule_id))
rules = (self.admin_client.list_metering_label_rules(
id=metering_label_rule_id))
self.assertEqual(len(rules['metering_label_rules']), 0)
@test.attr(type='smoke')
def test_list_metering_labels(self):
# Verify label filtering
_, body = self.admin_client.list_metering_labels(id=33)
body = self.admin_client.list_metering_labels(id=33)
metering_labels = body['metering_labels']
self.assertEqual(0, len(metering_labels))
@@ -79,22 +78,21 @@ class MeteringTestJSON(base.BaseAdminNetworkTest):
# Creates a label
name = data_utils.rand_name('metering-label-')
description = "label created by tempest"
_, body = (self.admin_client.create_metering_label(name=name,
description=description))
body = self.admin_client.create_metering_label(name=name,
description=description)
metering_label = body['metering_label']
self.addCleanup(self._delete_metering_label,
metering_label['id'])
# Assert whether created labels are found in labels list or fail
# if created labels are not found in labels list
resp, labels = (self.admin_client.list_metering_labels(
id=metering_label['id']))
labels = (self.admin_client.list_metering_labels(
id=metering_label['id']))
self.assertEqual(len(labels['metering_labels']), 1)
@test.attr(type='smoke')
def test_show_metering_label(self):
# Verifies the details of a label
_, body = (self.admin_client.show_metering_label(
self.metering_label['id']))
body = self.admin_client.show_metering_label(self.metering_label['id'])
metering_label = body['metering_label']
self.assertEqual(self.metering_label['id'], metering_label['id'])
self.assertEqual(self.metering_label['tenant_id'],
@@ -106,7 +104,7 @@ class MeteringTestJSON(base.BaseAdminNetworkTest):
@test.attr(type='smoke')
def test_list_metering_label_rules(self):
# Verify rule filtering
_, body = self.admin_client.list_metering_label_rules(id=33)
body = self.admin_client.list_metering_label_rules(id=33)
metering_label_rules = body['metering_label_rules']
self.assertEqual(0, len(metering_label_rules))
@@ -115,24 +113,24 @@ class MeteringTestJSON(base.BaseAdminNetworkTest):
# Creates a rule
remote_ip_prefix = ("10.0.1.0/24" if self._ip_version == 4
else "fd03::/64")
_, body = (self.admin_client.create_metering_label_rule(
remote_ip_prefix=remote_ip_prefix,
direction="ingress",
metering_label_id=self.metering_label['id']))
body = (self.admin_client.create_metering_label_rule(
remote_ip_prefix=remote_ip_prefix,
direction="ingress",
metering_label_id=self.metering_label['id']))
metering_label_rule = body['metering_label_rule']
self.addCleanup(self._delete_metering_label_rule,
metering_label_rule['id'])
# Assert whether created rules are found in rules list or fail
# if created rules are not found in rules list
resp, rules = (self.admin_client.list_metering_label_rules(
id=metering_label_rule['id']))
rules = (self.admin_client.list_metering_label_rules(
id=metering_label_rule['id']))
self.assertEqual(len(rules['metering_label_rules']), 1)
@test.attr(type='smoke')
def test_show_metering_label_rule(self):
# Verifies the details of a rule
_, body = (self.admin_client.show_metering_label_rule(
self.metering_label_rule['id']))
body = (self.admin_client.show_metering_label_rule(
self.metering_label_rule['id']))
metering_label_rule = body['metering_label_rule']
self.assertEqual(self.metering_label_rule['id'],
metering_label_rule['id'])

56
tempest/api/network/test_networks.py
View File

@@ -160,7 +160,7 @@ class NetworksTestJSON(base.BaseNetworkTest):
self.assertEqual('ACTIVE', network['status'])
# Verify network update
new_name = "New_network"
_, body = self.client.update_network(net_id, name=new_name)
body = self.client.update_network(net_id, name=new_name)
updated_net = body['network']
self.assertEqual(updated_net['name'], new_name)
# Find a cidr that is not in use yet and create a subnet with it
@@ -168,14 +168,14 @@ class NetworksTestJSON(base.BaseNetworkTest):
subnet_id = subnet['id']
# Verify subnet update
new_name = "New_subnet"
_, body = self.client.update_subnet(subnet_id, name=new_name)
body = self.client.update_subnet(subnet_id, name=new_name)
updated_subnet = body['subnet']
self.assertEqual(updated_subnet['name'], new_name)
@test.attr(type='smoke')
def test_show_network(self):
# Verify the details of a network
_, body = self.client.show_network(self.network['id'])
body = self.client.show_network(self.network['id'])
network = body['network']
for key in ['id', 'name']:
self.assertEqual(network[key], self.network[key])
@@ -184,8 +184,8 @@ class NetworksTestJSON(base.BaseNetworkTest):
def test_show_network_fields(self):
# Verify specific fields of a network
fields = ['id', 'name']
_, body = self.client.show_network(self.network['id'],
fields=fields)
body = self.client.show_network(self.network['id'],
fields=fields)
network = body['network']
self.assertEqual(sorted(network.keys()), sorted(fields))
for field_name in fields:
@@ -194,7 +194,7 @@ class NetworksTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_list_networks(self):
# Verify the network exists in the list of all networks
_, body = self.client.list_networks()
body = self.client.list_networks()
networks = [network['id'] for network in body['networks']
if network['id'] == self.network['id']]
self.assertNotEmpty(networks, "Created network not found in the list")
@@ -203,7 +203,7 @@ class NetworksTestJSON(base.BaseNetworkTest):
def test_list_networks_fields(self):
# Verify specific fields of the networks
fields = ['id', 'name']
_, body = self.client.list_networks(fields=fields)
body = self.client.list_networks(fields=fields)
networks = body['networks']
self.assertNotEmpty(networks, "Network list returned is empty")
for network in networks:
@@ -212,7 +212,7 @@ class NetworksTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_show_subnet(self):
# Verify the details of a subnet
_, body = self.client.show_subnet(self.subnet['id'])
body = self.client.show_subnet(self.subnet['id'])
subnet = body['subnet']
self.assertNotEmpty(subnet, "Subnet returned has no fields")
for key in ['id', 'cidr']:
@@ -223,8 +223,8 @@ class NetworksTestJSON(base.BaseNetworkTest):
def test_show_subnet_fields(self):
# Verify specific fields of a subnet
fields = ['id', 'network_id']
_, body = self.client.show_subnet(self.subnet['id'],
fields=fields)
body = self.client.show_subnet(self.subnet['id'],
fields=fields)
subnet = body['subnet']
self.assertEqual(sorted(subnet.keys()), sorted(fields))
for field_name in fields:
@@ -233,7 +233,7 @@ class NetworksTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_list_subnets(self):
# Verify the subnet exists in the list of all subnets
_, body = self.client.list_subnets()
body = self.client.list_subnets()
subnets = [subnet['id'] for subnet in body['subnets']
if subnet['id'] == self.subnet['id']]
self.assertNotEmpty(subnets, "Created subnet not found in the list")
@@ -242,7 +242,7 @@ class NetworksTestJSON(base.BaseNetworkTest):
def test_list_subnets_fields(self):
# Verify specific fields of subnets
fields = ['id', 'network_id']
_, body = self.client.list_subnets(fields=fields)
body = self.client.list_subnets(fields=fields)
subnets = body['subnets']
self.assertNotEmpty(subnets, "Subnet list returned is empty")
for subnet in subnets:
@@ -260,7 +260,7 @@ class NetworksTestJSON(base.BaseNetworkTest):
def test_delete_network_with_subnet(self):
# Creates a network
name = data_utils.rand_name('network-')
_, body = self.client.create_network(name=name)
body = self.client.create_network(name=name)
network = body['network']
net_id = network['id']
self.addCleanup(self._try_delete_network, net_id)
@@ -270,7 +270,7 @@ class NetworksTestJSON(base.BaseNetworkTest):
subnet_id = subnet['id']
# Delete network while the subnet still exists
_, body = self.client.delete_network(net_id)
body = self.client.delete_network(net_id)
# Verify that the subnet got automatically deleted.
self.assertRaises(exceptions.NotFound, self.client.show_subnet,
@@ -331,8 +331,8 @@ class NetworksTestJSON(base.BaseNetworkTest):
'gateway_ip': new_gateway, 'enable_dhcp': True}
new_name = "New_subnet"
_, body = self.client.update_subnet(subnet_id, name=new_name,
**kwargs)
body = self.client.update_subnet(subnet_id, name=new_name,
**kwargs)
updated_subnet = body['subnet']
kwargs['name'] = new_name
self.assertEqual(sorted(updated_subnet['dns_nameservers']),
@@ -350,7 +350,7 @@ class NetworksTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_external_network_visibility(self):
"""Verifies user can see external networks but not subnets."""
_, body = self.client.list_networks(**{'router:external': True})
body = self.client.list_networks(**{'router:external': True})
networks = [network['id'] for network in body['networks']]
self.assertNotEmpty(networks, "No external networks found")
@@ -364,7 +364,7 @@ class NetworksTestJSON(base.BaseNetworkTest):
# subnets_iter is a list (iterator) of lists. This flattens it to a
# list of UUIDs
public_subnets_iter = itertools.chain(*subnets_iter)
_, body = self.client.list_subnets()
body = self.client.list_subnets()
subnets = [sub['id'] for sub in body['subnets']
if sub['id'] in public_subnets_iter]
self.assertEmpty(subnets, "Public subnets visible")
@@ -396,7 +396,7 @@ class BulkNetworkOpsTestJSON(base.BaseNetworkTest):
for n in created_networks:
self.client.delete_network(n['id'])
# Asserting that the networks are not found in the list after deletion
resp, body = self.client.list_networks()
body = self.client.list_networks()
networks_list = [network['id'] for network in body['networks']]
for n in created_networks:
self.assertNotIn(n['id'], networks_list)
@@ -405,7 +405,7 @@ class BulkNetworkOpsTestJSON(base.BaseNetworkTest):
for n in created_subnets:
self.client.delete_subnet(n['id'])
# Asserting that the subnets are not found in the list after deletion
resp, body = self.client.list_subnets()
body = self.client.list_subnets()
subnets_list = [subnet['id'] for subnet in body['subnets']]
for n in created_subnets:
self.assertNotIn(n['id'], subnets_list)
@@ -414,7 +414,7 @@ class BulkNetworkOpsTestJSON(base.BaseNetworkTest):
for n in created_ports:
self.client.delete_port(n['id'])
# Asserting that the ports are not found in the list after deletion
resp, body = self.client.list_ports()
body = self.client.list_ports()
ports_list = [port['id'] for port in body['ports']]
for n in created_ports:
self.assertNotIn(n['id'], ports_list)
@@ -424,11 +424,11 @@ class BulkNetworkOpsTestJSON(base.BaseNetworkTest):
# Creates 2 networks in one request
network_names = [data_utils.rand_name('network-'),
data_utils.rand_name('network-')]
_, body = self.client.create_bulk_network(network_names)
body = self.client.create_bulk_network(network_names)
created_networks = body['networks']
self.addCleanup(self._delete_networks, created_networks)
# Asserting that the networks are found in the list after creation
resp, body = self.client.list_networks()
body = self.client.list_networks()
networks_list = [network['id'] for network in body['networks']]
for n in created_networks:
self.assertIsNotNone(n['id'])
@@ -455,11 +455,11 @@ class BulkNetworkOpsTestJSON(base.BaseNetworkTest):
}
subnets_list.append(p1)
del subnets_list[1]['name']
_, body = self.client.create_bulk_subnet(subnets_list)
body = self.client.create_bulk_subnet(subnets_list)
created_subnets = body['subnets']
self.addCleanup(self._delete_subnets, created_subnets)
# Asserting that the subnets are found in the list after creation
resp, body = self.client.list_subnets()
body = self.client.list_subnets()
subnets_list = [subnet['id'] for subnet in body['subnets']]
for n in created_subnets:
self.assertIsNotNone(n['id'])
@@ -480,11 +480,11 @@ class BulkNetworkOpsTestJSON(base.BaseNetworkTest):
}
port_list.append(p1)
del port_list[1]['name']
_, body = self.client.create_bulk_port(port_list)
body = self.client.create_bulk_port(port_list)
created_ports = body['ports']
self.addCleanup(self._delete_ports, created_ports)
# Asserting that the ports are found in the list after creation
resp, body = self.client.list_ports()
body = self.client.list_ports()
ports_list = [port['id'] for port in body['ports']]
for n in created_ports:
self.assertIsNotNone(n['id'])
@@ -534,7 +534,7 @@ class NetworksIpV6TestJSON(NetworksTestJSON):
# Verifies Subnet GW is None in IPv4
self.assertEqual(subnet2['gateway_ip'], None)
# Verifies all 2 subnets in the same network
_, body = self.client.list_subnets()
body = self.client.list_subnets()
subnets = [sub['id'] for sub in body['subnets']
if sub['network_id'] == network['id']]
test_subnet_ids = [sub['id'] for sub in (subnet1, subnet2)]

57
tempest/api/network/test_ports.py
View File

@@ -45,23 +45,23 @@ class PortsTestJSON(base.BaseNetworkTest):
def _delete_port(self, port_id):
self.client.delete_port(port_id)
_, body = self.client.list_ports()
body = self.client.list_ports()
ports_list = body['ports']
self.assertFalse(port_id in [n['id'] for n in ports_list])
@test.attr(type='smoke')
def test_create_update_delete_port(self):
# Verify port creation
_, body = self.client.create_port(network_id=self.network['id'])
body = self.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.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'])
@@ -72,7 +72,7 @@ class PortsTestJSON(base.BaseNetworkTest):
network2 = self.create_network(network_name=name)
network_list = [network1['id'], network2['id']]
port_list = [{'network_id': net_id} for net_id in network_list]
_, body = self.client.create_bulk_port(port_list)
body = self.client.create_bulk_port(port_list)
created_ports = body['ports']
port1 = created_ports[0]
port2 = created_ports[1]
@@ -86,7 +86,7 @@ class PortsTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_show_port(self):
# Verify the details of port
_, body = self.client.show_port(self.port['id'])
body = self.client.show_port(self.port['id'])
port = body['port']
self.assertIn('id', port)
# TODO(Santosh)- This is a temporary workaround to compare create_port
@@ -100,8 +100,8 @@ class PortsTestJSON(base.BaseNetworkTest):
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.client.show_port(self.port['id'],
fields=fields)
port = body['port']
self.assertEqual(sorted(port.keys()), sorted(fields))
for field_name in fields:
@@ -110,7 +110,7 @@ class PortsTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
def test_list_ports(self):
# Verify the port exists in the list of all ports
_, body = self.client.list_ports()
body = self.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")
@@ -121,14 +121,14 @@ class PortsTestJSON(base.BaseNetworkTest):
network = self.create_network()
self.create_subnet(network)
router = self.create_router(data_utils.rand_name('router-'))
resp, port = self.client.create_port(network_id=network['id'])
port = self.client.create_port(network_id=network['id'])
# Add router interface to port created above
resp, interface = self.client.add_router_interface_with_port_id(
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.client.list_ports(device_id=router['id'])
ports = port_list['ports']
self.assertEqual(len(ports), 1)
self.assertEqual(ports[0]['id'], port['port']['id'])
@@ -138,7 +138,7 @@ class PortsTestJSON(base.BaseNetworkTest):
def test_list_ports_fields(self):
# Verify specific fields of ports
fields = ['id', 'mac_address']
_, body = self.client.list_ports(fields=fields)
body = self.client.list_ports(fields=fields)
ports = body['ports']
self.assertNotEmpty(ports, "Port list returned is empty")
# Asserting the fields returned are correct
@@ -173,20 +173,19 @@ class PortsTestJSON(base.BaseNetworkTest):
self.create_subnet(self.network)
security_groups_list = list()
for name in security_groups_names:
_, group_create_body = self.client.create_security_group(
group_create_body = self.client.create_security_group(
name=name)
self.addCleanup(self.client.delete_security_group,
group_create_body['security_group']['id'])
security_groups_list.append(group_create_body['security_group']
['id'])
# Create a port
_, body = self.client.create_port(**post_body)
body = self.client.create_port(**post_body)
self.addCleanup(self.client.delete_port, body['port']['id'])
port = body['port']
# Update the port with security groups
update_body = {"security_groups": security_groups_list}
_, body = self.client.update_port(
port['id'], **update_body)
body = self.client.update_port(port['id'], **update_body)
# Verify the security groups updated to port
port_show = body['port']
for security_group in security_groups_list:
@@ -206,16 +205,16 @@ class PortsTestJSON(base.BaseNetworkTest):
@test.attr(type='smoke')
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.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'])
# Create a new port with user defined mac
_, body = self.client.create_port(network_id=self.network['id'],
mac_address=free_mac_address)
body = self.client.create_port(network_id=self.network['id'],
mac_address=free_mac_address)
self.addCleanup(self.client.delete_port, body['port']['id'])
port = body['port']
_, body = self.client.show_port(port['id'])
body = self.client.show_port(port['id'])
show_port = body['port']
self.assertEqual(free_mac_address,
show_port['mac_address'])
@@ -253,7 +252,7 @@ class PortsAdminExtendedAttrsTestJSON(base.BaseAdminNetworkTest):
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_client.create_port(**post_body)
port = body['port']
self.addCleanup(self.admin_client.delete_port, port['id'])
host_id = port['binding:host_id']
@@ -263,11 +262,11 @@ class PortsAdminExtendedAttrsTestJSON(base.BaseAdminNetworkTest):
@test.attr(type='smoke')
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_client.create_port(**post_body)
port = body['port']
self.addCleanup(self.admin_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_client.update_port(port['id'], **update_body)
updated_port = body['port']
host_id = updated_port['binding:host_id']
self.assertIsNotNone(host_id)
@@ -277,7 +276,7 @@ class PortsAdminExtendedAttrsTestJSON(base.BaseAdminNetworkTest):
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_client.create_port(**post_body)
port = body['port']
self.addCleanup(self.admin_client.delete_port, port['id'])
@@ -288,7 +287,7 @@ class PortsAdminExtendedAttrsTestJSON(base.BaseAdminNetworkTest):
# 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_client.list_ports()
ports_list = body['ports']
pids_list = [p['id'] for p in ports_list]
self.assertIn(port['id'], pids_list)
@@ -300,10 +299,10 @@ class PortsAdminExtendedAttrsTestJSON(base.BaseAdminNetworkTest):
@test.attr(type='smoke')
def test_show_port_binding_ext_attr(self):
_, body = self.admin_client.create_port(network_id=self.network['id'])
body = self.admin_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'])
body = self.admin_client.show_port(port['id'])
show_port = body['port']
self.assertEqual(port['binding:host_id'],
show_port['binding:host_id'])

53
tempest/api/network/test_routers.py
View File

@@ -57,7 +57,7 @@ class RoutersTest(base.BaseRouterTest):
# NOTE(salv-orlando): Do not invoke self.create_router
# as we need to check the response code
name = data_utils.rand_name('router-')
_, create_body = self.client.create_router(
create_body = self.client.create_router(
name, external_gateway_info={
"network_id": CONF.network.public_network_id},
admin_state_up=False)
@@ -68,24 +68,24 @@ class RoutersTest(base.BaseRouterTest):
CONF.network.public_network_id)
self.assertEqual(create_body['router']['admin_state_up'], False)
# Show details of the created router
_, show_body = self.client.show_router(create_body['router']['id'])
show_body = self.client.show_router(create_body['router']['id'])
self.assertEqual(show_body['router']['name'], name)
self.assertEqual(
show_body['router']['external_gateway_info']['network_id'],
CONF.network.public_network_id)
self.assertEqual(show_body['router']['admin_state_up'], False)
# List routers and verify if created router is there in response
_, list_body = self.client.list_routers()
list_body = self.client.list_routers()
routers_list = list()
for router in list_body['routers']:
routers_list.append(router['id'])
self.assertIn(create_body['router']['id'], routers_list)
# Update the name of router and verify if it is updated
updated_name = 'updated ' + name
_, update_body = self.client.update_router(create_body['router']['id'],
name=updated_name)
update_body = self.client.update_router(create_body['router']['id'],
name=updated_name)
self.assertEqual(update_body['router']['name'], updated_name)
resp, show_body = self.client.show_router(
show_body = self.client.show_router(
create_body['router']['id'])
self.assertEqual(show_body['router']['name'], updated_name)
@@ -100,8 +100,8 @@ class RoutersTest(base.BaseRouterTest):
self.addCleanup(self.identity_admin_client.delete_tenant, tenant_id)
name = data_utils.rand_name('router-')
_, create_body = self.admin_client.create_router(name,
tenant_id=tenant_id)
create_body = self.admin_client.create_router(name,
tenant_id=tenant_id)
self.addCleanup(self.admin_client.delete_router,
create_body['router']['id'])
self.assertEqual(tenant_id, create_body['router']['tenant_id'])
@@ -127,7 +127,7 @@ class RoutersTest(base.BaseRouterTest):
external_gateway_info = {
'network_id': CONF.network.public_network_id,
'enable_snat': enable_snat}
_, create_body = self.admin_client.create_router(
create_body = self.admin_client.create_router(
name, external_gateway_info=external_gateway_info)
self.addCleanup(self.admin_client.delete_router,
create_body['router']['id'])
@@ -141,14 +141,14 @@ class RoutersTest(base.BaseRouterTest):
subnet = self.create_subnet(network)
router = self._create_router(data_utils.rand_name('router-'))
# Add router interface with subnet id
_, interface = self.client.add_router_interface_with_subnet_id(
interface = self.client.add_router_interface_with_subnet_id(
router['id'], subnet['id'])
self.addCleanup(self._remove_router_interface_with_subnet_id,
router['id'], subnet['id'])
self.assertIn('subnet_id', interface.keys())
self.assertIn('port_id', interface.keys())
# Verify router id is equal to device id in port details
resp, show_port_body = self.client.show_port(
show_port_body = self.client.show_port(
interface['port_id'])
self.assertEqual(show_port_body['port']['device_id'],
router['id'])
@@ -158,23 +158,23 @@ class RoutersTest(base.BaseRouterTest):
network = self.create_network()
self.create_subnet(network)
router = self._create_router(data_utils.rand_name('router-'))
resp, port_body = self.client.create_port(
port_body = self.client.create_port(
network_id=network['id'])
# add router interface to port created above
_, interface = self.client.add_router_interface_with_port_id(
interface = self.client.add_router_interface_with_port_id(
router['id'], port_body['port']['id'])
self.addCleanup(self._remove_router_interface_with_port_id,
router['id'], port_body['port']['id'])
self.assertIn('subnet_id', interface.keys())
self.assertIn('port_id', interface.keys())
# Verify router id is equal to device id in port details
resp, show_port_body = self.client.show_port(
show_port_body = self.client.show_port(
interface['port_id'])
self.assertEqual(show_port_body['port']['device_id'],
router['id'])
def _verify_router_gateway(self, router_id, exp_ext_gw_info=None):
_, show_body = self.admin_client.show_router(router_id)
show_body = self.admin_client.show_router(router_id)
actual_ext_gw_info = show_body['router']['external_gateway_info']
if exp_ext_gw_info is None:
self.assertIsNone(actual_ext_gw_info)
@@ -184,14 +184,14 @@ class RoutersTest(base.BaseRouterTest):
self.assertEqual(v, actual_ext_gw_info[k])
def _verify_gateway_port(self, router_id):
resp, list_body = self.admin_client.list_ports(
list_body = self.admin_client.list_ports(
network_id=CONF.network.public_network_id,
device_id=router_id)
self.assertEqual(len(list_body['ports']), 1)
gw_port = list_body['ports'][0]
fixed_ips = gw_port['fixed_ips']
self.assertGreaterEqual(len(fixed_ips), 1)
resp, public_net_body = self.admin_client.show_network(
public_net_body = self.admin_client.show_network(
CONF.network.public_network_id)
public_subnet_id = public_net_body['network']['subnets'][0]
self.assertIn(public_subnet_id,
@@ -205,7 +205,6 @@ class RoutersTest(base.BaseRouterTest):
external_gateway_info={
'network_id': CONF.network.public_network_id})
# Verify operation - router
_, show_body = self.client.show_router(router['id'])
self._verify_router_gateway(
router['id'],
{'network_id': CONF.network.public_network_id})
@@ -249,7 +248,7 @@ class RoutersTest(base.BaseRouterTest):
self.client.update_router(router['id'], external_gateway_info={})
self._verify_router_gateway(router['id'])
# No gateway port expected
resp, list_body = self.admin_client.list_ports(
list_body = self.admin_client.list_ports(
network_id=CONF.network.public_network_id,
device_id=router['id'])
self.assertFalse(list_body['ports'])
@@ -289,31 +288,31 @@ class RoutersTest(base.BaseRouterTest):
cidr = netaddr.IPNetwork(self.subnet['cidr'])
next_hop = str(cidr[2])
destination = str(self.subnet['cidr'])
_, extra_route = self.client.update_extra_routes(self.router['id'],
next_hop, destination)
extra_route = self.client.update_extra_routes(self.router['id'],
next_hop, destination)
self.assertEqual(1, len(extra_route['router']['routes']))
self.assertEqual(destination,
extra_route['router']['routes'][0]['destination'])
self.assertEqual(next_hop,
extra_route['router']['routes'][0]['nexthop'])
_, show_body = self.client.show_router(self.router['id'])
show_body = self.client.show_router(self.router['id'])
self.assertEqual(destination,
show_body['router']['routes'][0]['destination'])
self.assertEqual(next_hop,
show_body['router']['routes'][0]['nexthop'])
def _delete_extra_routes(self, router_id):
resp, _ = self.client.delete_extra_routes(router_id)
self.client.delete_extra_routes(router_id)
@test.attr(type='smoke')
def test_update_router_admin_state(self):
self.router = self._create_router(data_utils.rand_name('router-'))
self.assertFalse(self.router['admin_state_up'])
# Update router admin state
_, update_body = self.client.update_router(self.router['id'],
admin_state_up=True)
update_body = self.client.update_router(self.router['id'],
admin_state_up=True)
self.assertTrue(update_body['router']['admin_state_up'])
_, show_body = self.client.show_router(self.router['id'])
show_body = self.client.show_router(self.router['id'])
self.assertTrue(show_body['router']['admin_state_up'])
@test.attr(type='smoke')
@@ -336,7 +335,7 @@ class RoutersTest(base.BaseRouterTest):
interface02['port_id'])
def _verify_router_interface(self, router_id, subnet_id, port_id):
_, show_port_body = self.client.show_port(port_id)
show_port_body = self.client.show_port(port_id)
interface_port = show_port_body['port']
self.assertEqual(router_id, interface_port['device_id'])
self.assertEqual(subnet_id,

18
tempest/api/network/test_security_groups.py
View File

@@ -42,7 +42,7 @@ class SecGroupTest(base.BaseSecGroupTest):
remote_ip_prefix=None):
# Create Security Group rule with the input params and validate
# that SG rule is created with the same parameters.
resp, rule_create_body = self.client.create_security_group_rule(
rule_create_body = self.client.create_security_group_rule(
security_group_id=sg_id,
direction=direction,
ethertype=ethertype,
@@ -71,7 +71,7 @@ class SecGroupTest(base.BaseSecGroupTest):
@test.attr(type='smoke')
def test_list_security_groups(self):
# Verify the that security group belonging to tenant exist in list
_, body = self.client.list_security_groups()
body = self.client.list_security_groups()
security_groups = body['security_groups']
found = None
for n in security_groups:
@@ -85,7 +85,7 @@ class SecGroupTest(base.BaseSecGroupTest):
group_create_body, name = self._create_security_group()
# List security groups and verify if created group is there in response
_, list_body = self.client.list_security_groups()
list_body = self.client.list_security_groups()
secgroup_list = list()
for secgroup in list_body['security_groups']:
secgroup_list.append(secgroup['id'])
@@ -93,7 +93,7 @@ class SecGroupTest(base.BaseSecGroupTest):
# Update the security group
new_name = data_utils.rand_name('security-')
new_description = data_utils.rand_name('security-description')
_, update_body = self.client.update_security_group(
update_body = self.client.update_security_group(
group_create_body['security_group']['id'],
name=new_name,
description=new_description)
@@ -102,7 +102,7 @@ class SecGroupTest(base.BaseSecGroupTest):
self.assertEqual(update_body['security_group']['description'],
new_description)
# Show details of the updated security group
resp, show_body = self.client.show_security_group(
show_body = self.client.show_security_group(
group_create_body['security_group']['id'])
self.assertEqual(show_body['security_group']['name'], new_name)
self.assertEqual(show_body['security_group']['description'],
@@ -115,7 +115,7 @@ class SecGroupTest(base.BaseSecGroupTest):
# Create rules for each protocol
protocols = ['tcp', 'udp', 'icmp']
for protocol in protocols:
_, rule_create_body = self.client.create_security_group_rule(
rule_create_body = self.client.create_security_group_rule(
security_group_id=group_create_body['security_group']['id'],
protocol=protocol,
direction='ingress',
@@ -123,7 +123,7 @@ class SecGroupTest(base.BaseSecGroupTest):
)
# Show details of the created security rule
_, show_rule_body = self.client.show_security_group_rule(
show_rule_body = self.client.show_security_group_rule(
rule_create_body['security_group_rule']['id']
)
create_dict = rule_create_body['security_group_rule']
@@ -133,7 +133,7 @@ class SecGroupTest(base.BaseSecGroupTest):
"%s does not match." % key)
# List rules and verify created rule is in response
_, rule_list_body = self.client.list_security_group_rules()
rule_list_body = self.client.list_security_group_rules()
rule_list = [rule['id']
for rule in rule_list_body['security_group_rules']]
self.assertIn(rule_create_body['security_group_rule']['id'],
@@ -221,7 +221,7 @@ class SecGroupTest(base.BaseSecGroupTest):
direction = 'ingress'
protocol = 17
security_group_id = group_create_body['security_group']['id']
_, rule_create_body = self.client.create_security_group_rule(
rule_create_body = self.client.create_security_group_rule(
security_group_id=security_group_id,
direction=direction,
protocol=protocol

2
tempest/api/network/test_security_groups_negative.py
View File

@@ -164,7 +164,7 @@ class NegativeSecGroupTest(base.BaseSecGroupTest):
min_port = 66
max_port = 67
# Create a rule with valid params
resp, _ = self.client.create_security_group_rule(
self.client.create_security_group_rule(
security_group_id=body['security_group']['id'],
direction='ingress',
ethertype=self.ethertype,

2
tempest/api/network/test_service_type_management.py
View File

@@ -27,5 +27,5 @@ class ServiceTypeManagementTestJSON(base.BaseNetworkTest):
@test.skip_because(bug="1400370")
@test.attr(type='smoke')
def test_service_provider_list(self):
_, body = self.client.list_service_providers()
body = self.client.list_service_providers()
self.assertIsInstance(body['service_providers'], list)

72
tempest/api/network/test_vpnaas_extensions.py
View File

@@ -57,13 +57,13 @@ class VPNaaSTestJSON(base.BaseAdminNetworkTest):
def _delete_ike_policy(self, ike_policy_id):
# Deletes a ike policy and verifies if it is deleted or not
ike_list = list()
resp, all_ike = self.client.list_ikepolicies()
all_ike = self.client.list_ikepolicies()
for ike in all_ike['ikepolicies']:
ike_list.append(ike['id'])
if ike_policy_id in ike_list:
self.client.delete_ikepolicy(ike_policy_id)
# Asserting that the policy is not found in list after deletion
resp, ikepolicies = self.client.list_ikepolicies()
ikepolicies = self.client.list_ikepolicies()
ike_id_list = list()
for i in ikepolicies['ikepolicies']:
ike_id_list.append(i['id'])
@@ -86,7 +86,7 @@ class VPNaaSTestJSON(base.BaseAdminNetworkTest):
def _delete_vpn_service(self, vpn_service_id):
self.client.delete_vpnservice(vpn_service_id)
# Asserting if vpn service is found in the list after deletion
_, body = self.client.list_vpnservices()
body = self.client.list_vpnservices()
vpn_services = [vs['id'] for vs in body['vpnservices']]
self.assertNotIn(vpn_service_id, vpn_services)
@@ -97,7 +97,7 @@ class VPNaaSTestJSON(base.BaseAdminNetworkTest):
# TODO(jroovers) This is a temporary workaround to get the tenant_id
# of the the current client. Replace this once tenant_isolation for
# neutron is fixed.
_, body = self.client.show_network(self.network['id'])
body = self.client.show_network(self.network['id'])
return body['network']['tenant_id']
@test.attr(type='smoke')
@@ -105,15 +105,15 @@ class VPNaaSTestJSON(base.BaseAdminNetworkTest):
tenant_id = self._get_tenant_id()
# Create IPSec policy for the newly created tenant
name = data_utils.rand_name('ipsec-policy')
_, body = (self.admin_client.
create_ipsecpolicy(name=name, tenant_id=tenant_id))
body = (self.admin_client.
create_ipsecpolicy(name=name, tenant_id=tenant_id))
ipsecpolicy = body['ipsecpolicy']
self.assertIsNotNone(ipsecpolicy['id'])
self.addCleanup(self.admin_client.delete_ipsecpolicy,
ipsecpolicy['id'])
# Assert that created ipsec policy is found in API list call
_, body = self.client.list_ipsecpolicies()
body = self.client.list_ipsecpolicies()
ipsecpolicies = [policy['id'] for policy in body['ipsecpolicies']]
self.assertIn(ipsecpolicy['id'], ipsecpolicies)
@@ -128,7 +128,7 @@ class VPNaaSTestJSON(base.BaseAdminNetworkTest):
external_network_id=self.ext_net_id)
self.create_router_interface(router2['id'], subnet2['id'])
name = data_utils.rand_name('vpn-service')
_, body = self.admin_client.create_vpnservice(
body = self.admin_client.create_vpnservice(
subnet_id=subnet2['id'],
router_id=router2['id'],
name=name,
@@ -138,7 +138,7 @@ class VPNaaSTestJSON(base.BaseAdminNetworkTest):
self.assertIsNotNone(vpnservice['id'])
self.addCleanup(self.admin_client.delete_vpnservice, vpnservice['id'])
# Assert that created vpnservice is found in API list call
_, body = self.client.list_vpnservices()
body = self.client.list_vpnservices()
vpn_services = [vs['id'] for vs in body['vpnservices']]
self.assertIn(vpnservice['id'], vpn_services)
@@ -148,24 +148,24 @@ class VPNaaSTestJSON(base.BaseAdminNetworkTest):
# Create IKE policy for the newly created tenant
name = data_utils.rand_name('ike-policy')
_, body = (self.admin_client.
create_ikepolicy(name=name, ike_version="v1",
encryption_algorithm="aes-128",
auth_algorithm="sha1",
tenant_id=tenant_id))
body = (self.admin_client.
create_ikepolicy(name=name, ike_version="v1",
encryption_algorithm="aes-128",
auth_algorithm="sha1",
tenant_id=tenant_id))
ikepolicy = body['ikepolicy']
self.assertIsNotNone(ikepolicy['id'])
self.addCleanup(self.admin_client.delete_ikepolicy, ikepolicy['id'])
# Assert that created ike policy is found in API list call
_, body = self.client.list_ikepolicies()
body = self.client.list_ikepolicies()
ikepolicies = [ikp['id'] for ikp in body['ikepolicies']]
self.assertIn(ikepolicy['id'], ikepolicies)
@test.attr(type='smoke')
def test_list_vpn_services(self):
# Verify the VPN service exists in the list of all VPN services
_, body = self.client.list_vpnservices()
body = self.client.list_vpnservices()
vpnservices = body['vpnservices']
self.assertIn(self.vpnservice['id'], [v['id'] for v in vpnservices])
@@ -178,14 +178,14 @@ class VPNaaSTestJSON(base.BaseAdminNetworkTest):
external_network_id=self.ext_net_id)
self.create_router_interface(router1['id'], subnet1['id'])
name = data_utils.rand_name('vpn-service1')
_, body = self.client.create_vpnservice(subnet_id=subnet1['id'],
router_id=router1['id'],
name=name,
admin_state_up=True)
body = self.client.create_vpnservice(subnet_id=subnet1['id'],
router_id=router1['id'],
name=name,
admin_state_up=True)
vpnservice = body['vpnservice']
self.addCleanup(self._delete_vpn_service, vpnservice['id'])
# Assert if created vpnservices are not found in vpnservices list
resp, body = self.client.list_vpnservices()
body = self.client.list_vpnservices()
vpn_services = [vs['id'] for vs in body['vpnservices']]
self.assertIsNotNone(vpnservice['id'])
self.assertIn(vpnservice['id'], vpn_services)
@@ -198,7 +198,7 @@ class VPNaaSTestJSON(base.BaseAdminNetworkTest):
@test.attr(type='smoke')
def test_show_vpn_service(self):
# Verifies the details of a vpn service
_, body = self.client.show_vpnservice(self.vpnservice['id'])
body = self.client.show_vpnservice(self.vpnservice['id'])
vpnservice = body['vpnservice']
self.assertEqual(self.vpnservice['id'], vpnservice['id'])
self.assertEqual(self.vpnservice['name'], vpnservice['name'])
@@ -214,7 +214,7 @@ class VPNaaSTestJSON(base.BaseAdminNetworkTest):
@test.attr(type='smoke')
def test_list_ike_policies(self):
# Verify the ike policy exists in the list of all IKE policies
_, body = self.client.list_ikepolicies()
body = self.client.list_ikepolicies()
ikepolicies = body['ikepolicies']
self.assertIn(self.ikepolicy['id'], [i['id'] for i in ikepolicies])
@@ -222,11 +222,11 @@ class VPNaaSTestJSON(base.BaseAdminNetworkTest):
def test_create_update_delete_ike_policy(self):
# Creates a IKE policy
name = data_utils.rand_name('ike-policy')
_, body = (self.client.create_ikepolicy(
name=name,
ike_version="v1",
encryption_algorithm="aes-128",
auth_algorithm="sha1"))
body = (self.client.create_ikepolicy(
name=name,
ike_version="v1",
encryption_algorithm="aes-128",
auth_algorithm="sha1"))
ikepolicy = body['ikepolicy']
self.assertIsNotNone(ikepolicy['id'])
self.addCleanup(self._delete_ike_policy, ikepolicy['id'])
@@ -240,7 +240,7 @@ class VPNaaSTestJSON(base.BaseAdminNetworkTest):
'lifetime': {'units': "seconds", 'value': 2000}}
self.client.update_ikepolicy(ikepolicy['id'], **new_ike)
# Confirm that update was successful by verifying using 'show'
_, body = self.client.show_ikepolicy(ikepolicy['id'])
body = self.client.show_ikepolicy(ikepolicy['id'])
ike_policy = body['ikepolicy']
for key, value in new_ike.iteritems():
self.assertIn(key, ike_policy)
@@ -248,14 +248,14 @@ class VPNaaSTestJSON(base.BaseAdminNetworkTest):
# Verification of ike policy delete
self.client.delete_ikepolicy(ikepolicy['id'])
_, body = self.client.list_ikepolicies()
body = self.client.list_ikepolicies()
ikepolicies = [ikp['id'] for ikp in body['ikepolicies']]
self.assertNotIn(ike_policy['id'], ikepolicies)
@test.attr(type='smoke')
def test_show_ike_policy(self):
# Verifies the details of a ike policy
_, body = self.client.show_ikepolicy(self.ikepolicy['id'])
body = self.client.show_ikepolicy(self.ikepolicy['id'])
ikepolicy = body['ikepolicy']
self.assertEqual(self.ikepolicy['id'], ikepolicy['id'])
self.assertEqual(self.ikepolicy['name'], ikepolicy['name'])
@@ -277,7 +277,7 @@ class VPNaaSTestJSON(base.BaseAdminNetworkTest):
@test.attr(type='smoke')
def test_list_ipsec_policies(self):
# Verify the ipsec policy exists in the list of all ipsec policies
_, body = self.client.list_ipsecpolicies()
body = self.client.list_ipsecpolicies()
ipsecpolicies = body['ipsecpolicies']
self.assertIn(self.ipsecpolicy['id'], [i['id'] for i in ipsecpolicies])
@@ -288,7 +288,7 @@ class VPNaaSTestJSON(base.BaseAdminNetworkTest):
'pfs': 'group5',
'encryption_algorithm': "aes-128",
'auth_algorithm': 'sha1'}
_, resp_body = self.client.create_ipsecpolicy(**ipsec_policy_body)
resp_body = self.client.create_ipsecpolicy(**ipsec_policy_body)
ipsecpolicy = resp_body['ipsecpolicy']
self.addCleanup(self._delete_ipsec_policy, ipsecpolicy['id'])
self._assertExpected(ipsec_policy_body, ipsecpolicy)
@@ -298,8 +298,8 @@ class VPNaaSTestJSON(base.BaseAdminNetworkTest):
'name': data_utils.rand_name("New-IPSec"),
'encryption_algorithm': "aes-256",
'lifetime': {'units': "seconds", 'value': '2000'}}
_, body = self.client.update_ipsecpolicy(ipsecpolicy['id'],
**new_ipsec)
body = self.client.update_ipsecpolicy(ipsecpolicy['id'],
**new_ipsec)
updated_ipsec_policy = body['ipsecpolicy']
self._assertExpected(new_ipsec, updated_ipsec_policy)
# Verification of ipsec policy delete
@@ -310,6 +310,6 @@ class VPNaaSTestJSON(base.BaseAdminNetworkTest):
@test.attr(type='smoke')
def test_show_ipsec_policy(self):
# Verifies the details of an ipsec policy
_, body = self.client.show_ipsecpolicy(self.ipsecpolicy['id'])
body = self.client.show_ipsecpolicy(self.ipsecpolicy['id'])
ipsecpolicy = body['ipsecpolicy']
self._assertExpected(self.ipsecpolicy, ipsecpolicy)

2
tempest/api/orchestration/base.py
View File

@@ -51,7 +51,7 @@ class BaseOrchestrationTest(tempest.test.BaseTestCase):
@classmethod
def _get_default_network(cls):
_, networks = cls.network_client.list_networks()
networks = cls.network_client.list_networks()
for net in networks['networks']:
if net['name'] == CONF.compute.fixed_network_name:
return net

8
tempest/api/orchestration/stacks/test_neutron_resources.py
View File

@@ -107,7 +107,7 @@ class NeutronResourcesTestJSON(base.BaseOrchestrationTest):
def test_created_network(self):
"""Verifies created network."""
network_id = self.test_resources.get('Network')['physical_resource_id']
_, body = self.network_client.show_network(network_id)
body = self.network_client.show_network(network_id)
network = body['network']
self.assertIsInstance(network, dict)
self.assertEqual(network_id, network['id'])
@@ -119,7 +119,7 @@ class NeutronResourcesTestJSON(base.BaseOrchestrationTest):
def test_created_subnet(self):
"""Verifies created subnet."""
subnet_id = self.test_resources.get('Subnet')['physical_resource_id']
_, body = self.network_client.show_subnet(subnet_id)
body = self.network_client.show_subnet(subnet_id)
subnet = body['subnet']
network_id = self.test_resources.get('Network')['physical_resource_id']
self.assertEqual(subnet_id, subnet['id'])
@@ -137,7 +137,7 @@ class NeutronResourcesTestJSON(base.BaseOrchestrationTest):
def test_created_router(self):
"""Verifies created router."""
router_id = self.test_resources.get('Router')['physical_resource_id']
_, body = self.network_client.show_router(router_id)
body = self.network_client.show_router(router_id)
router = body['router']
self.assertEqual(self.neutron_basic_template['resources'][
'Router']['properties']['name'], router['name'])
@@ -152,7 +152,7 @@ class NeutronResourcesTestJSON(base.BaseOrchestrationTest):
router_id = self.test_resources.get('Router')['physical_resource_id']
network_id = self.test_resources.get('Network')['physical_resource_id']
subnet_id = self.test_resources.get('Subnet')['physical_resource_id']
_, body = self.network_client.list_ports()
body = self.network_client.list_ports()
ports = body['ports']
router_ports = filter(lambda port: port['device_id'] ==
router_id, ports)

36
tempest/cmd/cleanup_service.py
View File

@@ -85,7 +85,7 @@ def _get_priv_net_id(prv_net_name, tenant_name):
net_cl = am.network_client
id_cl = am.identity_client
_, networks = net_cl.list_networks()
networks = net_cl.list_networks()
tenant = id_cl.get_tenant_by_name(tenant_name)
t_id = tenant['id']
n_id = None
@@ -357,7 +357,7 @@ class NetworkService(BaseService):
def list(self):
client = self.client
_, networks = client.list_networks()
networks = client.list_networks()
networks = self._filter_by_tenant_id(networks['networks'])
# filter out networks declared in tempest.conf
if self.is_preserve:
@@ -385,7 +385,7 @@ class NetworkIpSecPolicyService(NetworkService):
def list(self):
client = self.client
_, ipsecpols = client.list_ipsecpolicies()
ipsecpols = client.list_ipsecpolicies()
ipsecpols = ipsecpols['ipsecpolicies']
ipsecpols = self._filter_by_tenant_id(ipsecpols)
LOG.debug("List count, %s IP Security Policies" % len(ipsecpols))
@@ -410,7 +410,7 @@ class NetworkFwPolicyService(NetworkService):
def list(self):
client = self.client
_, fwpols = client.list_firewall_policies()
fwpols = client.list_firewall_policies()
fwpols = fwpols['firewall_policies']
fwpols = self._filter_by_tenant_id(fwpols)
LOG.debug("List count, %s Firewall Policies" % len(fwpols))
@@ -435,7 +435,7 @@ class NetworkFwRulesService(NetworkService):
def list(self):
client = self.client
_, fwrules = client.list_firewall_rules()
fwrules = client.list_firewall_rules()
fwrules = fwrules['firewall_rules']
fwrules = self._filter_by_tenant_id(fwrules)
LOG.debug("List count, %s Firewall Rules" % len(fwrules))
@@ -460,7 +460,7 @@ class NetworkIkePolicyService(NetworkService):
def list(self):
client = self.client
_, ikepols = client.list_ikepolicies()
ikepols = client.list_ikepolicies()
ikepols = ikepols['ikepolicies']
ikepols = self._filter_by_tenant_id(ikepols)
LOG.debug("List count, %s IKE Policies" % len(ikepols))
@@ -485,7 +485,7 @@ class NetworkVpnServiceService(NetworkService):
def list(self):
client = self.client
_, vpnsrvs = client.list_vpnservices()
vpnsrvs = client.list_vpnservices()
vpnsrvs = vpnsrvs['vpnservices']
vpnsrvs = self._filter_by_tenant_id(vpnsrvs)
LOG.debug("List count, %s VPN Services" % len(vpnsrvs))
@@ -510,7 +510,7 @@ class NetworkFloatingIpService(NetworkService):
def list(self):
client = self.client
_, flips = client.list_floatingips()
flips = client.list_floatingips()
flips = flips['floatingips']
flips = self._filter_by_tenant_id(flips)
LOG.debug("List count, %s Network Floating IPs" % len(flips))
@@ -535,7 +535,7 @@ class NetworkRouterService(NetworkService):
def list(self):
client = self.client
_, routers = client.list_routers()
routers = client.list_routers()
routers = routers['routers']
routers = self._filter_by_tenant_id(routers)
if self.is_preserve:
@@ -551,7 +551,7 @@ class NetworkRouterService(NetworkService):
for router in routers:
try:
rid = router['id']
_, ports = client.list_router_interfaces(rid)
ports = client.list_router_interfaces(rid)
ports = ports['ports']
for port in ports:
subid = port['fixed_ips'][0]['subnet_id']
@@ -570,7 +570,7 @@ class NetworkHealthMonitorService(NetworkService):
def list(self):
client = self.client
_, hms = client.list_health_monitors()
hms = client.list_health_monitors()
hms = hms['health_monitors']
hms = self._filter_by_tenant_id(hms)
LOG.debug("List count, %s Health Monitors" % len(hms))
@@ -595,7 +595,7 @@ class NetworkMemberService(NetworkService):
def list(self):
client = self.client
_, members = client.list_members()
members = client.list_members()
members = members['members']
members = self._filter_by_tenant_id(members)
LOG.debug("List count, %s Members" % len(members))
@@ -620,7 +620,7 @@ class NetworkVipService(NetworkService):
def list(self):
client = self.client
_, vips = client.list_vips()
vips = client.list_vips()
vips = vips['vips']
vips = self._filter_by_tenant_id(vips)
LOG.debug("List count, %s VIPs" % len(vips))
@@ -645,7 +645,7 @@ class NetworkPoolService(NetworkService):
def list(self):
client = self.client
_, pools = client.list_pools()
pools = client.list_pools()
pools = pools['pools']
pools = self._filter_by_tenant_id(pools)
LOG.debug("List count, %s Pools" % len(pools))
@@ -670,7 +670,7 @@ class NetworMeteringLabelRuleService(NetworkService):
def list(self):
client = self.client
_, rules = client.list_metering_label_rules()
rules = client.list_metering_label_rules()
rules = rules['metering_label_rules']
rules = self._filter_by_tenant_id(rules)
LOG.debug("List count, %s Metering Label Rules" % len(rules))
@@ -695,7 +695,7 @@ class NetworMeteringLabelService(NetworkService):
def list(self):
client = self.client
_, labels = client.list_metering_labels()
labels = client.list_metering_labels()
labels = labels['metering_labels']
labels = self._filter_by_tenant_id(labels)
LOG.debug("List count, %s Metering Labels" % len(labels))
@@ -720,7 +720,7 @@ class NetworkPortService(NetworkService):
def list(self):
client = self.client
_, ports = client.list_ports()
ports = client.list_ports()
ports = ports['ports']
ports = self._filter_by_tenant_id(ports)
if self.is_preserve:
@@ -747,7 +747,7 @@ class NetworkSubnetService(NetworkService):
def list(self):
client = self.client
_, subnets = client.list_subnets()
subnets = client.list_subnets()
subnets = subnets['subnets']
subnets = self._filter_by_tenant_id(subnets)
if self.is_preserve:

22
tempest/cmd/javelin.py
View File

@@ -509,15 +509,10 @@ def destroy_images(images):
def _get_router_namespace(client, network):
network_id = _get_resource_by_name(client.networks,
'networks', network)['id']
resp, n_body = client.networks.list_routers()
if not resp_ok(resp):
raise ValueError("unable to routers list: [%s] %s" % (resp, n_body))
n_body = client.networks.list_routers()
for router in n_body['routers']:
router_id = router['id']
resp, r_body = client.networks.list_router_interfaces(router_id)
if not resp_ok(resp):
raise ValueError("unable to router interfaces list: [%s] %s" %
(resp, r_body))
r_body = client.networks.list_router_interfaces(router_id)
for port in r_body['ports']:
if port['network_id'] == network_id:
return "qrouter-%s" % router_id
@@ -527,9 +522,12 @@ def _get_resource_by_name(client, resource, name):
get_resources = getattr(client, 'list_%s' % resource)
if get_resources is None:
raise AttributeError("client doesn't have method list_%s" % resource)
r, body = get_resources()
if not resp_ok(r):
raise ValueError("unable to list %s: [%s] %s" % (resource, r, body))
# Until all tempest client methods are changed to return only one value,
# we cannot assume they all have the same signature so we need to discard
# the unused response first value it two values are being returned.
body = get_resources()
if type(body) == tuple:
body = body[1]
if isinstance(body, dict):
body = body[resource]
for res in body:
@@ -544,7 +542,7 @@ def create_networks(networks):
client = client_for_user(network['owner'])
# only create a network if the name isn't here
r, body = client.networks.list_networks()
body = client.networks.list_networks()
if any(item['name'] == network['name'] for item in body['networks']):
LOG.warning("Dupplicated network name: %s" % network['name'])
continue
@@ -596,7 +594,7 @@ def create_routers(routers):
client = client_for_user(router['owner'])
# only create a router if the name isn't here
r, body = client.networks.list_routers()
body = client.networks.list_routers()
if any(item['name'] == router['name'] for item in body['routers']):
LOG.warning("Dupplicated router name: %s" % router['name'])
continue

5
tempest/cmd/verify_tempest_config.py
View File

@@ -162,7 +162,10 @@ def get_enabled_extensions(service):
def verify_extensions(os, service, results):
extensions_client = get_extension_client(os, service)
__, resp = extensions_client.list_extensions()
if service == 'neutron':
resp = extensions_client.list_extensions()
else:
__, resp = extensions_client.list_extensions()
# For Nova, Cinder and Neutron we use the alias name rather than the
# 'name' field because the alias is considered to be the canonical
# name.

12
tempest/common/isolated_creds.py
View File

@@ -174,7 +174,7 @@ class IsolatedCreds(cred_provider.CredentialProvider):
return network, subnet, router
def _create_network(self, name, tenant_id):
_, resp_body = self.network_admin_client.create_network(
resp_body = self.network_admin_client.create_network(
name=name, tenant_id=tenant_id)
return resp_body['network']
@@ -184,7 +184,7 @@ class IsolatedCreds(cred_provider.CredentialProvider):
for subnet_cidr in base_cidr.subnet(mask_bits):
try:
if self.network_resources:
_, resp_body = self.network_admin_client.\
resp_body = self.network_admin_client.\
create_subnet(
network_id=network_id, cidr=str(subnet_cidr),
name=subnet_name,
@@ -192,7 +192,7 @@ class IsolatedCreds(cred_provider.CredentialProvider):
enable_dhcp=self.network_resources['dhcp'],
ip_version=4)
else:
_, resp_body = self.network_admin_client.\
resp_body = self.network_admin_client.\
create_subnet(network_id=network_id,
cidr=str(subnet_cidr),
name=subnet_name,
@@ -210,7 +210,7 @@ class IsolatedCreds(cred_provider.CredentialProvider):
def _create_router(self, router_name, tenant_id):
external_net_id = dict(
network_id=CONF.network.public_network_id)
_, resp_body = self.network_admin_client.create_router(
resp_body = self.network_admin_client.create_router(
router_name,
external_gateway_info=external_net_id,
tenant_id=tenant_id)
@@ -302,8 +302,8 @@ class IsolatedCreds(cred_provider.CredentialProvider):
def _cleanup_default_secgroup(self, tenant):
net_client = self.network_admin_client
_, resp_body = net_client.list_security_groups(tenant_id=tenant,
name="default")
resp_body = net_client.list_security_groups(tenant_id=tenant,
name="default")
secgroups_to_delete = resp_body['security_groups']
for secgroup in secgroups_to_delete:
try:

40
tempest/scenario/manager.py
View File

@@ -558,7 +558,7 @@ class NetworkScenarioTest(ScenarioTest):
if not tenant_id:
tenant_id = client.rest_client.tenant_id
name = data_utils.rand_name(namestart)
_, result = client.create_network(name=name, tenant_id=tenant_id)
result = client.create_network(name=name, tenant_id=tenant_id)
network = net_resources.DeletableNetwork(client=client,
**result['network'])
self.assertEqual(network.name, name)
@@ -585,7 +585,7 @@ class NetworkScenarioTest(ScenarioTest):
def temp(*args, **kwargs):
temp_method = self.admin_manager.network_client.__getattr__(
'list_%s' % resource_type)
_, resource_list = temp_method(*args, **kwargs)
resource_list = temp_method(*args, **kwargs)
return resource_list[resource_type]
return temp
@@ -634,7 +634,7 @@ class NetworkScenarioTest(ScenarioTest):
**kwargs
)
try:
_, result = client.create_subnet(**subnet)
result = client.create_subnet(**subnet)
break
except exceptions.Conflict as e:
is_overlapping_cidr = 'overlaps with another subnet' in str(e)
@@ -651,7 +651,7 @@ class NetworkScenarioTest(ScenarioTest):
if not client:
client = self.network_client
name = data_utils.rand_name(namestart)
_, result = client.create_port(
result = client.create_port(
name=name,
network_id=network.id,
tenant_id=network.tenant_id)
@@ -690,7 +690,7 @@ class NetworkScenarioTest(ScenarioTest):
port_id, ip4 = self._get_server_port_id_and_ip4(thing)
else:
ip4 = None
_, result = client.create_floatingip(
result = client.create_floatingip(
floating_network_id=external_network_id,
port_id=port_id,
tenant_id=thing['tenant_id'],
@@ -823,7 +823,7 @@ class NetworkScenarioTest(ScenarioTest):
sg_dict = dict(name=sg_name,
description=sg_desc)
sg_dict['tenant_id'] = tenant_id
_, result = client.create_security_group(**sg_dict)
result = client.create_security_group(**sg_dict)
secgroup = net_resources.DeletableSecurityGroup(
client=client,
**result['security_group']
@@ -883,7 +883,7 @@ class NetworkScenarioTest(ScenarioTest):
tenant_id=secgroup.tenant_id)
ruleset.update(kwargs)
_, sg_rule = client.create_security_group_rule(**ruleset)
sg_rule = client.create_security_group_rule(**ruleset)
sg_rule = net_resources.DeletableSecurityGroupRule(
client=client,
**sg_rule['security_group_rule']
@@ -937,9 +937,9 @@ class NetworkScenarioTest(ScenarioTest):
"""Wrapper utility that returns a test pool."""
client = self.network_client
name = data_utils.rand_name('pool')
_, resp_pool = client.create_pool(protocol=protocol, name=name,
subnet_id=subnet_id,
lb_method=lb_method)
resp_pool = client.create_pool(protocol=protocol, name=name,
subnet_id=subnet_id,
lb_method=lb_method)
pool = net_resources.DeletablePool(client=client, **resp_pool['pool'])
self.assertEqual(pool['name'], name)
self.addCleanup(self.delete_wrapper, pool.delete)
@@ -948,9 +948,9 @@ class NetworkScenarioTest(ScenarioTest):
def _create_member(self, address, protocol_port, pool_id):
"""Wrapper utility that returns a test member."""
client = self.network_client
_, resp_member = client.create_member(protocol_port=protocol_port,
pool_id=pool_id,
address=address)
resp_member = client.create_member(protocol_port=protocol_port,
pool_id=pool_id,
address=address)
member = net_resources.DeletableMember(client=client,
**resp_member['member'])
self.addCleanup(self.delete_wrapper, member.delete)
@@ -960,9 +960,9 @@ class NetworkScenarioTest(ScenarioTest):
"""Wrapper utility that returns a test vip."""
client = self.network_client
name = data_utils.rand_name('vip')
_, resp_vip = client.create_vip(protocol=protocol, name=name,
subnet_id=subnet_id, pool_id=pool_id,
protocol_port=protocol_port)
resp_vip = client.create_vip(protocol=protocol, name=name,
subnet_id=subnet_id, pool_id=pool_id,
protocol_port=protocol_port)
vip = net_resources.DeletableVip(client=client, **resp_vip['vip'])
self.assertEqual(vip['name'], name)
self.addCleanup(self.delete_wrapper, vip.delete)
@@ -1007,9 +1007,9 @@ class NetworkScenarioTest(ScenarioTest):
if not tenant_id:
tenant_id = client.rest_client.tenant_id
name = data_utils.rand_name(namestart)
_, result = client.create_router(name=name,
admin_state_up=True,
tenant_id=tenant_id)
result = client.create_router(name=name,
admin_state_up=True,
tenant_id=tenant_id)
router = net_resources.DeletableRouter(client=client,
**result['router'])
self.assertEqual(router.name, name)
@@ -1274,7 +1274,7 @@ class OrchestrationScenarioTest(ScenarioTest):
@classmethod
def _get_default_network(cls):
_, networks = cls.networks_client.list_networks()
networks = cls.networks_client.list_networks()
for net in networks:
if net['label'] == CONF.compute.fixed_network_name:
return net

2
tempest/scenario/test_baremetal_basic_ops.py
View File

@@ -112,7 +112,7 @@ class BaremetalBasicOps(manager.BaremetalScenarioTest):
def validate_ports(self):
for port in self.get_ports(self.node['uuid']):
n_port_id = port['extra']['vif_port_id']
_, body = self.network_client.show_port(n_port_id)
body = self.network_client.show_port(n_port_id)
n_port = body['port']
self.assertEqual(n_port['device_id'], self.instance['id'])
self.assertEqual(n_port['mac_address'], port['address'])

56
tempest/services/network/json/network_client.py
View File

@@ -61,13 +61,13 @@ class NetworkClientJSON(network_client_base.NetworkClientBase):
resp, body = self.put(uri, body)
self.rest_client.expected_success(200, resp.status)
body = json.loads(body)
return resp, body['quota']
return rest_client.ResponseBody(resp, body['quota'])
def reset_quotas(self, tenant_id):
uri = '%s/quotas/%s' % (self.uri_prefix, tenant_id)
resp, body = self.delete(uri)
self.rest_client.expected_success(204, resp.status)
return resp, body
return rest_client.ResponseBody(resp, body)
def create_router(self, name, admin_state_up=True, **kwargs):
post_body = {'router': kwargs}
@@ -78,7 +78,7 @@ class NetworkClientJSON(network_client_base.NetworkClientBase):
resp, body = self.post(uri, body)
self.rest_client.expected_success(201, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def _update_router(self, router_id, set_enable_snat, **kwargs):
uri = '%s/routers/%s' % (self.uri_prefix, router_id)
@@ -105,7 +105,7 @@ class NetworkClientJSON(network_client_base.NetworkClientBase):
resp, body = self.put(uri, update_body)
self.rest_client.expected_success(200, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def update_router(self, router_id, **kwargs):
"""Update a router leaving enable_snat to its default value."""
@@ -132,7 +132,7 @@ class NetworkClientJSON(network_client_base.NetworkClientBase):
resp, body = self.put(uri, update_body)
self.rest_client.expected_success(200, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def add_router_interface_with_port_id(self, router_id, port_id):
uri = '%s/routers/%s/add_router_interface' % (self.uri_prefix,
@@ -142,7 +142,7 @@ class NetworkClientJSON(network_client_base.NetworkClientBase):
resp, body = self.put(uri, update_body)
self.rest_client.expected_success(200, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def remove_router_interface_with_subnet_id(self, router_id, subnet_id):
uri = '%s/routers/%s/remove_router_interface' % (self.uri_prefix,
@@ -152,7 +152,7 @@ class NetworkClientJSON(network_client_base.NetworkClientBase):
resp, body = self.put(uri, update_body)
self.rest_client.expected_success(200, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def remove_router_interface_with_port_id(self, router_id, port_id):
uri = '%s/routers/%s/remove_router_interface' % (self.uri_prefix,
@@ -162,7 +162,7 @@ class NetworkClientJSON(network_client_base.NetworkClientBase):
resp, body = self.put(uri, update_body)
self.rest_client.expected_success(200, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def associate_health_monitor_with_pool(self, health_monitor_id,
pool_id):
@@ -177,7 +177,7 @@ class NetworkClientJSON(network_client_base.NetworkClientBase):
resp, body = self.post(uri, body)
self.rest_client.expected_success(201, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def disassociate_health_monitor_with_pool(self, health_monitor_id,
pool_id):
@@ -185,14 +185,14 @@ class NetworkClientJSON(network_client_base.NetworkClientBase):
health_monitor_id)
resp, body = self.delete(uri)
self.rest_client.expected_success(204, resp.status)
return resp, body
return rest_client.ResponseBody(resp, body)
def list_router_interfaces(self, uuid):
uri = '%s/ports?device_id=%s' % (self.uri_prefix, uuid)
resp, body = self.get(uri)
self.rest_client.expected_success(200, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def update_agent(self, agent_id, agent_info):
"""
@@ -205,14 +205,14 @@ class NetworkClientJSON(network_client_base.NetworkClientBase):
resp, body = self.put(uri, body)
self.rest_client.expected_success(200, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def list_pools_hosted_by_one_lbaas_agent(self, agent_id):
uri = '%s/agents/%s/loadbalancer-pools' % (self.uri_prefix, agent_id)
resp, body = self.get(uri)
self.rest_client.expected_success(200, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def show_lbaas_agent_hosting_pool(self, pool_id):
uri = ('%s/lb/pools/%s/loadbalancer-agent' %
@@ -220,21 +220,21 @@ class NetworkClientJSON(network_client_base.NetworkClientBase):
resp, body = self.get(uri)
self.rest_client.expected_success(200, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def list_routers_on_l3_agent(self, agent_id):
uri = '%s/agents/%s/l3-routers' % (self.uri_prefix, agent_id)
resp, body = self.get(uri)
self.rest_client.expected_success(200, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def list_l3_agents_hosting_router(self, router_id):
uri = '%s/routers/%s/l3-agents' % (self.uri_prefix, router_id)
resp, body = self.get(uri)
self.rest_client.expected_success(200, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def add_router_to_l3_agent(self, agent_id, router_id):
uri = '%s/agents/%s/l3-routers' % (self.uri_prefix, agent_id)
@@ -243,35 +243,35 @@ class NetworkClientJSON(network_client_base.NetworkClientBase):
resp, body = self.post(uri, body)
self.rest_client.expected_success(201, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def remove_router_from_l3_agent(self, agent_id, router_id):
uri = '%s/agents/%s/l3-routers/%s' % (
self.uri_prefix, agent_id, router_id)
resp, body = self.delete(uri)
self.rest_client.expected_success(204, resp.status)
return resp, body
return rest_client.ResponseBody(resp, body)
def list_dhcp_agent_hosting_network(self, network_id):
uri = '%s/networks/%s/dhcp-agents' % (self.uri_prefix, network_id)
resp, body = self.get(uri)
self.rest_client.expected_success(200, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def list_networks_hosted_by_one_dhcp_agent(self, agent_id):
uri = '%s/agents/%s/dhcp-networks' % (self.uri_prefix, agent_id)
resp, body = self.get(uri)
self.rest_client.expected_success(200, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def remove_network_from_dhcp_agent(self, agent_id, network_id):
uri = '%s/agents/%s/dhcp-networks/%s' % (self.uri_prefix, agent_id,
network_id)
resp, body = self.delete(uri)
self.rest_client.expected_success(204, resp.status)
return resp, body
return rest_client.ResponseBody(resp, body)
def create_ikepolicy(self, name, **kwargs):
post_body = {
@@ -286,7 +286,7 @@ class NetworkClientJSON(network_client_base.NetworkClientBase):
resp, body = self.post(uri, body)
self.rest_client.expected_success(201, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def update_extra_routes(self, router_id, nexthop, destination):
uri = '%s/routers/%s' % (self.uri_prefix, router_id)
@@ -300,7 +300,7 @@ class NetworkClientJSON(network_client_base.NetworkClientBase):
resp, body = self.put(uri, body)
self.rest_client.expected_success(200, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def delete_extra_routes(self, router_id):
uri = '%s/routers/%s' % (self.uri_prefix, router_id)
@@ -314,14 +314,14 @@ class NetworkClientJSON(network_client_base.NetworkClientBase):
resp, body = self.put(uri, body)
self.rest_client.expected_success(200, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def list_lb_pool_stats(self, pool_id):
uri = '%s/lb/pools/%s/stats' % (self.uri_prefix, pool_id)
resp, body = self.get(uri)
self.rest_client.expected_success(200, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def add_dhcp_agent_to_network(self, agent_id, network_id):
post_body = {'network_id': network_id}
@@ -330,7 +330,7 @@ class NetworkClientJSON(network_client_base.NetworkClientBase):
resp, body = self.post(uri, body)
self.rest_client.expected_success(201, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def insert_firewall_rule_in_policy(self, firewall_policy_id,
firewall_rule_id, insert_after="",
@@ -346,7 +346,7 @@ class NetworkClientJSON(network_client_base.NetworkClientBase):
resp, body = self.put(uri, body)
self.rest_client.expected_success(200, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)
def remove_firewall_rule_from_policy(self, firewall_policy_id,
firewall_rule_id):
@@ -357,4 +357,4 @@ class NetworkClientJSON(network_client_base.NetworkClientBase):
resp, body = self.put(uri, update_body)
self.rest_client.expected_success(200, resp.status)
body = json.loads(body)
return resp, body
return rest_client.ResponseBody(resp, body)

17
tempest/services/network/network_client_base.py
View File

@@ -13,6 +13,7 @@
import time
import urllib
from tempest.common import rest_client
from tempest.common.utils import misc
from tempest import config
from tempest import exceptions
@@ -113,7 +114,7 @@ class NetworkClientBase(object):
resp, body = self.get(uri)
result = {plural_name: self.deserialize_list(body)}
self.rest_client.expected_success(200, resp.status)
return resp, result
return rest_client.ResponseBody(resp, result)
return _list
@@ -123,7 +124,7 @@ class NetworkClientBase(object):
uri = '%s/%s' % (self.get_uri(plural), resource_id)
resp, body = self.delete(uri)
self.rest_client.expected_success(204, resp.status)
return resp, body
return rest_client.ResponseBody(resp, body)
return _delete
@@ -139,7 +140,7 @@ class NetworkClientBase(object):
resp, body = self.get(uri)
body = self.deserialize_single(body)
self.rest_client.expected_success(200, resp.status)
return resp, body
return rest_client.ResponseBody(resp, body)
return _show
@@ -151,7 +152,7 @@ class NetworkClientBase(object):
resp, body = self.post(uri, post_data)
body = self.deserialize_single(body)
self.rest_client.expected_success(201, resp.status)
return resp, body
return rest_client.ResponseBody(resp, body)
return _create
@@ -163,7 +164,7 @@ class NetworkClientBase(object):
resp, body = self.put(uri, post_data)
body = self.deserialize_single(body)
self.rest_client.expected_success(200, resp.status)
return resp, body
return rest_client.ResponseBody(resp, body)
return _update
@@ -189,7 +190,7 @@ class NetworkClientBase(object):
resp, body = self.post(uri, body)
body = {'networks': self.deserialize_list(body)}
self.rest_client.expected_success(201, resp.status)
return resp, body
return rest_client.ResponseBody(resp, body)
def create_bulk_subnet(self, subnet_list):
post_data = {'subnets': subnet_list}
@@ -198,7 +199,7 @@ class NetworkClientBase(object):
resp, body = self.post(uri, body)
body = {'subnets': self.deserialize_list(body)}
self.rest_client.expected_success(201, resp.status)
return resp, body
return rest_client.ResponseBody(resp, body)
def create_bulk_port(self, port_list):
post_data = {'ports': port_list}
@@ -207,7 +208,7 @@ class NetworkClientBase(object):
resp, body = self.post(uri, body)
body = {'ports': self.deserialize_list(body)}
self.rest_client.expected_success(201, resp.status)
return resp, body
return rest_client.ResponseBody(resp, body)
def wait_for_resource_deletion(self, resource_type, id):
"""Waits for a resource to be deleted."""

24
tempest/services/network/resources.py
View File

@@ -82,7 +82,7 @@ class DeletableSubnet(DeletableResource):
self._router_ids = set()
def update(self, *args, **kwargs):
_, result = self.client.update_subnet(subnet=self.id, *args, **kwargs)
result = self.client.update_subnet(subnet=self.id, *args, **kwargs)
super(DeletableSubnet, self).update(**result['subnet'])
def add_to_router(self, router_id):
@@ -108,9 +108,9 @@ class DeletableRouter(DeletableResource):
return self.update(external_gateway_info=dict())
def update(self, *args, **kwargs):
_, result = self.client.update_router(self.id,
*args,
**kwargs)
result = self.client.update_router(self.id,
*args,
**kwargs)
return super(DeletableRouter, self).update(**result['router'])
def delete(self):
@@ -121,15 +121,15 @@ class DeletableRouter(DeletableResource):
class DeletableFloatingIp(DeletableResource):
def refresh(self, *args, **kwargs):
_, result = self.client.show_floatingip(self.id,
*args,
**kwargs)
result = self.client.show_floatingip(self.id,
*args,
**kwargs)
super(DeletableFloatingIp, self).update(**result['floatingip'])
def update(self, *args, **kwargs):
_, result = self.client.update_floatingip(self.id,
*args,
**kwargs)
result = self.client.update_floatingip(self.id,
*args,
**kwargs)
super(DeletableFloatingIp, self).update(**result['floatingip'])
def __repr__(self):
@@ -183,5 +183,5 @@ class DeletableVip(DeletableResource):
self.client.delete_vip(self.id)
def refresh(self):
_, result = self.client.show_vip(self.id)
super(DeletableVip, self).update(**result['vip'])
result = self.client.show_vip(self.id)
super(DeletableVip, self).update(**result['vip'])

12
tempest/tests/cmd/test_verify_tempest_config.py
View File

@@ -222,9 +222,9 @@ class TestDiscovery(base.TestCase):
def test_verify_extensions_neutron(self):
def fake_list_extensions():
return (None, {'extensions': [{'alias': 'fake1'},
{'alias': 'fake2'},
{'alias': 'not_fake'}]})
return {'extensions': [{'alias': 'fake1'},
{'alias': 'fake2'},
{'alias': 'not_fake'}]}
fake_os = mock.MagicMock()
fake_os.network_client.list_extensions = fake_list_extensions
self.useFixture(mockpatch.PatchObject(
@@ -244,9 +244,9 @@ class TestDiscovery(base.TestCase):
def test_verify_extensions_neutron_all(self):
def fake_list_extensions():
return (None, {'extensions': [{'alias': 'fake1'},
{'alias': 'fake2'},
{'alias': 'not_fake'}]})
return {'extensions': [{'alias': 'fake1'},
{'alias': 'fake2'},
{'alias': 'not_fake'}]}
fake_os = mock.MagicMock()
fake_os.network_client.list_extensions = fake_list_extensions
self.useFixture(mockpatch.PatchObject(

20
tempest/tests/test_tenant_isolation.py
View File

@@ -92,24 +92,21 @@ class TestTenantIsolation(base.TestCase):
net_fix = self.useFixture(mockpatch.PatchObject(
iso_creds.network_admin_client,
'create_network',
return_value=({'status': 200},
{'network': {'id': id, 'name': name}})))
return_value={'network': {'id': id, 'name': name}}))
return net_fix
def _mock_subnet_create(self, iso_creds, id, name):
subnet_fix = self.useFixture(mockpatch.PatchObject(
iso_creds.network_admin_client,
'create_subnet',
return_value=({'status': 200},
{'subnet': {'id': id, 'name': name}})))
return_value={'subnet': {'id': id, 'name': name}}))
return subnet_fix
def _mock_router_create(self, id, name):
router_fix = self.useFixture(mockpatch.PatchObject(
json_network_client.NetworkClientJSON,
'create_router',
return_value=({'status': 200},
{'router': {'id': id, 'name': name}})))
return_value={'router': {'id': id, 'name': name}}))
return router_fix
@mock.patch('tempest.common.rest_client.RestClient')
@@ -242,12 +239,11 @@ class TestTenantIsolation(base.TestCase):
@mock.patch('tempest.common.rest_client.RestClient')
def test_network_cleanup(self, MockRestClient):
def side_effect(**args):
return ({'status': 200},
{"security_groups": [{"tenant_id": args['tenant_id'],
"name": args['name'],
"description": args['name'],
"security_group_rules": [],
"id": "sg-%s" % args['tenant_id']}]})
return {"security_groups": [{"tenant_id": args['tenant_id'],
"name": args['name'],
"description": args['name'],
"security_group_rules": [],
"id": "sg-%s" % args['tenant_id']}]}
iso_creds = isolated_creds.IsolatedCreds('test class',
password='fake_password')
# Create primary tenant and network