Pytest: Add networking test cases

1. 233-Verify-VSwitch-Settings-Cli.robot
2. 237-Online-Static-Configuration.robot
3. 238-Show-Data-Network-Api.robot
4. 239-Show-Data-Network-Range-Api.robot
5. 240-Add-Security-Group-To-Instance.robot
6. 241-Delete-Security-Group.robot
7. 242-Modify-Security-Group-Rule.robot
8. 243-Packet-Type-Security-Rule-Enforcement.robot
9. 244-Modify-CPU-Parameters.robot
10. 245-Multiple-Float-IPs-On-Multipe-Interfaces.robot
11. 246-Modify-OAM-Subnet.robot
12. 249-Interface-Not-Deleted-On-Unlock-Host.robot
13. 250-Distributed-Virtual-Router-Integrate-VM.robot
14. 251-Verify-Ethernet-OAM-Inteface-Is-Updated.robot
15. 253-Provider-Network-Type.robot
16. 255-Verify-DVR-Router-Deletion.robot
17. 256-Multiple-Ports-Same-Network.robot
18. 266-Change-OAM-Ip-And-Reinstall-Controller.robot
19. 329-QoS-Policy-Creation-Api.robot
20. 354-Create-And-Delete-Trunk.robot
21. 364-Create-Trunk-Childport-Equal-Mac-address.robot
22. 376-Verify-Horizon-Network-Topology-Config-Details.robot
23. 377-Horizon-Network-Topology-Select-Data-Networklist.robot
24. 395-Horizon-Topology-Provider-Network-Setup-Failure.robot
25. 401-Test-Verify-Compatible-NTP-Configuration.robot
26. 402-Not-Compatible-Configuration-External-OAM.robot
27. 403-Configuration-Validation-External-OAM.robot
28. 405-Trunk-With-VM-Lifecycle.robot
29. 411-Test-NAT-On-Router.robot
30. 419-Vxlan-Data-Network-Valid-Port-Creation.robot
31. 421-Vxlan-DataInterface-Address-Mode-Set-To-Static.robot
32. 425-LLDP-Accelerated-NIC.robot
33. 431-Vxlan-Interface-Address-Mode-Modifiable-When-Host-Locked.robot
34. 432-Vxlan-Data-Network-With-Valid-Multicast-Address.robot
35. 433-Vxlan-Data-Interface-Duplicate-Address-Not-Allowed.robot
36. 434-Verify-Vxlan-Data-Network-Segment-Range-Overlap.robot

Signed-off-by: Yong Fu <fuyong@neusoft.com>
Change-Id: Ic179bdb5734e128dda6cc4d181562c360204720f
This commit is contained in:
Yong Fu 2021-05-27 10:31:08 +08:00
parent 005b2d5dc2
commit f860297dd5
39 changed files with 2238 additions and 4 deletions

View File

@ -1,3 +1,12 @@
from pytest import fixture, skip
from testfixtures.resource_mgmt import *
from testfixtures.resource_create import *
from testfixtures.config_host import *
from keywords import system_helper
from utils.tis_log import LOG
@fixture(scope='session')
def no_aio_system():
LOG.fixture_step("(Session) Skip if AIO system")
if system_helper.is_aio_system():
skip('skip if AIO system')

View File

@ -0,0 +1,61 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Verify packet type security rule enforcement tunder traffic
# add security group to running VM instance
# delete security group associated with VM.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
import os
from pytest import fixture, mark
from consts.auth import Tenant
from consts.stx import GuestImages
from keywords import network_helper, nova_helper, glance_helper, vm_helper
from utils import cli
from utils.clients.ssh import ControllerClient
@fixture(scope="module")
def launch_instance():
net_id = network_helper.create_network(name='vm-net', cleanup="module")[1]
network_helper.create_subnet(name='vm-subnet', network='vm-net',
subnet_range='10.0.0.0/24', dhcp=True,
ip_version=4, cleanup="module")
net_id_list = [{"net-id": net_id}]
fl_id = nova_helper.create_flavor(name='vm-flavor', vcpus=1, ram=2048, root_disk=4,
is_public=True, add_default_specs=False, cleanup="module")[1]
im_id = glance_helper.create_image(name="vm-image", source_image_file=os.path.join(
GuestImages.DEFAULT["image_dir"], "cirros-0.4.0-x86_64-disk.img"), disk_format="qcow2",
cleanup="module")[1]
vm_id = vm_helper.boot_vm(name="vm-0", flavor=fl_id, nics=net_id_list, source="image",
source_id=im_id, cleanup="module")[1]
return vm_id
@mark.networking
def test_add_security_group_to_instance(launch_instance):
"""
240-Add-Security-Group-To-Instance.robot
241-Delete-Security-Group.robot
Args:
launch_instance:
Returns:
"""
con_ssh = ControllerClient.get_active_controller()
network_helper.create_security_group('test_group', project='admin', cleanup='module')
# Add Security Group To VM
vm_helper.add_security_group(launch_instance, 'test_group')
code = network_helper.delete_security_group('test_group', fail_ok=True)[0]
assert code != 0
cli.openstack('server remove security group', '{} test_group'.format(launch_instance),
ssh_client=con_ssh, auth_info=Tenant.get('admin'))

View File

@ -0,0 +1,195 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Change oam ip addresses then perform reinstall controllers.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
import time
from pytest import mark
from consts.auth import Tenant, HostLinuxUser
from consts.stx import EventLogID
from keywords import system_helper, host_helper, container_helper, storage_helper, kube_helper
from utils import cli, exceptions, table_parser
from utils.clients.ssh import ControllerClient, SSHClient, CONTROLLER_PROMPT
from utils.tis_log import LOG
def wait_for_host_delete_status(host):
LOG.info("waiting for {} to delete".format(host))
end_time = time.time() + 300
exists = system_helper.host_exists(host)
while time.time() < end_time:
if not exists:
LOG.info("{} has been deleted".format(host))
return 0
time.sleep(20)
exists = system_helper.host_exists(host)
err_msg = "Timed out waiting for {} to delete".format(host)
raise exceptions.VMTimeout(err_msg)
def wait_for_host_install_status(host):
LOG.info("waiting for {} install_state status: completed".format(host))
end_time = time.time() + 2400
current_status = system_helper.get_host_values(host, "install_state")[0]
while time.time() < end_time:
if current_status == "completed":
LOG.info("host status has reached completed")
return 0
time.sleep(30)
current_status = system_helper.get_host_values(host, "install_state")[0]
err_msg = "Timed out waiting for {} install_state status: completed. {} " \
"install_state status: {}".format(host, host, current_status)
raise exceptions.VMTimeout(err_msg)
def unlock_host(host, con_ssh):
application_status = container_helper.get_apps(application="stx-openstack")[0]
if application_status == "applying":
container_helper.abort_app("stx-openstack")
host_helper.unlock_host(host, con_ssh=con_ssh, available_only=False,
check_hypervisor_up=False, check_webservice_up=False,
check_subfunc=False, check_containers=False)
def swact_host(host):
con_ssh = ControllerClient.get_active_controller()
if host == 'controller-0':
oam_ip = 'oam_c1_ip'
else:
oam_ip = 'oam_c0_ip'
cmd = "oam-show |grep {} | awk '{{print $4}}'".format(oam_ip)
oam_c1_ip = cli.system(cmd)[1]
node_ssh = SSHClient(oam_c1_ip, HostLinuxUser.get_user(),
HostLinuxUser.get_password(),
CONTROLLER_PROMPT)
exitcode, msg = cli.system('host-swact', host, ssh_client=con_ssh,
fail_ok=False, auth_info=Tenant.get('admin_platform'))
assert exitcode == 0, msg
time.sleep(120)
con_ssh.close()
node_ssh.connect(retry=True, retry_timeout=30)
ControllerClient.set_active_controller(node_ssh)
def reinstall_controller_node():
con_ssh = ControllerClient.get_active_controller()
auth_info = Tenant.get('admin_platform')
LOG.info("Provide OAM Network Interface")
table_ = table_parser.table(cli.system("interface-network-list", "controller-0",
ssh_client=con_ssh,
auth_info=auth_info)[1])
# Get OAM interface
oam_if = table_parser.get_values(table_, "ifname", **{"network_name": "oam"})[0]
args = " -n oam0 -c platform controller-1 {}".format(oam_if)
cli.system('host-if-modify', args, ssh_client=con_ssh,
auth_info=auth_info)
cli.system("interface-network-assign", "controller-1 oam0 oam", ssh_client=con_ssh,
auth_info=auth_info)
# Setup Cluster-host Interfaces
LOG.info("Setup Cluster-host Interfaces")
cli.system("interface-network-assign", "controller-1 mgmt0 cluster-host", ssh_client=con_ssh,
auth_info=auth_info)
if system_helper.is_aio_duplex():
# Duplex
LOG.info("Configure data interfaces for controller-1.")
# Get Interface UUID
table_ = table_parser.table(cli.system("host-if-list", "{} -a".format("controller-1"))[1])
data_uuid = table_parser.get_values(table_, "uuid", **{"class": "None"})
# Add Interface To Data Network
args0 = "-m 1500 -n data0 -c data controller-1 {}".format(data_uuid[0])
args1 = "-m 1500 -n data1 -c data controller-1 {}".format(data_uuid[-1])
cli.system('host-if-modify', args0, ssh_client=con_ssh, auth_info=auth_info)
cli.system('host-if-modify', args1, ssh_client=con_ssh, auth_info=auth_info)
cli.system("interface-datanetwork-assign", "controller-1 {} {}"
.format(data_uuid[0], "physnet0"),
ssh_client=con_ssh, auth_info=auth_info)
cli.system("interface-datanetwork-assign", "controller-1 {} {}"
.format(data_uuid[-1], "physnet1"),
ssh_client=con_ssh, auth_info=auth_info)
LOG.info("Set up disk partition for nova-local volume group")
rootfs = system_helper.get_host_values("controller-1", "rootfs_device")[0]
uuid = storage_helper.get_host_disks("controller-1", **{"device_node": rootfs})[0]
# Add an OSD on controller-0 for Ceph
disk_uuid = storage_helper.get_host_disks("controller-1", **{"device_node": "/dev/sdb"})[0]
cli.system("host-stor-add", " controller-1 {}".format(disk_uuid),
ssh_client=con_ssh, auth_info=auth_info)
# Set up disk partition for nova-local volume group
args = " -t lvm_phys_vol controller-1 {} 100".format(uuid)
out = cli.system('host-disk-partition-add', args, ssh_client=con_ssh, auth_info=auth_info)[1]
new_uuid = table_parser.get_value_two_col_table(table_parser.table(out), "uuid")
# Add Local Volume Group
cli.system("host-lvg-add", "controller-1 nova-local", ssh_client=con_ssh,
auth_info=auth_info)
# Add Physical Volume
cli.system("host-pv-add", "controller-1 nova-local {}".format(new_uuid), ssh_client=con_ssh,
auth_info=auth_info)
# Enable Containerized Services
labels = ["openstack-control-plane", "openstack-compute-node", "openvswitch", "sriov"]
host_helper.assign_host_labels("controller-1", labels, unlock=False)
unlock_host("controller-1", con_ssh)
else:
# MN-Local
# Add ODS To Tier
host_helper.assign_host_labels("controller-1", ["openstack-control-plane"], unlock=False)
unlock_host("controller-1", con_ssh)
@mark.networking
def test_change_oam_ip_and_reinstall_controller(no_simplex):
"""
266-Change-OAM-Ip-And-Reinstall-Controller.robot
Args:
no_simplex:
Returns:
"""
con_ssh = ControllerClient.get_active_controller()
auth_info = Tenant.get('admin_platform')
cli.system('oam-modify', 'oam_c1_ip=10.10.10.10', ssh_client=con_ssh, auth_info=auth_info)
out = system_helper.wait_for_alarm(alarm_id=EventLogID.CONFIG_OUT_OF_DATE, timeout=360)[0]
assert out, 'alarm 250.001 not found'
host = 'controller-1'
host_helper.lock_host(host)
output = con_ssh.exec_sudo_cmd('dmesg | grep -i paravirtualized')[1]
if 'KVM' in output and 'bare hardware' not in output:
with host_helper.ssh_to_host(host) as host_ssh:
LOG.info("Clear partition information")
host_ssh.exec_sudo_cmd("dd if=/dev/zero of=/dev/sda bs=512 count=1")
cli.system('host-delete', host, ssh_client=con_ssh, auth_info=auth_info)
wait_for_host_delete_status(host)
time.sleep(60)
table_ = table_parser.table(
cli.system('host-list', ssh_client=con_ssh, auth_info=auth_info)[1])
host_id = table_parser.get_values(table_, "id", **{"hostname": "None"})[0]
LOG.info("host id is {}".format(host_id))
cli.system('host-update', '{} personality=controller hostname=controller-1'.format(host_id),
ssh_client=con_ssh, auth_info=auth_info)
wait_for_host_install_status(host)
reinstall_controller_node()
time.sleep(60)
swact_host("controller-0")
host_helper.lock_unlock_hosts("controller-0")
swact_host("controller-1")
application_status = container_helper.get_apps(application="stx-openstack")[0]
if application_status == "applying":
container_helper.abort_app(app_name="stx-openstack")
pods_status = kube_helper.wait_for_pods_healthy(namespace="openstack", timeout=20,
con_ssh=con_ssh, fail_ok=True)
if not pods_status:
container_helper.remove_app(app_name="stx-openstack", applied_timeout=600)
container_helper.apply_app(app_name="stx-openstack", applied_timeout=3600,
check_interval=30, wait_for_alarm_gone=False)
assert EventLogID.CONFIG_OUT_OF_DATE not in \
system_helper.get_alarms(fields=("Alarm ID",), alarm_id=EventLogID.CONFIG_OUT_OF_DATE)

