openstack
/
fuel-qa
Code Issues Proposed changes

Tests for IPv6 support in neutron 

Check that IPv6 addressing and routing is working in Fuel
Closes-bug: #1518979

Change-Id: I12cfdd6bdafa748dd40ceb5d06ec051d38210e67
This commit is contained in:
Alexey Stepanov 2016-01-15 20:14:50 +03:00
parent 40543622d5
commit 67f154ed33
6 changed files with 343 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
doc/base_tests.rst
View File

@ -232,6 +232,11 @@ Test Neutron VXLAN base
.. automodule:: fuelweb_test.tests.test_neutron_tun_base
:members:
Test Neutron IPv6 base functionality
------------------------------------
.. automodule:: fuelweb_test.tests.test_neutron_ipv6
:members:
Test Node reinstallation
------------------------
.. automodule:: fuelweb_test.tests.test_node_reinstallation

99
fuelweb_test/helpers/os_actions.py
View File

@ -64,6 +64,57 @@ class OpenStackActions(common.Common):
if servers:
return servers
def create_server(
self,
name=None,
security_groups=None,
flavor_id=None,
net_id=None,
timeout=100
):
""" Creates simple server, like in OSTF.
:param name: server name, if None -> test-serv + random suffix
:param security_groups: list, if None -> ssh + icmp v4 & icmp v6
:param flavor_id: micro_flavor if None
:param net_id: network id, could be omitted.
:param timeout: int=100
:return: Server, in started state
"""
def find_micro_flavor():
return [
flavor for flavor in self.nova.flavors.list()
if flavor.name == 'm1.micro'].pop()
if not name:
name = "test-serv" + str(random.randint(1, 0x7fffffff))
if not security_groups:
security_groups = [self.create_sec_group_for_ssh()]
if not flavor_id:
flavor_id = find_micro_flavor().id
nics = [{'net-id': net_id}] if net_id else None
srv = self.nova.servers.create(
name=name,
image=self._get_cirros_image().id,
flavor=flavor_id,
security_groups=[sec_group.name for sec_group in security_groups],
nics=nics)
try:
helpers.wait(
lambda: self.get_instance_detail(srv).status == "ACTIVE",
timeout=timeout)
return self.get_instance_detail(srv.id)
except TimeoutError:
logger.debug("Create server failed by timeout")
asserts.assert_equal(
self.get_instance_detail(srv).status,
"ACTIVE",
"Instance do not reach active state, current state"
" is {0}".format(self.get_instance_detail(srv).status))
def create_server_for_migration(self, neutron=True, scenario='',
timeout=100, file=None, key_name=None,
label=None, flavor=1, **kwargs):
@ -79,8 +130,7 @@ class OpenStackActions(common.Common):
image_id = self._get_cirros_image().id
security_group[self.keystone.tenant_id] =\
self.create_sec_group_for_ssh()
security_group = [security_group[
self.keystone.tenant_id].name]
security_group = [security_group[self.keystone.tenant_id].name]
if neutron:
net_label = label if label else 'net04'
@ -175,6 +225,13 @@ class OpenStackActions(common.Common):
'from_port': -1,
'to_port': -1,
'cidr': '0.0.0.0/0',
},
{
# ping6
'ip_protocol': 'icmp',
'from_port': -1,
'to_port': -1,
'cidr': '::/0',
}
]
@ -335,7 +392,9 @@ class OpenStackActions(common.Common):
if user:
return user
return self.keystone.users.create(
name=username, password=passw, tenant_id=tenant.id)
name=username,
password=passw,
tenant_id=tenant.id)
def update_user_enabled(self, user, enabled=True):
self.keystone.users.update_enabled(user, enabled)
@ -511,9 +570,12 @@ class OpenStackActions(common.Common):
body['network'].update(kwargs)
return self.neutron.create_network(body)
def create_subnet(self, subnet_name, network_id, cidr, ip_version=4):
def create_subnet(
self, subnet_name, network_id, cidr, ip_version=4, **kwargs):
body = {"subnet": {"name": subnet_name, "network_id": network_id,
"ip_version": ip_version, "cidr": cidr}}
if kwargs:
body['subnet'].update(kwargs)
subnet = self.neutron.create_subnet(body)
return subnet['subnet']
@ -530,3 +592,32 @@ class OpenStackActions(common.Common):
body["port_id"] = port_id
self.neutron.add_interface_router(router_id, body)
return None
def create_router(self, name, tenant):
"""Creates router at neutron.
:param name: str, router name
:param tenant: tenant
:return: router object
"""
external_network = None
for network in self.neutron.list_networks()["networks"]:
if network.get("router:external"):
external_network = network
if not external_network:
raise RuntimeError('Cannot find the external network.')
gw_info = {
"network_id": external_network["id"],
"enable_snat": True
}
router_info = {
"router": {
"name": name,
"external_gateway_info": gw_info,
"tenant_id": tenant.id
}
}
return self.neutron.create_router(router_info)['router']

