| # Copyright 2012 OpenStack Foundation |
| # 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 tempest import config |
| from tempest import exceptions |
| from tempest.lib.common.utils import data_utils |
| from tempest.lib.common.utils import test_utils |
| from tempest.lib.common import waiters as lib_waiters |
| from tempest.lib import exceptions as lib_exc |
| import tempest.test |
| |
| CONF = config.CONF |
| |
| |
| class BaseNetworkTest(tempest.test.BaseTestCase): |
| """Base class for the Neutron tests. |
| |
| Per the Neutron API Guide, API v1.x was removed from the source code tree |
| (docs.openstack.org/api/openstack-network/2.0/content/Overview-d1e71.html) |
| Therefore, v2.x of the Neutron API is assumed. It is also assumed that the |
| following options are defined in the [network] section of etc/tempest.conf: |
| |
| project_network_cidr with a block of cidr's from which smaller blocks |
| can be allocated for project networks |
| |
| project_network_mask_bits with the mask bits to be used to partition |
| the block defined by project-network_cidr |
| |
| Finally, it is assumed that the following option is defined in the |
| [service_available] section of etc/tempest.conf |
| |
| neutron as True |
| """ |
| |
| force_tenant_isolation = False |
| credentials = ['primary'] |
| |
| # Default to ipv4. |
| _ip_version = 4 |
| |
| @classmethod |
| def skip_checks(cls): |
| super(BaseNetworkTest, cls).skip_checks() |
| if not CONF.service_available.neutron: |
| raise cls.skipException("Neutron support is required") |
| if cls._ip_version == 6 and not CONF.network_feature_enabled.ipv6: |
| raise cls.skipException("IPv6 Tests are disabled.") |
| |
| @classmethod |
| def setup_credentials(cls): |
| # Create no network resources for these test. |
| cls.set_network_resources() |
| super(BaseNetworkTest, cls).setup_credentials() |
| |
| @classmethod |
| def setup_clients(cls): |
| super(BaseNetworkTest, cls).setup_clients() |
| cls.agents_client = cls.os_primary.network_agents_client |
| cls.network_extensions_client =\ |
| cls.os_primary.network_extensions_client |
| cls.networks_client = cls.os_primary.networks_client |
| cls.routers_client = cls.os_primary.routers_client |
| cls.subnetpools_client = cls.os_primary.subnetpools_client |
| cls.subnets_client = cls.os_primary.subnets_client |
| cls.ports_client = cls.os_primary.ports_client |
| cls.floating_ips_port_forwarding_client =\ |
| cls.os_primary.floating_ips_port_forwarding_client |
| cls.quotas_client = cls.os_primary.network_quotas_client |
| cls.floating_ips_client = cls.os_primary.floating_ips_client |
| cls.security_groups_client = cls.os_primary.security_groups_client |
| cls.security_group_rules_client = ( |
| cls.os_primary.security_group_rules_client) |
| cls.network_versions_client = cls.os_primary.network_versions_client |
| cls.service_providers_client = cls.os_primary.service_providers_client |
| cls.tags_client = cls.os_primary.tags_client |
| cls.log_resource_client = cls.os_primary.log_resource_client |
| cls.loggable_resource_client = cls.os_primary.loggable_resource_client |
| |
| @classmethod |
| def resource_setup(cls): |
| super(BaseNetworkTest, cls).resource_setup() |
| cls.ethertype = "IPv" + str(cls._ip_version) |
| if cls._ip_version == 4: |
| cls.cidr = netaddr.IPNetwork(CONF.network.project_network_cidr) |
| cls.mask_bits = CONF.network.project_network_mask_bits |
| elif cls._ip_version == 6: |
| cls.cidr = netaddr.IPNetwork(CONF.network.project_network_v6_cidr) |
| cls.mask_bits = CONF.network.project_network_v6_mask_bits |
| |
| @classmethod |
| def create_network(cls, network_name=None, **kwargs): |
| """Wrapper utility that returns a test network.""" |
| network_name = network_name or data_utils.rand_name( |
| cls.__name__ + '-test-network', prefix=CONF.resource_name_prefix) |
| |
| body = cls.networks_client.create_network(name=network_name, **kwargs) |
| network = body['network'] |
| cls.addClassResourceCleanup(test_utils.call_and_ignore_notfound_exc, |
| cls.networks_client.delete_network, |
| network['id']) |
| return network |
| |
| @classmethod |
| def create_subnet(cls, network, gateway='', cidr=None, mask_bits=None, |
| ip_version=None, client=None, **kwargs): |
| """Wrapper utility that returns a test subnet.""" |
| # allow tests to use admin client |
| if not client: |
| client = cls.subnets_client |
| |
| # The cidr and mask_bits depend on the ip version. |
| ip_version = ip_version if ip_version is not None else cls._ip_version |
| gateway_not_set = gateway == '' |
| if ip_version == 4: |
| cidr = cidr or netaddr.IPNetwork(CONF.network.project_network_cidr) |
| mask_bits = mask_bits or CONF.network.project_network_mask_bits |
| elif ip_version == 6: |
| cidr = (cidr or |
| netaddr.IPNetwork(CONF.network.project_network_v6_cidr)) |
| mask_bits = mask_bits or CONF.network.project_network_v6_mask_bits |
| # Find a cidr that is not in use yet and create a subnet with it |
| for subnet_cidr in cidr.subnet(mask_bits): |
| if gateway_not_set: |
| gateway_ip = str(netaddr.IPAddress(subnet_cidr) + 1) |
| else: |
| gateway_ip = gateway |
| try: |
| body = client.create_subnet( |
| network_id=network['id'], |
| cidr=str(subnet_cidr), |
| ip_version=ip_version, |
| gateway_ip=gateway_ip, |
| **kwargs) |
| break |
| except lib_exc.BadRequest as e: |
| is_overlapping_cidr = 'overlaps with another subnet' in str(e) |
| if not is_overlapping_cidr: |
| raise |
| else: |
| message = 'Available CIDR for subnet creation could not be found' |
| raise exceptions.BuildErrorException(message) |
| subnet = body['subnet'] |
| cls.addClassResourceCleanup(test_utils.call_and_ignore_notfound_exc, |
| cls.subnets_client.delete_subnet, |
| subnet['id']) |
| return subnet |
| |
| @classmethod |
| def create_port(cls, network, **kwargs): |
| if 'name' not in kwargs: |
| kwargs['name'] = data_utils.rand_name( |
| cls.__name__, prefix=CONF.resource_name_prefix) |
| """Wrapper utility that returns a test port.""" |
| body = cls.ports_client.create_port(network_id=network['id'], |
| **kwargs) |
| port = body['port'] |
| cls.addClassResourceCleanup(test_utils.call_and_ignore_notfound_exc, |
| cls.ports_client.delete_port, port['id']) |
| return port |
| |
| @classmethod |
| def update_port(cls, port, **kwargs): |
| """Wrapper utility that updates a test port.""" |
| body = cls.ports_client.update_port(port['id'], |
| **kwargs) |
| return body['port'] |
| |
| @classmethod |
| def create_router(cls, router_name=None, admin_state_up=False, |
| external_network_id=None, enable_snat=None, |
| **kwargs): |
| router_name = router_name or data_utils.rand_name( |
| cls.__name__ + "-router", prefix=CONF.resource_name_prefix) |
| |
| ext_gw_info = {} |
| if external_network_id: |
| ext_gw_info['network_id'] = external_network_id |
| if enable_snat is not None: |
| ext_gw_info['enable_snat'] = enable_snat |
| body = cls.routers_client.create_router( |
| name=router_name, external_gateway_info=ext_gw_info, |
| admin_state_up=admin_state_up, **kwargs) |
| router = body['router'] |
| cls.addClassResourceCleanup(test_utils.call_and_ignore_notfound_exc, |
| cls.delete_router, router) |
| return router |
| |
| @classmethod |
| def create_floatingip(cls, external_network_id): |
| """Wrapper utility that returns a test floating IP.""" |
| body = cls.floating_ips_client.create_floatingip( |
| floating_network_id=external_network_id) |
| fip = body['floatingip'] |
| cls.addClassResourceCleanup(test_utils.call_and_ignore_notfound_exc, |
| cls.floating_ips_client.delete_floatingip, |
| fip['id']) |
| return fip |
| |
| @classmethod |
| def create_router_interface(cls, router_id, subnet_id): |
| """Wrapper utility that returns a router interface.""" |
| interface = cls.routers_client.add_router_interface( |
| router_id, subnet_id=subnet_id) |
| return interface |
| |
| @classmethod |
| def delete_router(cls, router): |
| body = cls.ports_client.list_ports(device_id=router['id']) |
| interfaces = body['ports'] |
| for i in interfaces: |
| test_utils.call_and_ignore_notfound_exc( |
| cls.routers_client.remove_router_interface, router['id'], |
| subnet_id=i['fixed_ips'][0]['subnet_id']) |
| lib_waiters.wait_router_interface_removed( |
| cls.ports_client, router['id'], |
| subnet_id=i['fixed_ips'][0]['subnet_id']) |
| cls.routers_client.delete_router(router['id']) |
| |
| |
| class BaseAdminNetworkTest(BaseNetworkTest): |
| |
| credentials = ['primary', 'admin'] |
| |
| @classmethod |
| def setup_clients(cls): |
| super(BaseAdminNetworkTest, cls).setup_clients() |
| cls.admin_agents_client = cls.os_admin.network_agents_client |
| cls.admin_networks_client = cls.os_admin.networks_client |
| cls.admin_routers_client = cls.os_admin.routers_client |
| cls.admin_subnets_client = cls.os_admin.subnets_client |
| cls.admin_ports_client = cls.os_admin.ports_client |
| cls.admin_quotas_client = cls.os_admin.network_quotas_client |
| cls.admin_floating_ips_client = cls.os_admin.floating_ips_client |
| cls.admin_metering_labels_client = cls.os_admin.metering_labels_client |
| cls.admin_metering_label_rules_client = ( |
| cls.os_admin.metering_label_rules_client) |