View File

@ -0,0 +1,28 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Configuration validation External OAM.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
from pytest import mark
from keywords import system_helper
@mark.networking
def test_configuration_validation_external_oam():
"""
403-Configuration-Validation-External-OAM.robot
Args:
Returns:
"""
message = system_helper.modify_oam_ips(**{'oam_subnet': '10.10.0.0/164'}, fail_ok=True)[1]
assert 'Invalid subnet oam_subnet 10.10.0.0/164.Please configure a valid' in message

View File

@ -0,0 +1,69 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# This test case creates a trunk network and assosiate port to it.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
import getpass
import os
import yaml
from pytest import mark
from consts.auth import HostLinuxUser, Tenant
from keywords import container_helper, network_helper, common
from utils import cli, table_parser
from utils.clients.ssh import ControllerClient
desired_caps = {"conf": {"neutron": {"DEFAULT": {"service_plugins": ['router', 'trunk']}}}}
@mark.networking
def test_create_and_delete_trunk(no_simplex):
"""
354-Create-And-Delete-Trunk.robot
Args:
no_simplex:
Returns:
"""
con_ssh = ControllerClient.get_active_controller()
try:
net_id = network_helper.create_network(network_type='vlan',
name='net-a', cleanup="function")[1]
network_helper.create_subnet(net_id, subnet_range='192.168.3.0/24',
name='subnet-a', cleanup="function")
user_name = getpass.getuser()
yaml_path = os.path.join('/home/{}/'.format(user_name), "neutron-overrides-trunk.yaml")
with open(yaml_path, "w", encoding="utf-8") as f:
yaml.dump(desired_caps, f)
common.scp_from_test_server_to_active_controller(yaml_path, HostLinuxUser.get_home())
app_dir = HostLinuxUser.get_home()
file_path = os.path.join(app_dir, "neutron-overrides-trunk.yaml")
container_helper.update_helm_override(app_name='stx-openstack', chart='neutron',
namespace='openstack', yaml_file=file_path)
container_helper.apply_app(app_name='stx-openstack', applied_timeout=1200,
check_interval=30)
network_helper.create_port(net_id, 'parentport', cleanup="function")
network_helper.create_port(net_id, 'subport', cleanup="function")
args = '--parent-port parentport --subport port=subport,segmentation-type=vlan,' \
'segmentation-id=123 trunk00'
code, output = cli.openstack('network trunk create', args,
ssh_client=con_ssh, fail_ok=False,
auth_info=Tenant.get('admin'))
table_ = table_parser.table(output)
trunk_id = table_parser.get_value_two_col_table(table_, 'id')
network_helper.unset_trunk(trunk_id, sub_ports="subport")
network_helper.delete_trunks(trunk_id)
finally:
cli.system('helm-override-delete', "stx-openstack neutron openstack",
ssh_client=con_ssh, auth_info=Tenant.get('admin_platform'))
container_helper.apply_app(app_name='stx-openstack', applied_timeout=1200,
check_interval=30)

View File

@ -0,0 +1,75 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# This test case creates a trunk network and assign parent port
# and and child port with the same MAC address.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
import getpass
import os
import yaml
from pytest import mark
from consts.auth import HostLinuxUser, Tenant
from keywords import network_helper, common, container_helper
from utils import cli, table_parser
from utils.clients.ssh import ControllerClient
desired_caps = {"conf": {"neutron": {"DEFAULT": {"service_plugins": ['router', 'trunk']}}}}
@mark.networking
def test_create_trunk_childport_equal_mac_address(no_simplex):
"""
364-Create-Trunk-Childport-Equal-Mac-address.robot
Args:
no_simplex:
Returns:
"""
con_ssh = ControllerClient.get_active_controller()
net_id_a = network_helper.create_network(network_type='vlan',
name='net-a', cleanup="function")[1]
network_helper.create_subnet(net_id_a, subnet_range='192.168.3.0/24',
name='subnet-a', cleanup="function")
net_id_b = network_helper.create_network(network_type='vlan',
name='net-b', cleanup="function")[1]
network_helper.create_subnet(net_id_b, subnet_range='192.168.4.0/24',
name='subnet-b', cleanup="function")
user_name = getpass.getuser()
yaml_path = os.path.join('/home/{}/'.format(user_name), "neutron-overrides-trunk.yaml")
with open(yaml_path, "w", encoding="utf-8") as f:
yaml.dump(desired_caps, f)
common.scp_from_test_server_to_active_controller(yaml_path, HostLinuxUser.get_home())
app_dir = HostLinuxUser.get_home()
file_path = os.path.join(app_dir, "neutron-overrides-trunk.yaml")
container_helper.update_helm_override(app_name='stx-openstack', chart='neutron',
namespace='openstack', yaml_file=file_path)
container_helper.apply_app(app_name='stx-openstack', applied_timeout=1200,
check_interval=30)
parent_port = network_helper.create_port(net_id_a, 'parentport', cleanup="function")[1]
mac_address = network_helper.get_port_values(parent_port, ('mac_address',))[0]
child_port = network_helper.create_port(net_id_b, 'subport',
mac_addr=mac_address, cleanup="function")[1]
args = '--parent-port parentport --subport port=subport,segmentation-type=vlan,' \
'segmentation-id=123 trunk00'
code, output = cli.openstack('network trunk create', args,
ssh_client=con_ssh, fail_ok=False,
auth_info=Tenant.get('admin'))
table_ = table_parser.table(output)
trunk_id = table_parser.get_value_two_col_table(table_, 'id')
child_address = network_helper.get_port_values(child_port, ('mac_address',))[0]
assert child_address == child_address
network_helper.delete_trunks(trunk_id)
cli.system('helm-override-delete', "stx-openstack neutron openstack",
ssh_client=con_ssh, auth_info=Tenant.get('admin_platform'))
container_helper.apply_app(app_name='stx-openstack', applied_timeout=1200,
check_interval=30)

View File