14
fuelweb_test/models/fuel_web_client.py
View File

@ -993,7 +993,12 @@ class FuelWebClient(object):
return [n for n in nodes if set(roles) <= set(n[role_status])]
@logwrap
def get_ssh_for_node(self, node_name):
def get_node_ip_by_devops_name(self, node_name):
"""Get node ip by it's devops name (like "slave-01" and etc)
:param node_name: str
:return: str
"""
try:
node = self.get_nailgun_node_by_devops_node(
self.environment.d_env.get_node(name=node_name))
@ -1001,7 +1006,12 @@ class FuelWebClient(object):
node = self.get_nailgun_node_by_fqdn(node_name)
assert_true(node is not None,
'Node with name "{0}" not found!'.format(node_name))
return self.environment.d_env.get_ssh_to_remote(node['ip'])
return node['ip']
@logwrap
def get_ssh_for_node(self, node_name):
return self.environment.d_env.get_ssh_to_remote(
self.get_node_ip_by_devops_name(node_name))
@logwrap
def get_ssh_for_role(self, nodes_dict, role):

1
fuelweb_test/run_tests.py
View File

@ -49,6 +49,7 @@ def import_tests():
from tests import test_neutron # noqa
from tests import test_neutron_public # noqa
from tests import test_neutron_tun # noqa
from tests import test_neutron_ipv6 # noqa
from tests import test_pullrequest # noqa
from tests import test_services # noqa
from tests import test_ha_one_controller # noqa

2
fuelweb_test/tests/test_neutron.py
View File

@ -79,7 +79,7 @@ class NeutronVlan(TestBasic):
self.fuel_web.run_ostf(
cluster_id=cluster_id)
self.env.make_snapshot("deploy_neutron_vlan")
self.env.make_snapshot("deploy_neutron_vlan", is_make=True)
@test(groups=["neutron", "ha", "ha_neutron", "classic_provisioning"])

229
fuelweb_test/tests/test_neutron_ipv6.py Normal file
View File

