Merge "[Tempest]: Changes done to add missing methods from upstream"

This commit is contained in:
Jenkins 2017-03-08 18:45:51 +00:00 committed by Gerrit Code Review
commit 08506554cd
4 changed files with 31 additions and 3 deletions

View File

@ -128,7 +128,6 @@ class TestDvrBasicOps(manager.NetworkScenarioTest):
subnets_client = self.subnets_client
network = self._create_network(
networks_client=networks_client,
routers_client=routers_client,
namestart=namestart)
router_kwargs = dict(client=routers_client, namestart=namestart)
@ -253,6 +252,18 @@ class TestDvrBasicOps(manager.NetworkScenarioTest):
ipatxt = ssh_client.exec_command("ip address")
return reg.findall(ipatxt)
def _list_subnets(self, *args, **kwargs):
"""List subnets using admin creds """
subnets_list = self.admin_manager.subnets_client.list_subnets(
*args, **kwargs)
return subnets_list['subnets']
def _list_ports(self, *args, **kwargs):
"""List ports using admin creds """
ports_list = self.admin_manager.ports_client.list_ports(
*args, **kwargs)
return ports_list['ports']
def _check_network_internal_connectivity(self, network,
should_connect=True):
"""

View File

@ -122,6 +122,12 @@ class TestProviderSecurityGroup(manager.NetworkScenarioTest):
client.delete_subnet, subnet['id'])
return subnet
def _list_ports(self, *args, **kwargs):
"""List ports using admin creds """
ports_list = self.admin_manager.ports_client.list_ports(
*args, **kwargs)
return ports_list['ports']
def get_port_id(self, network_id, subnet_id, instance):
_, instance_addr = instance["addresses"].items()[0]
instance_fixed_ip = instance_addr[0]["addr"]

View File

@ -76,8 +76,7 @@ class TestRouterNoNATOps(manager.NetworkScenarioTest):
self.servers = []
def _setup_network_topo(self, enable_snat=None):
self.security_group = self._create_security_group(
tenant_id=self.tenant_id)
self.security_group = self._create_security_group()
self.network = self._create_network()
self.subnet = self._create_subnet(self.network)
self.router = self._create_router(
@ -154,6 +153,12 @@ class TestRouterNoNATOps(manager.NetworkScenarioTest):
if address['version'] == CONF.validation.ip_version_for_ssh:
return address['addr']
def _list_ports(self, *args, **kwargs):
"""List ports using admin creds """
ports_list = self.admin_manager.ports_client.list_ports(
*args, **kwargs)
return ports_list['ports']
def _check_network_internal_connectivity(self, network,
should_connect=True):
floating_ip, server = self.floating_ip_tuple

View File

@ -199,6 +199,12 @@ class TestMicroSegmentationOps(manager.NetworkScenarioTest):
def _get_server_key(self, server):
return self.keypairs[server['key_name']]['private_key']
def _list_ports(self, *args, **kwargs):
"""List ports using admin creds """
ports_list = self.admin_manager.ports_client.list_ports(
*args, **kwargs)
return ports_list['ports']
def _check_network_internal_connectivity(self, network, fip_tuple,
should_connect=True):
floating_ip, server = fip_tuple