@ -0,0 +1,80 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# This test to create Distributed Virtual router to be integrating VM.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
import os
from pytest import fixture, mark
from consts.stx import GuestImages
from keywords import network_helper, nova_helper, glance_helper, vm_helper, host_helper
from utils import cli
from utils.tis_log import LOG
@fixture(scope="module")
def launch_instance():
net_id = network_helper.get_networks(full_name='public-net0')[0]
net_id_list = [{"net-id": net_id}]
fl_id = nova_helper.create_flavor(name='vm-flavor', vcpus=1, ram=2048, root_disk=4,
is_public=True, add_default_specs=False, cleanup="module")[1]
im_id = glance_helper.create_image(name="vm-image", source_image_file=os.path.join(
GuestImages.DEFAULT["image_dir"], "cirros-0.4.0-x86_64-disk.img"), disk_format="qcow2",
cleanup="module")[1]
vm_id = vm_helper.boot_vm(name="vm-0", flavor=fl_id, nics=net_id_list, source="image",
source_id=im_id, cleanup="module")[1]
return vm_id
def check_ping_instance(ip, netid):
LOG.info('start to check ping instance')
host_list = host_helper.get_hypervisors()
for host in host_list:
with host_helper.ssh_to_host(host) as node_ssh:
cmd = 'ip netns | grep --color=never {}'.format(netid)
rc, ns = node_ssh.exec_cmd(cmd=cmd)
if ns and netid in ns.split()[0]:
target = host
netns = ns.split()[0]
break
with host_helper.ssh_to_host(target) as node_ssh:
ping_cmd = "ip netns exec {} ping -c 5 {} " \
.format(netns, ip)
node_ssh.send_sudo(cmd=ping_cmd)
index = node_ssh.expect(['5 received, 0% packet loss'])
assert index == 0, "Ping instance from instance failed"
@mark.networking
def test_distributed_virtual_router_integrate_vm(no_aio_system, launch_instance):
"""
250-Distributed-Virtual-Router-Integrate-VM.robot
Args:
no_aio_system:
launch_instance:
Returns:
"""
external_id = network_helper.get_networks(full_name='external-net0')[0]
subnet_uuid = network_helper.get_subnets(full_name='external-subnet0')[0]
network_helper.create_router('router1', cleanup="module", distributed=True)
vm_subnet = network_helper.get_subnets(full_name='public-subnet0')[0]
network_helper.add_router_interface('router1', subnet=vm_subnet)
network_helper.set_router('router1', external_gateway=external_id)
float_ip = network_helper.create_floating_ip(external_net=external_id,
subnet=subnet_uuid, cleanup="module")[1]
cli.openstack('server add floating ip', "{} {}".format(launch_instance, float_ip))
out = cli.openstack("router show", "router1 | grep 'enable_snat' | awk '{{print$7}}'")[1]
LOG.info("out is {}".format(out))
net_id = network_helper.get_networks(full_name='public-net0')[0]
check_ping_instance(float_ip, net_id)

View File

@ -0,0 +1,38 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# In Horizon data network page select each data network.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
from pytest import mark
from testfixtures.horizon import *
from utils.tis_log import LOG
from utils.horizon.pages.admin.platform import datanetworks
@mark.networking
def test_horizon_network_topology_select_data_networklist(tenant_home_pg_container_starlingx):
"""
377-Horizon-Network-Topology-Select-Data-Networklist.robot
Args:
tenant_home_pg_container_starlingx:
Returns:
"""
LOG.info("Go to Admin Data Networks")
datanetworks_pg = datanetworks.DatanetworksPage(
tenant_home_pg_container_starlingx.driver, tenant_home_pg_container_starlingx.port)
datanetworks_pg.go_to_target_page()
datanetworks_pg.go_to_datanetwork_detail_page('physnet0')
datanetworks_pg.go_to_target_page()
datanetworks_pg.go_to_datanetwork_detail_page('physnet1')
datanetworks_pg.go_to_target_page()

View File

@ -0,0 +1,38 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# In Horizon network page to create data network with
# existing network name, It throughs error while creation of network.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
from pytest import mark
from testfixtures.horizon import *
from utils.horizon.regions import messages
from utils.tis_log import LOG
from utils.horizon.pages.admin.platform import datanetworks
@mark.networking
def test_horizon_topology_provider_network_setup_failure(tenant_home_pg_container_starlingx):
"""
395-Horizon-Topology-Provider-Network-Setup-Failure.robot
Args:
tenant_home_pg_container_starlingx:
Returns:
"""
LOG.info("Go to Admin Data Networks")
datanetworks_pg = datanetworks.DatanetworksPage(
tenant_home_pg_container_starlingx.driver, tenant_home_pg_container_starlingx.port)
datanetworks_pg.go_to_target_page()
datanetworks_pg.create_datanetwork('physnet0', network_type='vlan')
assert datanetworks_pg.find_message_and_dismiss(messages.ERROR)

View File

@ -0,0 +1,38 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Test to Interface can't be deleted when host is unlocked.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
from pytest import mark
from consts.auth import Tenant
from keywords import host_helper, system_helper
from utils import cli
from utils.clients.ssh import ControllerClient
@mark.networking
def test_interface_not_deleted_on_unlock_host():
"""
249-Interface-Not-Deleted-On-Unlock-Host.robot
Args:
Returns:
"""
host = 'controller-0'
con_ssh = ControllerClient.get_active_controller()
status = system_helper.get_host_values(host, 'administrative')[0]
if status == 'locked':
host_helper.unlock_host(host)
uuid = host_helper.get_host_interfaces(host, field='uuid')[0]
code, output = cli.system('host-if-delete', "{} {}".format(host, uuid), fail_ok=True,
ssh_client=con_ssh, auth_info=Tenant.get('admin_platform'))
assert code != 0 and 'Host must be locked' in output

View File

@ -0,0 +1,47 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# LLDP TLVs accelerated Data NIC bonded.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
from pytest import mark
from keywords import system_helper, host_helper
from utils import cli
from utils.tis_log import LOG
@mark.networking
def test_lldp_accelerated_nic(no_aio_system):
"""
425-LLDP-Accelerated-NIC.robot
Args:
no_aio_system:
Returns:
"""
if system_helper.is_aio_system():
hosts = system_helper.get_controllers()
else:
hosts = system_helper.get_computes()
for host in hosts:
lldps = host_helper.get_host_lldp_agents(host, field="local_port")
for lldp in lldps:
args = "{} -a | grep {} | awk '{{print$4}}'".format(host, lldp)
interface = cli.system('host-if-list', args)[1]
if 'data' not in interface:
continue
accelerated = host_helper.get_host_interface_values(host, interface, fields='accelerated')[0]
LOG.info("accelerated is {}".format(accelerated))
assert '[True]' == accelerated
port = host_helper.get_host_interface_values(host, interface, fields='ports')[0]
port = port.split('\'')[1]
with host_helper.ssh_to_host(host) as node_ssh:
node_ssh.exec_cmd('ip addr show {}'.format(port), fail_ok=False)

View File

@ -0,0 +1,40 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Test Case to modify the function from cpu in compute nodes.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
from pytest import mark
from consts.auth import Tenant
from keywords import host_helper, system_helper
from utils import table_parser, cli
from utils.clients.ssh import ControllerClient
@mark.networking
def test_modify_cpu_parameters(no_aio_system):
"""
244-Modify-CPU-Parameters.robot
Args:
no_aio_system:
Returns:
"""
compute = system_helper.get_computes()[0]
try:
con_ssh = ControllerClient.get_active_controller()
table_ = host_helper.get_host_cpu_list_table(host=compute, con_ssh=con_ssh)
phy_cores = table_parser.get_values(table_, 'phy_core', **{'assigned_function': 'vswitch'})
host_helper.lock_host(compute, force=True)
cli.system('host-cpu-modify', '-f vswitch -p0 {} {}'.format(len(phy_cores) + 1, compute),
ssh_client=con_ssh, auth_info=Tenant.get('admin_platform'))
finally:
host_helper.unlock_host(compute)

View File

@ -0,0 +1,76 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Test to check OAM can be changed correctly.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
import time
from pytest import mark
from consts.auth import Tenant, HostLinuxUser
from keywords import host_helper, system_helper, kube_helper, container_helper
from utils import cli
from utils.clients.ssh import ControllerClient, SSHClient, CONTROLLER_PROMPT
def swact_host(host):
con_ssh = ControllerClient.get_active_controller()
if host == 'controller-0':
oam_ip = 'oam_c1_ip'
else:
oam_ip = 'oam_c0_ip'
cmd = "oam-show |grep {} | awk '{{print $4}}'".format(oam_ip)
oam_c1_ip = cli.system(cmd)[1]
node_ssh = SSHClient(oam_c1_ip, HostLinuxUser.get_user(),
HostLinuxUser.get_password(),
CONTROLLER_PROMPT)
exitcode, msg = cli.system('host-swact', host, ssh_client=con_ssh,
fail_ok=False, auth_info=Tenant.get('admin_platform'))
assert exitcode == 0, msg
time.sleep(120)
con_ssh.close()
node_ssh.connect(retry=True, retry_timeout=30)
ControllerClient.set_active_controller(node_ssh)
@mark.networking
def test_modify_oam_subnet(no_simplex):
"""
246-Modify-OAM-Subnet.robot
Args:
no_simplex:
Returns:
"""
con_ssh = ControllerClient.get_active_controller()
auth_info = Tenant.get('admin_platform')
cmd = "oam-show |grep oam_c1_ip | awk '{{print $4}}'"
oam_ip = cli.system(cmd)[1]
cli.system('oam-modify', 'oam_c1_ip=10.10.10.11', ssh_client=con_ssh, auth_info=auth_info)
host_helper.lock_unlock_hosts('controller-1')
assert system_helper.get_host_values('controller-1', fields='config_status')[0] is None
swact_host('controller-0')
host_helper.lock_unlock_hosts('controller-0')
assert system_helper.get_host_values('controller-0', fields='config_status')[0] is None
swact_host('controller-1')
pods_status = kube_helper.wait_for_pods_healthy(namespace="openstack", timeout=300, fail_ok=True)
if not pods_status:
# container_helper.remove_app(app_name="stx-openstack", applied_timeout=600)
container_helper.apply_app(app_name="stx-openstack", applied_timeout=3600,
check_interval=30, wait_for_alarm_gone=False)
con_ssh = ControllerClient.get_active_controller()
cli.system('oam-modify', 'oam_c1_ip={}'.format(oam_ip), ssh_client=con_ssh, auth_info=auth_info)
host_helper.lock_unlock_hosts('controller-1')
assert system_helper.get_host_values('controller-1', fields='config_status')[0] is None
swact_host('controller-0')
host_helper.lock_unlock_hosts('controller-0')
assert system_helper.get_host_values('controller-0', fields='config_status')[0] is None
swact_host('controller-1')

