# Copyright 2024 Red Hat, Inc. # All Rights Reserved. # # 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. import netaddr from neutron_tempest_plugin import config from tempest.common.utils.linux import remote_client from tempest import exceptions from tempest.lib import decorators from whitebox_neutron_tempest_plugin.tests.scenario import base CONF = config.CONF WB_CONF = CONF.whitebox_neutron_plugin_options class NetworkSriovBaseTest(base.ProviderBaseTest): @classmethod def setup_clients(cls): super(NetworkSriovBaseTest, cls).setup_clients() cls.secgroup_client = cls.manager.network_client @classmethod def skip_checks(cls): super(NetworkSriovBaseTest, cls).skip_checks() if not (CONF.neutron_plugin_options.advanced_image_ref or CONF.neutron_plugin_options.default_image_is_advanced): raise cls.skipException( 'Advanced image is required to run these tests.') @classmethod def resource_setup(cls): super(NetworkSriovBaseTest, cls).resource_setup() if not cls.has_sriov_support: raise cls.skipException("Skipped because SRIOV is not supported") cls.router = cls.create_router_by_client() # set image and flavor in case default image is not advanced if not CONF.neutron_plugin_options.default_image_is_advanced: cls.flavor_ref = \ CONF.neutron_plugin_options.advanced_image_flavor_ref cls.image_ref = CONF.neutron_plugin_options.advanced_image_ref cls.username = CONF.neutron_plugin_options.advanced_image_ssh_user def _create_network_port(self, port_type, cidr, reused_net=None): return super(NetworkSriovBaseTest, self)._create_network_port( port_type, reuse_port=False, use_provider_net=False, reused_tenant_net=reused_net, cidr=cidr) def _test_create_instance_with_network_port(self, port_type, cidr=None): security_groups, networks, user_data, config_drive = \ self._create_network_port(port_type, cidr) self.create_router_interface(self.router['id'], self.subnet['id']) return self._create_server_with_fip( security_groups, [networks], user_data=user_data, config_drive=config_drive, vm_name='vm_' + port_type) def _test_create_instance_with_two_tenant_networks_two_ports( self, port_types, cidrs=[None, None], reused_nets=[None, None]): _, networks, user_data, config_drive = \ self._create_network_port(port_types[0], cidrs[0], reused_nets[0]) self.create_router_interface(self.router['id'], self.subnet['id']) # create secondary network and port _, networks_alt, user_data_alt, config_drive_alt = \ self._create_network_port(port_types[1], cidrs[1], reused_nets[1]) return self._create_server_with_several_networks( [networks, networks_alt], user_data=user_data, config_drive=config_drive, vm_name="vm-" + self._testMethodName) def _create_server_with_fip( self, security_groups, networks, user_data, config_drive, vm_name): server = self._create_server( security_groups=security_groups, networks=networks, user_data=user_data, config_drive=config_drive, name=vm_name) port = self.client.list_ports( network_id=self.network['id'], device_id=server['id'])['ports'][0] self.fip = self.create_and_associate_floatingip(port['id']) return server def _create_server_with_several_networks( self, networks, user_data, config_drive, vm_name): server = self._create_server( networks=networks, user_data=user_data, config_drive=config_drive, name=vm_name) port0_id = networks[0]['port'] self.fips.append(self.create_and_associate_floatingip(port0_id)) return server def remove_server_and_network(self, server): self.manager.servers_client.delete_server(server['id']) self.client.delete_port(self.port_id) self.client.delete_network(self.network['id']) def _test_create_instances_with_ports(self, port1, port2): cidr = netaddr.IPNetwork('10.100.0.0/16') server1 = self._test_create_instance_with_network_port( port1, cidr=cidr) server_ssh_client = remote_client.RemoteClient( self.floating_ips[0]['floating_ip_address'], self.username, pkey=self.keypair['private_key'], server=server1) fip1 = self.fip cidr = netaddr.IPNetwork('10.200.0.0/16') self._test_create_instance_with_network_port( port2, cidr=cidr) fip2 = self.fip for fip in (fip1, fip2): self.check_connectivity(fip['floating_ip_address'], self.username, self.keypair['private_key']) server_ssh_client.ping_host(self.fip['fixed_ip_address']) class NetworkSriov01AdvancedTest(NetworkSriovBaseTest): @decorators.idempotent_id('71a73ba9-646f-4d79-b5c9-7832c814d09d') def test_create_instances_with_normal_normal_ports(self): self._test_create_instances_with_ports('normal', 'normal') class NetworkSriov02AdvancedTest(NetworkSriovBaseTest): @decorators.idempotent_id('e90efadc-54af-4a17-8fc5-22f15629b297') def test_create_instances_with_normal_vf_ports(self): self._test_create_instances_with_ports('direct', 'normal') class NetworkSriov03AdvancedTest(NetworkSriovBaseTest): @decorators.idempotent_id('1757205f-a479-49df-a0f6-411ee54f400c') def test_create_instances_with_vf_vf_ports(self): self._test_create_instances_with_ports('direct', 'direct') class NetworkSriov04Test(NetworkSriovBaseTest): @decorators.idempotent_id('26dba72c-1754-4fca-8c70-1feda1915327') def test_create_instance_with_normal_port(self): self._test_create_instance_with_network_port('normal') self.check_connectivity(self.fip['floating_ip_address'], self.username, self.keypair['private_key']) class NetworkSriov05Test(NetworkSriovBaseTest): @decorators.idempotent_id('39a69007-845f-43aa-adae-4efc808674f4') def test_create_instance_with_direct_port(self): self._test_create_instance_with_network_port('direct') self.check_connectivity(self.fip['floating_ip_address'], self.username, self.keypair['private_key']) class NetworkSriov06Test(NetworkSriovBaseTest): @decorators.idempotent_id('010c1bbb-7b6a-4b55-9eac-5147b69e42f8') def test_create_instance_with_direct_phy_port(self): self._test_create_instance_with_network_port('direct-physical') self.check_connectivity(self.fip['floating_ip_address'], self.username, self.keypair['private_key']) class NetworkSriov07AdvancedTest(NetworkSriovBaseTest): @decorators.idempotent_id('e72b0b01-50b9-4d80-b8c4-9f59447122af') def test_create_instances_with_vf_pf_ports(self): self._test_create_instances_with_ports('direct-physical', 'direct') class NetworkSriov08AdvancedTest(NetworkSriovBaseTest): @decorators.idempotent_id('f6c9a9f1-9e20-4e12-94b8-cf1f45437a40') def test_create_instances_with_normal_pf_ports(self): self._test_create_instances_with_ports('direct-physical', 'normal') class NetworkSriov09AdvancedTest(NetworkSriovBaseTest): @decorators.idempotent_id('31dcaaa3-35d0-4188-8a78-5400b397d7fc') def test_create_instances_with_pf_pf_ports(self): self._test_create_instances_with_ports('direct-physical', 'direct-physical') class NetworkSriov10PFVFNegativeTest(NetworkSriovBaseTest): """Assuming we have host with one SRIOV port and it's already occupied, then creation of VM with VF or PF port should fail We create a VM with PF port in order to occupy SRIOV port. Creating VM with PF or VF should fail as there is no available port """ @decorators.idempotent_id('f2707194-a91d-4e47-8a14-934efa5a6422') def test_try_create_instance_with_pf_port(self): cidr_template = '10.{second_byte}.0.0/16' for i in range(CONF.compute.min_compute_nodes * WB_CONF.sriov_pfs_per_host): cidr_value = cidr_template.format(second_byte=10 + i) self._test_create_instance_with_network_port( port_type='direct-physical', cidr=netaddr.IPNetwork(cidr_value)) self.assertRaises( exceptions.BuildErrorException, self._test_create_instance_with_network_port, port_type='direct-physical', cidr=netaddr.IPNetwork('11.11.1.0/24')) self.assertRaises( exceptions.BuildErrorException, self._test_create_instance_with_network_port, port_type='direct', cidr=netaddr.IPNetwork('21.21.2.0/24')) class NetworkSriov11Test(NetworkSriovBaseTest): @decorators.idempotent_id('7f4dae87-e04a-4e91-80ff-b0baf0ba9c95') def test_create_instance_with_macvtap_port(self): self._test_create_instance_with_network_port('macvtap') self.check_connectivity(self.fip['floating_ip_address'], self.username, self.keypair['private_key']) class NetworkSriov12MultiplePortsTest(NetworkSriovBaseTest): @decorators.idempotent_id('250c2b7b-c4eb-4c3e-813a-6189581ba62e') def test_vms_with_two_ports_different_networks(self): self.fips = [] port_types = ['normal', 'direct'] cidrs = [netaddr.IPNetwork('10.100.0.0/16'), netaddr.IPNetwork('10.101.0.0/16')] self._test_create_instance_with_two_tenant_networks_two_ports( port_types=port_types, cidrs=cidrs) reused_nets = self.networks[0:2] self._test_create_instance_with_two_tenant_networks_two_ports( port_types=port_types, cidrs=cidrs, reused_nets=reused_nets) # check connectivity with FIPs for fip in self.fips: self.check_connectivity(fip['floating_ip_address'], self.username, self.keypair['private_key']) # check connectivity from both VMs to all 4 ports for fip in self.fips: server_ssh_client = remote_client.RemoteClient( fip['floating_ip_address'], self.username, pkey=self.keypair['private_key']) for port in self.ports: fixed_ip = port['fixed_ips'][0]['ip_address'] server_ssh_client.ping_host(fixed_ip)