@ -0,0 +1,229 @@
# Copyright 2016 Mirantis, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from proboscis.asserts import assert_equal
from proboscis import test
from paramiko import ChannelException
from devops.helpers.helpers import wait
from devops.error import TimeoutError
from fuelweb_test.helpers import os_actions
from fuelweb_test.helpers.decorators import log_snapshot_after_test
from fuelweb_test.tests.base_test_case import TestBasic
from fuelweb_test import logger
@test(groups=["thread_1", "neutron"])
class TestNeutronIPv6(TestBasic):
"""NeutronIPv6."""
@test(depends_on_groups=['deploy_neutron_vlan'],
groups=['deploy_neutron_ip_v6',
"nova", "nova-compute", "neutron_ipv6"])
@log_snapshot_after_test
def deploy_neutron_ip_v6(self):
"""Check IPv6 only functionality for Neutron VLAN
Scenario:
1. Revert deploy_neutron_vlan snapshot
2. Create two dualstack network IPv6 subnets
(should be in SLAAC mode,
address space should not intersect).
3. Create virtual router and set gateway.
4. Attach this subnets to the router.
5. Create a Security Group,
that allows SSH and ICMP for both IPv4 and IPv6.
6. Launch two instances, one for each network.
7. Lease a floating IP.
8. Attach Floating IP for main instance.
9. SSH to the main instance and ping6 another instance.
Duration 10m
Snapshot deploy_neutron_ip_v6
"""
self.show_step(1, initialize=True)
self.env.revert_snapshot("deploy_neutron_vlan")
cluster_id = self.fuel_web.get_last_created_cluster()
public_vip = self.fuel_web.get_public_vip(cluster_id)
logger.info('Public vip is %s', public_vip)
os_conn = os_actions.OpenStackActions(
controller_ip=public_vip,
user='simpleVlan',
passwd='simpleVlan',
tenant='simpleVlan'
)
tenant = os_conn.get_tenant('simpleVlan')
self.show_step(2)
net1 = os_conn.create_network(
network_name='net1',
tenant_id=tenant.id)['network']
net2 = os_conn.create_network(
network_name='net2',
tenant_id=tenant.id)['network']
subnet_1_v4 = os_conn.create_subnet(
subnet_name='subnet_1_v4',
network_id=net1['id'],
cidr='192.168.100.0/24',
ip_version=4)
subnet_1_v6 = os_conn.create_subnet(
subnet_name='subnet_1_v6',
network_id=net1['id'],
ip_version=6,
cidr="2001:db8:100::/64",
gateway_ip="2001:db8:100::1",
ipv6_ra_mode="slaac",
ipv6_address_mode="slaac")
subnet_2_v4 = os_conn.create_subnet(
subnet_name='subnet_2_v4',
network_id=net2['id'],
cidr='192.168.200.0/24',
ip_version=4)
subnet_2_v6 = os_conn.create_subnet(
subnet_name='subnet_2_v6',
network_id=net2['id'],
ip_version=6,
cidr="2001:db8:200::/64",
gateway_ip="2001:db8:200::1",
ipv6_ra_mode="slaac",
ipv6_address_mode="slaac")
self.show_step(3)
router = os_conn.create_router('test_router', tenant=tenant)
self.show_step(4)
os_conn.add_router_interface(
router_id=router["id"],
subnet_id=subnet_1_v4["id"])
os_conn.add_router_interface(
router_id=router["id"],
subnet_id=subnet_1_v6["id"])
os_conn.add_router_interface(
router_id=router["id"],
subnet_id=subnet_2_v4["id"])
os_conn.add_router_interface(
router_id=router["id"],
subnet_id=subnet_2_v6["id"])
self.show_step(5)
security_group = os_conn.create_sec_group_for_ssh()
self.show_step(6)
instance1 = os_conn.create_server(
name='instance1',
security_groups=[security_group],
net_id=net1['id'],
)
instance2 = os_conn.create_server(
name='instance2',
security_groups=[security_group],
net_id=net2['id'],
)
self.show_step(7)
self.show_step(8)
floating_ip = os_conn.assign_floating_ip(instance1)
floating_ip2 = os_conn.assign_floating_ip(instance2)
self.show_step(9)
instance1_ipv6 = [
addr['addr'] for addr in instance1.addresses[net1['name']]
if addr['version'] == 6].pop()
instance2_ipv6 = [
addr['addr'] for addr in instance2.addresses[net2['name']]
if addr['version'] == 6].pop()
logger.info(
'\ninstance1:\n'
'\tFloatingIP: {ip!s}\n'
'\tIPv6 address: {ipv6!s}'.format(
ip=floating_ip.ip,
ipv6=instance1_ipv6))
logger.info(
'\ninstance2:\n'
'\tFloatingIP: {ip!s}\n'
'\tIPv6 address: {ipv6!s}'.format(
ip=floating_ip2.ip,
ipv6=instance2_ipv6))
with self.fuel_web.get_ssh_for_node("slave-01") as remote:
def ssh_ready(vm_host):
try:
os_conn.execute_through_host(
ssh=remote,
vm_host=vm_host,
cmd="ls -la",
creds=("cirros", "cubswin:)")
)
return True
except ChannelException:
return False
for vm_host, hostname in (
(floating_ip.ip, instance1),
(floating_ip2.ip, instance2)
):
try:
wait(lambda: ssh_ready(vm_host), timeout=120)
except TimeoutError:
raise TimeoutError(
'ssh is not ready on host '
'{hostname:s} ({ip:s}) '
'at timeout 120s'.format(
hostname=hostname, ip=vm_host))
res = os_conn.execute_through_host(
ssh=remote,
vm_host=floating_ip.ip,
cmd="{ping:s} -q "
"-c{count:d} "
"-w{deadline:d} "
"-s{packetsize:d} "
"{dst_address:s}".format(
ping='ping6',
count=10,
deadline=20,
packetsize=1452,
dst_address=instance2_ipv6),
creds=("cirros", "cubswin:)")
)
logger.info('Ping results: \n\t{res:s}'.format(res=res['stdout']))
assert_equal(
res['exit_code'],
0,
'Ping failed with error code: {code:d}\n'
'\tSTDOUT: {stdout:s}\n'
'\tSTDERR: {stderr:s}'.format(
code=res['exit_code'],
stdout=res['stdout'],
stderr=res['stderr'],
))
self.env.make_snapshot('deploy_neutron_ip_v6')