View File

@ -0,0 +1,54 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Verify packet type security rule enforcement under traffic
# modify rule while VM instance running.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
import os
from pytest import fixture, mark
from consts.stx import GuestImages
from keywords import network_helper, vm_helper, nova_helper, glance_helper
@fixture(scope="module")
def launch_instance():
net_id = network_helper.create_network(name='vm-net', cleanup="module")[1]
network_helper.create_subnet(name='vm-subnet', network='vm-net',
subnet_range='10.0.0.0/24', dhcp=True,
ip_version=4, cleanup="module")
net_id_list = [{"net-id": net_id}]
fl_id = nova_helper.create_flavor(name='vm-flavor', vcpus=1, ram=2048, root_disk=4,
is_public=True, add_default_specs=False, cleanup="module")[1]
im_id = glance_helper.create_image(name="vm-image", source_image_file=os.path.join(
GuestImages.DEFAULT["image_dir"], "cirros-0.4.0-x86_64-disk.img"), disk_format="qcow2",
cleanup="module")[1]
vm_id = vm_helper.boot_vm(name="vm-0", flavor=fl_id, nics=net_id_list, source="image",
source_id=im_id, cleanup="module")[1]
return vm_id
@mark.networking
def test_modify_security_group_rule(launch_instance):
"""
242-Modify-Security-Group-Rule.robot
Args:
launch_instance:
Returns:
"""
network_helper.create_security_group('test_group', project='admin', cleanup='module')
network_helper.create_security_group_rule(group='test_group', dst_port='22:22', project='admin',
protocol='tcp', remote_ip='0.0.0.0/0',
cleanup='module')
# Add Security Group To VM
vm_helper.add_security_group(launch_instance, 'test_group')

View File

@ -0,0 +1,69 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Test to verify DVR can be deleted.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
import os
from pytest import fixture, mark
from consts.stx import GuestImages
from keywords import network_helper, nova_helper, glance_helper, vm_helper
from utils import cli
@fixture(scope="module")
def launch_instance():
net_id = network_helper.create_network(name='vm-net', cleanup="module")[1]
subnet_id = network_helper.create_subnet(name='vm-subnet', network='vm-net',
subnet_range='10.0.0.0/24', dhcp=True,
ip_version=4, cleanup="module")[1]
net_id_list_0 = [{"net-id": net_id}]
fl_id = nova_helper.create_flavor(name='vm-flavor', vcpus=1, ram=2048, root_disk=4,
is_public=True, add_default_specs=False, cleanup="module")[1]
im_id = glance_helper.create_image(name="vm-image", source_image_file=os.path.join(
GuestImages.DEFAULT["image_dir"], "cirros-0.4.0-x86_64-disk.img"), disk_format="qcow2",
cleanup="module")[1]
vm_id_0 = vm_helper.boot_vm(name="vm-0", flavor=fl_id, nics=net_id_list_0, source="image",
source_id=im_id, cleanup="module")[1]
vm_id_1 = vm_helper.boot_vm(name="vm-1", flavor=fl_id, nics=net_id_list_0, source="image",
source_id=im_id, cleanup="module")[1]
return {
"subnet_id": subnet_id,
"vm_id_0": vm_id_0,
"vm_id_1": vm_id_1,
}
@mark.networking
def test_multiple_float_ips_on_multiple_interfaces(launch_instance):
"""
245-Multiple-Float-IPs-On-Multipe-Interfaces.robot
Args:
launch_instance:
Returns:
"""
subnet_id = launch_instance['subnet_id']
vm_id_0 = launch_instance['vm_id_0']
vm_id_1 = launch_instance['vm_id_1']
network_helper.create_router('router1', cleanup="module")
network_helper.add_router_interface('router1', subnet=subnet_id)
uuid_net = network_helper.get_networks(full_name='external-net0')[0]
uuid_sub = network_helper.get_subnets(full_name='external-subnet0')[0]
network_helper.set_router('router1', external_gateway=uuid_net)
float_ip0 = network_helper.create_floating_ip(external_net=uuid_net,
subnet=uuid_sub, cleanup="module")[1]
float_ip1 = network_helper.create_floating_ip(external_net=uuid_net,
subnet=uuid_sub, cleanup="module")[1]
cli.openstack('server add floating ip', "{} {}".format(vm_id_0, float_ip0))
cli.openstack('server add floating ip', "{} {}".format(vm_id_1, float_ip1))

View File

@ -0,0 +1,45 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Networking/VM Same networks used for 2 virtual devices.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
import os
from pytest import mark
from consts.stx import GuestImages
from keywords import glance_helper, network_helper, nova_helper, vm_helper
from utils import cli
@mark.networking
def test_multiple_ports_same_network():
"""
256-Multiple-Ports-Same-Network.robot
Args:
Returns:
"""
fl_id = nova_helper.get_flavors(name="m1.tiny")[0]
im_id = glance_helper.create_image(name="cirros_image", source_image_file=os.path.join(
GuestImages.DEFAULT["image_dir"], "cirros-0.4.0-x86_64-disk.img"), disk_format="qcow2",
cleanup="module")[1]
net_id = network_helper.get_networks(full_name='private-net0')[0]
net_id_list = [{"net-id": net_id}]
vm_id = vm_helper.boot_vm(flavor=fl_id, nics=net_id_list, source="image",
source_id=im_id, cleanup="module")[1]
vm_id_1 = vm_helper.boot_vm(flavor=fl_id, nics=net_id_list, source="image",
source_id=im_id, cleanup="module")[1]
vm_ip_1 = vm_helper.get_vm_values(vm_id, 'addresses')[0].split('=')[1]
vm_ip_2 = vm_helper.get_vm_values(vm_id_1, 'addresses')[0].split('=')[1]
arg1 = "| grep {} | awk '{{print$2}}'".format(vm_ip_1)
arg2 = "| grep {} | awk '{{print$2}}'".format(vm_ip_2)
assert cli.openstack('port list', arg1)[1] != cli.openstack('port list', arg2)[1]

View File

@ -0,0 +1,36 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Test case to verify a router can enable and diable NAT.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
import json
import re
from pytest import mark
from keywords import network_helper
@mark.networking
def test_nat_on_router():
"""
411-Test-NAT-On-Router.robot
Args:
Returns:
"""
router_id = network_helper.create_router(name="router_test", cleanup="function")[1]
network_helper.set_router(router="router_test", enable_snat=False, external_gateway="external-net0")
values = network_helper.get_router_values(router_id, fields='external_gateway_info')
assert re.search("\"enable_snat\": false", json.dumps(values[0]))
network_helper.set_router(router="router_test", enable_snat=True, external_gateway="external-net0")
values = network_helper.get_router_values(router_id, fields='external_gateway_info')
assert re.search("\"enable_snat\": true", json.dumps(values[0]))

View File

@ -0,0 +1,39 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Not compatible configuration for "External OAM".
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
from pytest import mark
from keywords import system_helper
@mark.networking
def test_not_compatible_configuration_external_oam():
"""
402-Not-Compatible-Configuration-External-OAM.robot
Args:
Returns:
"""
oam_info = system_helper.get_oam_values(fields='oam_c0_ip')
oam_ip = None
for key, value in oam_info.items():
if value is not None:
oam_ip = value
message = system_helper.modify_oam_ips(fail_ok=True, **{'oam_c0_ip': 'abcdafd'})[1]
assert 'Invalid address abcdafd in oam_c0_ip' in message
oam_info_now = system_helper.get_oam_values(fields='oam_c0_ip')
oam_ip_now = None
for key, value in oam_info_now.items():
if value is not None:
oam_ip_now = value
assert oam_ip_now == oam_ip

View File

@ -0,0 +1,82 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Test to check ntp servers can change.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
import time
from pytest import mark
from consts.auth import HostLinuxUser, Tenant
from consts.stx import EventLogID
from keywords import system_helper, host_helper
from utils import cli
from utils.clients import ssh
from utils.clients.ssh import SSHClient, CONTROLLER_PROMPT
from utils.tis_log import LOG
def swact_host(host):
con_ssh = ssh.ControllerClient.get_active_controller()
if host == 'controller-0':
oam_ip = 'oam_c1_ip'
else:
oam_ip = 'oam_c0_ip'
cmd = "oam-show |grep {} | awk '{{print $4}}'".format(oam_ip)
oam_c1_ip = cli.system(cmd)[1]
node_ssh = SSHClient(oam_c1_ip, HostLinuxUser.get_user(),
HostLinuxUser.get_password(),
CONTROLLER_PROMPT)
exitcode, msg = cli.system('host-swact', host, ssh_client=con_ssh,
fail_ok=False, auth_info=Tenant.get('admin_platform'))
assert exitcode == 0, msg
time.sleep(120)
con_ssh.close()
node_ssh.connect(retry=True, retry_timeout=30)
ssh.ControllerClient.set_active_controller(node_ssh)
@mark.networking
def test_online_static_configuration():
"""
237-Online-Static-Configuration.robot
Args:
Returns:
"""
ntp = system_helper.get_ntp_values()
system_helper.modify_ntp(ntp_servers='test.ntp.intel.com', check_first=False, clear_alarm=False)
new_ntp = system_helper.get_ntp_values()
LOG.info("new ntp is {}".format(new_ntp))
assert 'test.ntp.intel.com' in new_ntp
system_helper.modify_ntp(ntp_servers=ntp, check_first=False, clear_alarm=False)
ntp_1 = system_helper.get_ntp_values()
assert ntp == ntp_1
hosts = system_helper.get_controllers()
host_helper.lock_unlock_hosts(hosts[-1])
if not system_helper.is_aio_simplex():
swact_host("controller-0")
host_helper.lock_unlock_hosts("controller-0")
swact_host("controller-1")
if not system_helper.is_aio_system():
computes = system_helper.get_computes()
for compute in computes:
status = system_helper.get_host_values(compute, fields="config_status")[0]
if "Config out-of-date" == status:
host_helper.lock_unlock_hosts(compute)
if system_helper.is_storage_system():
storages = system_helper.get_storage_nodes()
for storage in storages:
status = system_helper.get_host_values(storage, fields="config_status")[0]
if "Config out-of-date" == status:
host_helper.lock_unlock_hosts(storage)
assert EventLogID.CONFIG_OUT_OF_DATE not in \
system_helper.get_alarms(fields=("Alarm ID",), alarm_id=EventLogID.CONFIG_OUT_OF_DATE)

View File

@ -0,0 +1,84 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Verify packet type security rule enforcement based on applied security group to instance.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
import os
from pytest import fixture, mark
from consts.auth import Tenant
from consts.stx import GuestImages
from keywords import network_helper, nova_helper, vm_helper, glance_helper, host_helper
from utils import cli
from utils.clients.ssh import ControllerClient
from utils.tis_log import LOG
@fixture(scope="module")
def launch_instance():
net_id = network_helper.create_network(name='vm-net', cleanup="module")[1]
network_helper.create_subnet(name='vm-subnet', network='vm-net',
subnet_range='10.0.0.0/24', dhcp=True,
ip_version=4, cleanup="module")
net_id_list = [{"net-id": net_id}]
fl_id = nova_helper.create_flavor(name='vm-flavor', vcpus=1, ram=2048, root_disk=4,
is_public=True, add_default_specs=False, cleanup="module")[1]
im_id = glance_helper.create_image(name="vm-image", source_image_file=os.path.join(
GuestImages.DEFAULT["image_dir"], "cirros-0.4.0-x86_64-disk.img"), disk_format="qcow2",
cleanup="module")[1]
vm_id = vm_helper.boot_vm(name="vm-0", flavor=fl_id, nics=net_id_list, source="image",
source_id=im_id, cleanup="module")[1]
return vm_id, net_id
def check_ping_instance(ip, netid):
LOG.info('start to check ping instance')
host_list = host_helper.get_hypervisors()
for host in host_list:
with host_helper.ssh_to_host(host) as node_ssh:
cmd = 'ip netns | grep --color=never {}'.format(netid)
rc, ns = node_ssh.exec_cmd(cmd=cmd)
if ns and netid in ns.split()[0]:
target = host
netns = ns.split()[0]
break
with host_helper.ssh_to_host(target) as node_ssh:
ping_cmd = "ip netns exec {} ping -c 5 {} ".format(netns, ip)
node_ssh.send_sudo(cmd=ping_cmd)
index = node_ssh.expect(['5 received, 0% packet loss'])
assert index == 0, "Ping instance from instance failed"
return target, netns
@mark.networking
def test_packet_type_security_rule_enforcement(launch_instance):
"""
243-Packet-Type-Security-Rule-Enforcement.robot
Args:
launch_instance:
Returns:
"""
con_ssh = ControllerClient.get_active_controller()
network_helper.create_security_group('test_group', project='admin', cleanup='module')
network_helper.create_security_group_rule(group='test_group', project='admin', protocol='icmp',
remote_ip='0.0.0.0/0', cleanup='module')
# Add Security Group To VM
vm_helper.add_security_group(launch_instance[0], 'test_group')
ip = vm_helper.get_vm_values(launch_instance[0], 'addresses')[0].split('=')[1]
check_ping_instance(ip, launch_instance[1])
network_helper.create_security_group_rule(group='test_group', dst_port='22:22', project='admin',
protocol='tcp', remote_ip='0.0.0.0/0',
cleanup='module')
cli.openstack('server remove security group', '{} test_group'.format(launch_instance[0]),
ssh_client=con_ssh, auth_info=Tenant.get('admin'))

View File

@ -0,0 +1,34 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Provider network type must be: flat, vlan or vxlan.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
from pytest import mark
from keywords import kube_helper
from utils import exceptions
@mark.system_inventory
def test_provider_network_type():
"""
253-Provider-Network-Type.robot
Args:
Returns:
"""
pods = kube_helper.get_pods(namespace='openstack')
pod_list = list(filter(lambda i: 'neutron-server' in i, pods))
result = kube_helper.exec_kube_cmd(sub_cmd='exec',
args='{} -it -n openstack cat '
'/etc/neutron/plugins/ml2/ml2_conf.ini'.format(pod_list[0]))[1]
if "vlan,vxlan" not in result and "flat" not in result:
raise exceptions.VMTimeout("Provider network type must be: flat, vlan or vxlan.")

View File

@ -0,0 +1,66 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Test to create QoS Policy using REST api.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
import getpass
import os
import yaml
from pytest import mark
from consts.auth import HostLinuxUser, Tenant
from keywords import common, container_helper, html_helper
from utils import cli
from utils.clients import ssh
from utils.tis_log import LOG
dest_path = {"conf": {"neutron": {"DEFAULT": {"service_plugins": ['router', 'qos']}},
"plugins": {"ml2_conf": {"ml2": {"extension_drivers": ['port_security', 'qos']}},
"openvswitch_agent": {"agent": {"extensions": ['qos']}}}}}
part1 = "Content-Type: application/json"
part2 = "User-Agent: openstacksdk/0.25.0 keystoneauth1/0.0.0 python-requests/2.14.2 CPython/2.7.5"
url = "http://neutron.openstack.svc.cluster.local:80/v2.0/qos/policies"
@mark.networking
def test_qos_policy_creation_api():
"""
329-QoS-Policy-Creation-Api.robot
Args:
Returns:
"""
con_ssh = ssh.ControllerClient.get_active_controller()
token = html_helper.get_user_token()
user_name = getpass.getuser()
yaml_path = os.path.join('/home/{}/'.format(user_name), "neutron-overrides-qos.yaml")
with open(yaml_path, "w", encoding="utf-8") as f:
yaml.dump(dest_path, f)
common.scp_from_test_server_to_active_controller(yaml_path, HostLinuxUser.get_home())
app_dir = HostLinuxUser.get_home()
file_path = os.path.join(app_dir, "neutron-overrides-qos.yaml")
container_helper.update_helm_override(app_name='stx-openstack', chart='neutron',
namespace='openstack', yaml_file=file_path)
container_helper.apply_app(app_name='stx-openstack', applied_timeout=1200,
check_interval=30)
part3 = "X-Auth-Token: {}".format(token)
part4 = '{"policy": {"name": "bw-limiter"}}'
cmd = "curl -g -i -X POST {} -H '{}' -H '{}' -H '{}' -d '{}'".format(url, part1, part2, part3, part4)
LOG.info("cmd is {}".format(cmd))
code, out = con_ssh.exec_cmd(cmd)
assert code == 0
cli.system('helm-override-delete', "stx-openstack neutron openstack",
ssh_client=con_ssh, auth_info=Tenant.get('admin_platform'))
container_helper.apply_app(app_name='stx-openstack', applied_timeout=1200,
check_interval=30)

View File

@ -0,0 +1,42 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Test to show network details using REST api on given network.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
from pytest import mark
from keywords import network_helper, html_helper
from utils.clients import ssh
from utils.tis_log import LOG
part1 = "Accept: "
part2 = "User-Agent: opencksdk/0.17.2 keystoneauth1/3.10.0 python-requests/2.14.2 CPython/2.7.5"
url = "http://neutron.openstack.svc.cluster.local:80/v2.0/networks/"
@mark.networking
def test_show_data_network_api():
"""
238-Show-Data-Network-Api.robot
Args:
Returns:
"""
con_ssh = ssh.ControllerClient.get_active_controller()
network_id = network_helper.get_networks()[0]
token = html_helper.get_user_token()
part3 = "X-Auth-Token: {}".format(token)
cmd = "curl -g -i -X GET {}{} -H '{}' -H '{}' -H '{}'".format(url, network_id, part1, part2, part3)
LOG.info("cmd is {}".format(cmd))
code, out = con_ssh.exec_cmd(cmd)
assert code == 0

View File

@ -0,0 +1,44 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Test to show data network range using REST api for given network.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
from pytest import mark
from consts.auth import Tenant
from keywords import system_helper, html_helper
from utils import table_parser, cli
from utils.clients import ssh
part1 = "Accept: application/json"
part2 = "User-Agent: openstacksdk/0.25.0 keystoneauth1/0.0.0 python-requests/2.21.0 CPython/2.7.5"
url = "http://neutron-server.openstack.svc.cluster.local:9696/v2.0/network_segment_ranges/"
@mark.networking
def test_show_data_network_range_api():
"""
239-Show-Data-Network-Range-Api.robot
Args:
Returns:
"""
con_ssh = ssh.ControllerClient.get_active_controller()
data_network = system_helper.get_data_networks(**{'network_type': 'vlan'})[0]
table_ = table_parser.table(
cli.openstack('network segment range list', ssh_client=con_ssh, auth_info=Tenant.get('admin'))[1])
val = table_parser.get_values(table_, "ID", **{'Physical Network': data_network})[0]
token = html_helper.get_user_token()
part3 = "X-Auth-Token: {}".format(token)
cmd = "curl -g -i -X GET {}{} -H '{}' -H '{}' -H '{}'".format(url, val, part1, part2, part3)
code, out = con_ssh.exec_cmd(cmd)
assert code == 0

View File

@ -0,0 +1,91 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# This test case creates a trunk network with subport
# and assign the intance to the port of trunk then execute
# instance lifecycle.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
import getpass
import os
import yaml
from pytest import mark
from consts.auth import Tenant, HostLinuxUser
from consts.stx import GuestImages
from keywords import network_helper, common, container_helper, nova_helper, glance_helper, vm_helper
from utils import cli, table_parser
from utils.clients.ssh import ControllerClient
desired_caps = {"conf": {"neutron": {"DEFAULT": {"service_plugins": ['router', 'trunk']}}}}
# Creating Instance
def launch_instance(net_id):
fl_id = nova_helper.get_flavors(name="m1.tiny")[0]
im_id = glance_helper.create_image(name="cirros_image", source_image_file=os.path.join(
GuestImages.DEFAULT["image_dir"], "cirros-0.4.0-x86_64-disk.img"), disk_format="qcow2",
cleanup="module")[1]
net_id_list = [{"net-id": net_id}]
vm_id = vm_helper.boot_vm_openstack(name="portinstance", flavor=fl_id, nics=net_id_list,
source="image", source_id=im_id, cleanup="module",
port="parentport")[1]
return vm_id
@mark.networking
def test_trunk_with_vm_lifecycle(no_simplex):
"""
405-Trunk-With-VM-Lifecycle.robot
Args:
no_simplex:
Returns:
"""
con_ssh = ControllerClient.get_active_controller()
net_id = network_helper.create_network(network_type='vlan',
name='net-a', cleanup="function")[1]
network_helper.create_subnet(net_id, subnet_range='192.168.3.0/24',
name='subnet-a', cleanup="function")
user_name = getpass.getuser()
yaml_path = os.path.join('/home/{}/'.format(user_name), "neutron-overrides-trunk.yaml")
with open(yaml_path, "w", encoding="utf-8") as f:
yaml.dump(desired_caps, f)
common.scp_from_test_server_to_active_controller(yaml_path, HostLinuxUser.get_home())
app_dir = HostLinuxUser.get_home()
file_path = os.path.join(app_dir, "neutron-overrides-trunk.yaml")
container_helper.update_helm_override(app_name='stx-openstack', chart='neutron',
namespace='openstack', yaml_file=file_path)
container_helper.apply_app(app_name='stx-openstack', applied_timeout=1200,
check_interval=30)
network_helper.create_port(net_id, 'parentport', cleanup="function")
network_helper.create_port(net_id, 'subport', cleanup="function")
args = '--parent-port parentport --subport port=subport,segmentation-type=vlan,' \
'segmentation-id=10 trunkseg'
code, output = cli.openstack('network trunk create', args,
ssh_client=con_ssh, fail_ok=False,
auth_info=Tenant.get('admin'))
table_ = table_parser.table(output)
trunk_id = table_parser.get_value_two_col_table(table_, 'id')
vm_id = launch_instance(net_id)
vm_helper.stop_vms(vm_id)
vm_helper.start_vms(vm_id)
vm_helper.pause_vm(vm_id)
vm_helper.unpause_vm(vm_id)
vm_helper.suspend_vm(vm_id)
vm_helper.resume_vm(vm_id)
vm_helper.cold_migrate_vm(vm_id)
network_helper.unset_trunk(trunk_id, sub_ports="subport")
network_helper.delete_trunks(trunk_id)
cli.system('helm-override-delete', "stx-openstack neutron openstack",
ssh_client=con_ssh, auth_info=Tenant.get('admin_platform'))
container_helper.apply_app(app_name='stx-openstack', applied_timeout=1200,
check_interval=30)

View File

@ -0,0 +1,64 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Test to verify compatible NTP configuration.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
import time
from pytest import mark
from consts.auth import HostLinuxUser, Tenant
from keywords import system_helper, host_helper
from utils import cli
from utils.clients import ssh
from utils.clients.ssh import SSHClient, CONTROLLER_PROMPT
def swact_host(host):
con_ssh = ssh.ControllerClient.get_active_controller()
if host == 'controller-0':
oam_ip = 'oam_c1_ip'
else:
oam_ip = 'oam_c0_ip'
cmd = "oam-show |grep {} | awk '{{print $4}}'".format(oam_ip)
oam_c1_ip = cli.system(cmd)[1]
node_ssh = SSHClient(oam_c1_ip, HostLinuxUser.get_user(),
HostLinuxUser.get_password(),
CONTROLLER_PROMPT)
exitcode, msg = cli.system('host-swact', host, ssh_client=con_ssh,
fail_ok=False, auth_info=Tenant.get('admin_platform'))
assert exitcode == 0, msg
time.sleep(120)
con_ssh.close()
node_ssh.connect(retry=True, retry_timeout=30)
ssh.ControllerClient.set_active_controller(node_ssh)
@mark.networking
def test_verify_compatible_ntp_configuration():
"""
401-Test-Verify-Compatible-NTP-Configuration.robot
Args:
Returns:
"""
ntp = system_helper.get_ntp_values()
system_helper.modify_ntp(ntp_servers='10.22.1.1,10.22.1.2',
check_first=False, clear_alarm=False)
new_ntp = system_helper.get_ntp_values()
assert '10.22.1.1,10.22.1.2' in new_ntp
system_helper.modify_ntp(ntp_servers=ntp, check_first=False, clear_alarm=False)
hosts = system_helper.get_controllers()
host_helper.lock_unlock_hosts(hosts[-1])
if not system_helper.is_aio_simplex():
swact_host("controller-0")
host_helper.lock_unlock_hosts("controller-0")
swact_host("controller-1")

View File

@ -0,0 +1,120 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Test to verify DVR can be deleted.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
import os
from pytest import fixture, mark
from consts.stx import GuestImages
from keywords import network_helper, nova_helper, glance_helper, vm_helper, host_helper
from utils import cli
from utils.tis_log import LOG
@fixture(scope="module")
def launch_instance():
net_id_0 = network_helper.create_network(name='vm-0-net', cleanup="module")[1]
subnet_id_0 = network_helper.create_subnet(name='vm-0-subnet', network='vm-0-net',
subnet_range='10.0.0.0/24', dhcp=True,
ip_version=4, cleanup="module")[1]
net_id_1 = network_helper.create_network(name='vm-1-net', cleanup="module")[1]
subnet_id_1 = network_helper.create_subnet(name='vm-1-subnet', network='vm-1-net',
subnet_range='10.0.1.0/24', dhcp=True,
ip_version=4, cleanup="module")[1]
net_id_list_0 = [{"net-id": net_id_0}]
net_id_list_1 = [{"net-id": net_id_1}]
fl_id_0 = nova_helper.create_flavor(name='vm-0-flavor', vcpus=1, ram=2048, root_disk=4,
is_public=True, add_default_specs=False, cleanup="module")[1]
im_id_0 = glance_helper.create_image(name="vm-0-image", source_image_file=os.path.join(
GuestImages.DEFAULT["image_dir"], "cirros-0.4.0-x86_64-disk.img"), disk_format="qcow2",
cleanup="module")[1]
vm_id_0 = vm_helper.boot_vm(name="vm-0", flavor=fl_id_0, nics=net_id_list_0, source="image",
source_id=im_id_0, cleanup="module")[1]
fl_id_1 = nova_helper.create_flavor(name='vm-0-flavor', vcpus=1, ram=2048, root_disk=4,
is_public=True, add_default_specs=False, cleanup="module")[1]
im_id_1 = glance_helper.create_image(name="vm-0-image", source_image_file=os.path.join(
GuestImages.DEFAULT["image_dir"], "cirros-0.4.0-x86_64-disk.img"), disk_format="qcow2",
cleanup="module")[1]
vm_id_1 = vm_helper.boot_vm(name="vm-1", flavor=fl_id_1, nics=net_id_list_1, source="image",
source_id=im_id_1, cleanup="module")[1]
network_helper.create_security_group('group', project='admin', cleanup='module')
network_helper.create_security_group_rule(group='group', dst_port='22:22', project='admin',
protocol='tcp', remote_ip='0.0.0.0/0',
cleanup='module')
network_helper.create_security_group_rule(group='group', project='admin', protocol='icmp',
remote_ip='0.0.0.0/0', cleanup='module')
# Add Security Group To VM
vm_helper.add_security_group(vm_id_0, 'group')
vm_helper.add_security_group(vm_id_1, 'group')
return {
"subnet_id_0": subnet_id_0,
"subnet_id_1": subnet_id_1,
"vm_id_0": vm_id_0,
"vm_id_1": vm_id_1,
"net_id_0": net_id_0,
"net_id_1": net_id_1
}
def check_ping_instance(ip, netid):
LOG.info('start to check ping instance')
host_list = host_helper.get_hypervisors()
for host in host_list:
with host_helper.ssh_to_host(host) as node_ssh:
cmd = 'ip netns | grep --color=never {}'.format(netid)
rc, ns = node_ssh.exec_cmd(cmd=cmd)
if ns and netid in ns.split()[0]:
target = host
netns = ns.split()[0]
break
with host_helper.ssh_to_host(target) as node_ssh:
ping_cmd = "ip netns exec {} ping -c 5 {} " \
.format(netns, ip)
node_ssh.send_sudo(cmd=ping_cmd)
index = node_ssh.expect(['5 received, 0% packet loss'])
assert index == 0, "Ping instance from instance failed"
@mark.networking
def test_verify_dvr_router_deletion(launch_instance):
"""
255-Verify-DVR-Router-Deletion.robot
Args:
launch_instance:
Returns:
"""
subnet_id_0 = launch_instance['subnet_id_0']
subnet_id_1 = launch_instance['subnet_id_1']
vm_id_0 = launch_instance['vm_id_0']
vm_id_1 = launch_instance['vm_id_1']
net_id_0 = launch_instance['net_id_0']
net_id_1 = launch_instance['net_id_1']
network_helper.create_router('router1', cleanup="module")
network_helper.add_router_interface('router1', subnet=subnet_id_0)
network_helper.add_router_interface('router1', subnet=subnet_id_1)
net_id = network_helper.get_networks(full_name='external-net0')[0]
subnet_id = network_helper.get_subnets(full_name='external-subnet0')[0]
network_helper.set_router('router1', external_gateway=net_id)
float_ip0 = network_helper.create_floating_ip(external_net=net_id,
subnet=subnet_id, cleanup="module")[1]
float_ip1 = network_helper.create_floating_ip(external_net=net_id,
subnet=subnet_id, cleanup="module")[1]
cli.openstack('server add floating ip', "{} {}".format(vm_id_0, float_ip0))
cli.openstack('server add floating ip', "{} {}".format(vm_id_1, float_ip1))
check_ping_instance(float_ip0, net_id_0)
check_ping_instance(float_ip1, net_id_1)
network_helper.unset_router('router1', external_gateway=net_id)
network_helper.remove_router_interface('router1', subnet=subnet_id_0)
network_helper.remove_router_interface('router1', subnet=subnet_id_1)

View File

@ -0,0 +1,84 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Verify ethernet OAM inteface is updated successfully on controller.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
import time
from pytest import mark
from consts.auth import Tenant, HostLinuxUser
from keywords import host_helper, container_helper
from utils import cli, table_parser
from utils.clients import ssh
from utils.clients.ssh import ControllerClient, SSHClient, CONTROLLER_PROMPT
def swact_host(host):
con_ssh = ControllerClient.get_active_controller()
if host == 'controller-0':
oam_ip = 'oam_c1_ip'
else:
oam_ip = 'oam_c0_ip'
cmd = "oam-show |grep {} | awk '{{print $4}}'".format(oam_ip)
oam_c1_ip = cli.system(cmd)[1]
node_ssh = SSHClient(oam_c1_ip, HostLinuxUser.get_user(),
HostLinuxUser.get_password(),
CONTROLLER_PROMPT)
exitcode, msg = cli.system('host-swact', host, ssh_client=con_ssh,
fail_ok=False, auth_info=Tenant.get('admin_platform'))
assert exitcode == 0, msg
time.sleep(120)
con_ssh.close()
node_ssh.connect(retry=True, retry_timeout=30)
ssh.ControllerClient.set_active_controller(node_ssh)
def verify_oam_interface(mtu, port_name, if_name):
host_helper.lock_host('controller-1')
host_helper.modify_host_interface('controller-1', port_name, mtu=mtu)
host_helper.unlock_host('controller-1')
swact_host('controller-0')
host_helper.lock_host('controller-0')
host_helper.modify_host_interface('controller-0', if_name, mtu=mtu)
host_helper.unlock_host('controller-0')
swact_host('controller-1')
@mark.networking
def test_verify_ethernet_oam_interface_is_updated(no_aio_system):
"""
251-Verify-Ethernet-OAM-Interface-Is-Updated.robot
Args:
no_aio_system:
Returns:
"""
con_ssh = ControllerClient.get_active_controller()
auth_info = Tenant.get('admin_platform')
table_ = table_parser.table(cli.system("interface-network-list", "controller-0",
ssh_client=con_ssh,
auth_info=auth_info)[1])
# Get OAM interface
if_name = table_parser.get_values(table_, "ifname", **{"network_name": "oam"})[0]
iftype = host_helper.get_host_interface_values('controller-0', fields='iftype', interface=if_name)[0]
assert iftype == 'ethernet'
port_name = cli.system("host-if-list", "controller-1 | grep {} | awk '{{print$4}}'".format(if_name),
ssh_client=con_ssh, auth_info=auth_info)[1]
old_mtu_ctrl = host_helper.get_host_interface_values('controller-1',
fields='imtu', interface=port_name)[0]
verify_oam_interface('9000', port_name, if_name)
verify_oam_interface(old_mtu_ctrl, port_name, if_name)
application_status = container_helper.get_apps(application="stx-openstack")[0]
if application_status == 'apply-failed':
container_helper.apply_app(app_name="stx-openstack", applied_timeout=3600,
check_interval=30, wait_for_alarm_gone=False)

View File

@ -0,0 +1,50 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# In Horizon Topology Navigation and details check.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
import time
from pytest import mark
from selenium.webdriver.common import by
from testfixtures.horizon import *
from keywords import system_helper
from utils.horizon.pages.admin.platform import providernetworkstopology
from utils.tis_log import LOG
@mark.networking
def test_verify_horizon_network_topology_config_details(no_aio_system, tenant_home_pg_container_starlingx):
"""
376-Verify-Horizon-Network-Topology-Config-Details.robot
Args:
no_aio_system:
tenant_home_pg_container_starlingx:
Returns:
"""
LOG.info("Go to Admin Data Network Topology")
providernetworktopology_pg = providernetworkstopology.ProviderNetworkTopologyPage(
tenant_home_pg_container_starlingx.driver, tenant_home_pg_container_starlingx.port)
providernetworktopology_pg.go_to_target_page()
time.sleep(10)
for data_network in system_helper.get_data_networks():
providernetworktopology_pg.driver.find_element_by_id('%s%s' % ('net-', data_network)).click()
providernetworktopology_pg._get_element(by.By.LINK_TEXT, 'Related Alarms').click()
assert not providernetworktopology_pg.alarm_table.rows
for compute in system_helper.get_computes():
providernetworktopology_pg.driver.find_element_by_id('%s%s' % ('host-', compute)).click()
providernetworktopology_pg._get_element(by.By.LINK_TEXT, 'LLDP').click()
providernetworktopology_pg._get_element(by.By.LINK_TEXT, 'Related Alarms').click()
assert not providernetworktopology_pg.alarm_table.rows

View File

@ -0,0 +1,44 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# This test is to verify the vSwitch settings from the cli,
# This test pass if any vswitch configuration was found.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
from pytest import mark
from consts.auth import Tenant
from keywords import system_helper
from utils import cli, table_parser
from utils.clients.ssh import ControllerClient
@mark.networking
def test_verify_vswitch_settings_cli():
"""
233-Verify-VSwitch-Settings-Cli.robot
Args:
Returns:
"""
if system_helper.is_aio_simplex():
host = "controller-0"
else:
host = system_helper.get_computes()[0]
con_ssh = ControllerClient.get_active_controller()
table_ = table_parser.table(cli.system('host-cpu-list', host)[1])
functions = table_parser.get_values(table_=table_,
target_header='assigned_function')
assert "vSwitch" in functions, "vSwitch not configured"
uuids = table_parser.get_values(table_=table_, target_header='uuid',
**{'assigned_function': 'vSwitch'})
for uuid in uuids:
cli.system("host-cpu-show", "1 {}".format(uuid), ssh_client=con_ssh,
auth_info=Tenant.get('admin_platform'))

View File

@ -0,0 +1,62 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Test case for vxlan data network segmentation range should not be overlap.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
from pytest import mark
from keywords import system_helper, host_helper, network_helper
from utils import cli, table_parser
addr_list = ['192.168.100.30 24', '192.168.100.40 24']
@mark.networking
def test_verify_vxlan_data_network_segment_range_overlap(no_aio_system):
"""
434-Verify-Vxlan-Data-Network-Segment-Range-Overlap.robot
Args:
no_aio_system:
Returns:
"""
computes = system_helper.get_computes()
for compute in computes:
host_helper.lock_host(compute)
system_helper.create_data_network("physnet3", "vxlan", multicast_group='224.0.0.1',
ttl='255', port_num='4789', cleanup='function')
for index, compute in enumerate(computes):
uuid = host_helper.get_host_interfaces(compute, field='uuid', **{'name': 'data0'})[0]
cli.system('host-if-modify', "-m 1600 -n data0 -c data {} {} --ipv4-mode=static".
format(compute, uuid))
cli.system('interface-datanetwork-assign', "{} {} physnet3".format(compute, uuid))
cli.system('host-addr-add', "{} {} {}".format(compute, uuid, addr_list[index]))
host_helper.unlock_host(compute)
if index == 1:
break
seg0_id = network_helper.create_segmentation_range('vxlan-seg0', minimum='300',
maximum='350', network_type='vxlan')
seg1_id = network_helper.create_segmentation_range('vxlan-seg1', minimum='399',
maximum='450', network_type='vxlan')
for compute in computes:
host_helper.lock_host(compute)
addr_uuid = host_helper.get_host_addresses(compute, field='uuid', ifname='data0')[0]
cli.system('host-addr-delete', addr_uuid)
uuid = host_helper.get_host_interfaces(compute, field='uuid', **{'name': 'data0'})[0]
cli.system('host-if-modify', "-m 1500 -n data0 -c data {} {} --ipv4-mode=disabled".
format(compute, uuid))
table_ = table_parser.table(
cli.system('interface-datanetwork-list', "{}".format(compute))[1])
val = table_parser.get_values(table_, "uuid", **{'datanetwork_name': 'physnet3'})[0]
cli.system('interface-datanetwork-remove', "{}".format(val))
host_helper.unlock_host(compute)
cli.openstack('network segment range delete', seg0_id)
cli.openstack('network segment range delete', seg1_id)

View File

@ -0,0 +1,55 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Test case for vxlan data interface should not allow duplicate ip address.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
from pytest import mark
from keywords import system_helper, host_helper
from utils import cli
addr_list = ['192.168.100.30 24', '192.168.100.40 24']
@mark.networking
def test_vxlan_data_interface_duplicate_address_not_allowed(no_aio_system):
"""
433-Vxlan-Data-Interface-Duplicate-Address-Not-Allowed.robot
Args:
no_aio_system:
Returns:
"""
computes = system_helper.get_computes()
for compute in computes:
host_helper.lock_host(compute)
data_network_uuid = system_helper.create_data_network("physnet3", "vxlan", multicast_group='224.0.0.1',
ttl='255', port_num='4789')[1]
for index, compute in enumerate(computes):
uuid = host_helper.get_host_interfaces(compute, field='uuid', **{'name': 'data0'})[0]
cli.system('host-if-modify', "-n data0 -c data {} {} --ipv4-mode=static".
format(compute, uuid))
cli.system('host-addr-add', "{} {} {}".format(compute, uuid, addr_list[index]))
code = cli.system('host-addr-add', "{} {} {}".
format(compute, uuid, addr_list[index]), fail_ok=True)[0]
assert code != 0, "Again add host same ip address to compute-0 data0, should through error"
host_helper.unlock_host(compute)
if index == 1:
break
for compute in computes:
host_helper.lock_host(compute)
addr_uuid = host_helper.get_host_addresses(compute, field='uuid', ifname='data0')[0]
uuid = host_helper.get_host_interfaces(compute, field='uuid', **{'name': 'data0'})[0]
cli.system('host-addr-delete', addr_uuid)
cli.system('host-if-modify', "-n data0 -c data {} {} --ipv4-mode=disabled".
format(compute, uuid))
host_helper.unlock_host(compute)
system_helper.delete_data_network(data_network_uuid)

View File

@ -0,0 +1,38 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Test case for vxlan data network must have a valid port number.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
from pytest import mark
from keywords import system_helper, host_helper
@mark.networking
def test_vxlan_data_network_valid_port_creation(no_aio_system):
"""
419-Vxlan-Data-Network-Valid-Port-Creation.robot
Args:
no_aio_system:
Returns:
"""
computes = system_helper.get_computes()
for compute in computes:
host_helper.lock_host(compute)
uuid = system_helper.create_data_network("physnet3", "vxlan", multicast_group="224.0.0.1",
ttl="255", port_num="4789")[1]
system_helper.delete_data_network(uuid)
uuid = system_helper.create_data_network("physnet3", "vxlan", multicast_group="224.0.0.1",
ttl="255", port_num="8472")[1]
system_helper.delete_data_network(uuid)
for compute in computes:
host_helper.unlock_host(compute)

View File

@ -0,0 +1,35 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Test case for vxlan data network with valid multicast address.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
from pytest import mark
from keywords import system_helper
@mark.networking
def test_vxlan_data_network_with_valid_multicast_address():
"""
432-Vxlan-Data-Network-With-Valid-Multicast-Address.robot
Args:
Returns:
"""
code = system_helper.create_data_network("physnet3", "vxlan", multicast_group='172.31.255.255',
ttl='255', port_num='4789', fail_ok=True)[0]
assert code != 0, '172.31.255.255 is not a valid multicast ip address'
code = system_helper.create_data_network("physnet3", "vxlan", multicast_group='0:0:0:0:0:0:0:0',
ttl='255', port_num='4789', fail_ok=True)[0]
assert code != 0, '0:0:0:0:0:0:0:0 is not a valid multicast ip address'
data_network_uuid = system_helper.create_data_network("physnet3", "vxlan", multicast_group='224.0.0.1',
ttl='255', port_num='4789')[1]
system_helper.delete_data_network(data_network_uuid)

View File

@ -0,0 +1,45 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Test case for vxlan data network creation and verify
# data interface address mode must be set to static
# before applying any ip address.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
from keywords import system_helper, host_helper
from utils import cli
def test_vxlan_datainterface_address_mode_set_to_static(no_aio_system):
"""
421-Vxlan-DataInterface-Address-Mode-Set-To-Static.robot
Args:
no_aio_system:
Returns:
"""
host = system_helper.get_computes()[0]
host_helper.lock_host(host)
data_network_uuid = system_helper.create_data_network("physnet3", "vxlan", multicast_group='224.0.0.1',
ttl='255', port_num='4789')[1]
uuid = host_helper.get_host_interfaces(host, field='uuid', **{'name': 'data0'})[0]
code = cli.system('host-addr-add', "{} {} 192.168.100.60 24".format(host, uuid), fail_ok=True)[0]
assert code != 0, "Non static mode, can't add host address"
cli.system('host-if-modify', "-n data0 -c data {} {} --ipv4-mode=static".
format(host, uuid))
cli.system('host-addr-add', "{} {} 192.168.100.60 24".format(host, uuid))
host_helper.unlock_host(host)
host_helper.lock_host(host)
addr_uuid = host_helper.get_host_addresses(host, field='uuid', ifname='data0')[0]
cli.system('host-addr-delete', addr_uuid)
cli.system('host-if-modify', "-n data0 -c data {} {} --ipv4-mode=disabled".
format(host, uuid))
system_helper.delete_data_network(data_network_uuid)
host_helper.unlock_host(host)

View File

@ -0,0 +1,60 @@
###
#
# Copyright (c) 2021 Intel Corporation
#
# SPDX-License-Identifier: Apache-2.0
#
# Test case for interface to be modifiable, when host is locked.
# And create vxlan data network to modify data0 interface
# address ipv4/ipv6 mode change(static, disabled). if setting static
# should be set ip address.
#
# Author(s): Yong.Fu <yongx.fu@intel.com>
#
###
from pytest import mark
from keywords import system_helper, host_helper
from utils import cli
addr_list = ['192.168.100.30 24', '192.168.100.40 24']
@mark.networking
def test_vxlan_interface_address_mode_modifiable_when_host_locked(no_aio_system):
"""
431-Vxlan-Interface-Address-Mode-Modifiable-When-Host-Locked.robot
Args:
no_aio_system:
Returns:
"""
computes = system_helper.get_computes()
for compute in computes:
uuid = host_helper.get_host_interfaces(compute, field='uuid', **{'name': 'data0'})[0]
code = cli.system('host-if-modify', "-n data0 -c data {} {} --ipv4-mode=static".
format(compute, uuid), fail_ok=True)[0]
assert code != 0, "without host lock to modify data0 interface, will through error"
for compute in computes:
host_helper.lock_host(compute)
data_network_uuid = system_helper.create_data_network("physnet3", "vxlan", multicast_group='224.0.0.1',
ttl='255', port_num='4789')[1]
for index, compute in enumerate(computes):
uuid = host_helper.get_host_interfaces(compute, field='uuid', **{'name': 'data0'})[0]
cli.system('host-if-modify', "-n data0 -c data {} {} --ipv4-mode=static".
format(compute, uuid))
cli.system('host-addr-add', "{} {} {}".format(compute, uuid, addr_list[index]))
host_helper.unlock_host(compute)
if index == 1:
break
for compute in computes:
host_helper.lock_host(compute)
addr_uuid = host_helper.get_host_addresses(compute, field='uuid', ifname='data0')[0]
uuid = host_helper.get_host_interfaces(compute, field='uuid', **{'name': 'data0'})[0]
cli.system('host-addr-delete', addr_uuid)
cli.system('host-if-modify', "-n data0 -c data {} {} --ipv4-mode=disabled".
format(compute, uuid))
host_helper.unlock_host(compute)
system_helper.delete_data_network(data_network_uuid)

View File

@ -47,8 +47,13 @@ def tenant_home_pg_container(driver, request):
auth_info=Tenant.get_primary())
def __login_base(request, driver, auth_info, port=None):
@fixture(scope='function')
def tenant_home_pg_container_starlingx(driver, request):
return __login_base(request=request, driver=driver,
auth_info=Tenant.get_primary(), port="8080")
def __login_base(request, driver, auth_info, port=None):
horizon.test_result = False
if not auth_info:
auth_info = Tenant.get_primary()
@ -61,7 +66,7 @@ def __login_base(request, driver, auth_info, port=None):
gmttime = datetime.datetime.utcnow().strftime("%Y%m%d%H%M%S")
video_path = ProjVar.get_var('LOG_DIR') + '/horizon/' + \
str(gmttime) + '.mp4'
str(gmttime) + '.mp4'
recorder = video_recorder.VideoRecorder(1920, 1080, os.environ['DISPLAY'],
video_path)
recorder.start()

View File

@ -0,0 +1,84 @@
from utils.horizon.pages import basepage
from utils.horizon.regions import tables, forms
class DatanetworksTable(tables.TableRegion):
name = "data_networks"
DATA_NETWORK_FORM_FIELDS = ("name", "mtu", "network_type")
@tables.bind_table_action('create')
def create_datanetwork(self, create_button):
create_button.click()
self.wait_till_spinner_disappears()
return forms.FormRegion(self.driver, field_mappings=self.DATA_NETWORK_FORM_FIELDS)
@tables.bind_table_action('delete')
def delete_datanetwork(self, delete_button):
delete_button.click()
return forms.BaseFormRegion(self.driver)
@tables.bind_row_action("delete")
def delete_datanetwork_by_row(self, delete_button, row):
delete_button.click()
return forms.BaseFormRegion(self.driver)
@tables.bind_row_action("update")
def edit_datanetwork(self, edit_button, row):
edit_button.click()
self.wait_till_spinner_disappears()
return forms.FormRegion(self.driver, field_mappings=self.DATA_NETWORK_FORM_FIELDS)
@tables.bind_row_anchor_column('Network Name')
def go_to_datanetwork_detail_page(self, row_link, row):
row_link.click()
class DatanetworksPage(basepage.BasePage):
PARTIAL_URL = 'admin/datanets/'
NETWORKS_TABLE_NAME_COLUMN = 'Network Name'
def _get_row_with_datanetwork_name(self, name):
return self.datanetworks_table.get_row(
self.NETWORKS_TABLE_NAME_COLUMN, name)
@property
def datanetworks_table(self):
return DatanetworksTable(self.driver)
def create_datanetwork(self, network_name, mtu=None, network_type=None):
create_datanetwork_form = self.datanetworks_table.create_datanetwork()
create_datanetwork_form.name.text = network_name
if mtu is not None:
create_datanetwork_form.mtu.text = mtu
if network_type is not None:
create_datanetwork_form.network_type.text = network_type
create_datanetwork_form.submit()
def edit_datanetwork(self, name, mtu=None, network_type=None):
row = self._get_row_with_datanetwork_name(name)
create_datanetwork_form = self.datanetworks_table.edit_datanetwork(row)
create_datanetwork_form.name.text = name
if mtu is not None:
create_datanetwork_form.mtu.text = mtu
if network_type is not None:
create_datanetwork_form.network_type.text = network_type
create_datanetwork_form.submit()
def delete_datanetwork(self, name):
row = self._get_row_with_datanetwork_name(name)
row.mark()
confirm_delete_networks_form = self.datanetworks_table.delete_datanetwork()
confirm_delete_networks_form.submit()
def delete_datanetwork_by_row(self, name):
row = self._get_row_with_datanetwork_name(name)
confirm_delete_networks_form = self.datanetworks_table.delete_datanetwork_by_row(row)
confirm_delete_networks_form.submit()
def get_datanetwork_info(self, network_name, header):
row = self._get_row_with_datanetwork_name(network_name)
return row.cells[header].text
def go_to_datanetwork_detail_page(self, datanetwork_name):
row = self._get_row_with_datanetwork_name(datanetwork_name)
self.datanetworks_table.go_to_datanetwork_detail_page(row)

View File

@ -48,6 +48,10 @@ class ContainerRegion(baseregion.BaseRegion):
# return self._get_element(*self._detail_view_locator)
class AlarmsTable(tables.TableRegion):
name = "alarms"
class ProviderNetworkList(ContainerRegion):
name = "network_list"
@ -74,6 +78,10 @@ class ProviderNetworkTopologyPage(basepage.BasePage):
def providernet_list(self):
return ProviderNetworkList(self.driver)
@property
def alarm_table(self):
return AlarmsTable(self.driver)
def providernet_detail(self):
return ProviderNetworkDetail(self.driver).get_content()