def test_create_bgp_speaker_illegal_local_asn(self):
     wrong_asn = 65537
     if utils.is_extension_enabled('bgp_4byte_asn', 'network'):
         wrong_asn = 4294967296
     self.assertRaises(lib_exc.BadRequest,
                       self.create_bgp_speaker,
                       local_as=wrong_asn)

def get_project_cleanup_services():
    project_services = []
    # TODO(gmann): Tempest should provide some plugin hook for cleanup
    # script extension to plugin tests also.
    if IS_NOVA:
        project_services.append(ServerService)
        project_services.append(KeyPairService)
        project_services.append(ServerGroupService)
        project_services.append(NovaQuotaService)
    if IS_NEUTRON:
        project_services.append(NetworkFloatingIpService)
        if utils.is_extension_enabled('metering', 'network'):
            project_services.append(NetworkMeteringLabelRuleService)
            project_services.append(NetworkMeteringLabelService)
        project_services.append(NetworkRouterService)
        project_services.append(NetworkPortService)
        project_services.append(NetworkSubnetService)
        project_services.append(NetworkService)
        project_services.append(NetworkSecGroupService)
        project_services.append(NetworkSubnetPoolsService)
    if IS_CINDER:
        project_services.append(SnapshotService)
        project_services.append(VolumeService)
        project_services.append(VolumeQuotaService)
    return project_services

 def resource_setup(cls):
     super(BaseFlowClassifierTest, cls).resource_setup()
     if not utils.is_extension_enabled('flow_classifier', 'network'):
         msg = "FlowClassifier Extension not enabled."
         raise cls.skipException(msg)
     cls.network = cls.create_network()
     cls.subnet = cls.create_subnet(cls.network)
     cls.host_id = socket.gethostname()

 def skip_checks(cls):
     super(AutoAllocateNetworkTest, cls).skip_checks()
     if not CONF.service_available.neutron:
         raise cls.skipException('Neutron is required')
     if not utils.is_extension_enabled('auto-allocated-topology',
                                       'network'):
         raise cls.skipException(
             'auto-allocated-topology extension is not available')

 def test_list_networks_fields(self):
     # Verify specific fields of the networks
     fields = ['id', 'name']
     if utils.is_extension_enabled('net-mtu', 'network'):
         fields.append('mtu')
     body = self.networks_client.list_networks(fields=fields)
     networks = body['networks']
     self.assertNotEmpty(networks, "Network list returned is empty")
     for network in networks:
         self.assertEqual(sorted(network.keys()), sorted(fields))

 def skip_checks(cls):
     super(FloatingIPAdminTestJSON, cls).skip_checks()
     if not utils.is_extension_enabled('router', 'network'):
         msg = "router extension not enabled."
         raise cls.skipException(msg)
     if not CONF.network.public_network_id:
         msg = "The public_network_id option must be specified."
         raise cls.skipException(msg)
     if not CONF.network_feature_enabled.floating_ips:
         raise cls.skipException("Floating ips are not available")

 def skip_checks(cls):
     super(BaseTestCase, cls).skip_checks()
     cfg = config.network
     if not utils.is_extension_enabled('lbaasv2', 'network'):
         msg = 'LBaaS Extension is not enabled'
         raise cls.skipException(msg)
     if not (cfg.project_networks_reachable or cfg.public_network_id):
         msg = ('Either tenant_networks_reachable must be "true", or '
                'public_network_id must be defined.')
         raise cls.skipException(msg)

 def setUp(self):
     LOG.debug("Initializing FWaaSScenarioTest Setup")
     super(TestFWaaS_v2, self).setUp()
     required_exts = ['fwaas_v2', 'security-group', 'router']
     # if self.router_insertion:
     #    required_exts.append('fwaasrouterinsertion')
     for ext in required_exts:
         if not utils.is_extension_enabled(ext, 'network'):
             msg = "%s Extension not enabled." % ext
             raise self.skipException(msg)
     LOG.debug("FWaaSScenarioTest Setup done.")

 def resource_setup(cls):
     super(TestPools, cls).resource_setup()
     if not utils.is_extension_enabled('lbaasv2', 'network'):
         msg = "lbaas extension not enabled."
         raise cls.skipException(msg)
     network_name = data_utils.rand_name('network-')
     cls.network = cls.create_network(network_name)
     cls.subnet = cls.create_subnet(cls.network)
     cls.load_balancer = cls._create_load_balancer(
         tenant_id=cls.subnet.get('tenant_id'),
         vip_subnet_id=cls.subnet.get('id'))

 def _setup_server(self, keypair):
     security_groups = []
     if utils.is_extension_enabled('security-group', 'network'):
         security_group = self._create_security_group()
         security_groups = [{'name': security_group['name']}]
     network, _, _ = self.create_networks()
     server = self.create_server(
         networks=[{'uuid': network['id']}],
         key_name=keypair['name'],
         security_groups=security_groups)
     return server

    def resource_setup(cls):
        super(BgpSpeakerTestJSONBase, cls).resource_setup()
        if not utils.is_extension_enabled('bgp', 'network'):
            msg = "BGP Speaker extension is not enabled."
            raise cls.skipException(msg)

        cls.admin_routerports = []
        cls.admin_floatingips = []
        cls.admin_routers = []
        cls.ext_net_id = CONF.network.public_network_id
        cls._setup_bgp_admin_client()
        cls._setup_bgp_non_admin_client()

 def test_show_network_fields(self):
     # Verify specific fields of a network
     fields = ['id', 'name']
     if utils.is_extension_enabled('net-mtu', 'network'):
         fields.append('mtu')
     body = self.networks_client.show_network(self.network['id'],
                                              fields=fields)
     network = body['network']
     self.assertEqual(sorted(network.keys()), sorted(fields))
     for field_name in fields:
         self.assertEqual(network[field_name], self.network[field_name])
     self.assertNotIn('tenant_id', network)
     self.assertNotIn('project_id', network)

 def resource_setup(cls):
     super(LoadBalancersTestJSON, cls).resource_setup()
     if not utils.is_extension_enabled('lbaasv2', 'network'):
         msg = "lbaas extension not enabled."
         raise cls.skipException(msg)
     network_name = data_utils.rand_name('network')
     cls.network = cls.create_network(network_name)
     cls.subnet = cls.create_subnet(cls.network)
     cls.create_lb_kwargs = {'tenant_id': cls.subnet['tenant_id'],
                             'vip_subnet_id': cls.subnet['id']}
     cls.load_balancer = \
         cls._create_active_load_balancer(**cls.create_lb_kwargs)
     cls.load_balancer_id = cls.load_balancer['id']

 def skip_checks(cls):
     super(TestNetworkBasicOps, cls).skip_checks()
     if not (CONF.network.project_networks_reachable
             or CONF.network.public_network_id):
         msg = ('Either project_networks_reachable must be "true", or '
                'public_network_id must be defined.')
         raise cls.skipException(msg)
     for ext in ['router', 'security-group']:
         if not utils.is_extension_enabled(ext, 'network'):
             msg = "%s extension not enabled." % ext
             raise cls.skipException(msg)
     if not CONF.network_feature_enabled.floating_ips:
         raise cls.skipException("Floating ips are not available")

    def test_list_show_extensions(self):
        # List available extensions for the project
        expected_alias = ['security-group', 'l3_agent_scheduler',
                          'ext-gw-mode', 'binding', 'quotas',
                          'agent', 'dhcp_agent_scheduler', 'provider',
                          'router', 'extraroute', 'external-net',
                          'allowed-address-pairs', 'extra_dhcp_opt',
                          'metering', 'dvr']
        expected_alias = [ext for ext in expected_alias if
                          utils.is_extension_enabled(ext, 'network')]
        actual_alias = list()
        extensions = self.network_extensions_client.list_extensions()
        list_extensions = extensions['extensions']
        # Show and verify the details of the available extensions
        for ext in list_extensions:
            ext_name = ext['name']
            ext_alias = ext['alias']
            actual_alias.append(ext['alias'])
            ext_details = self.network_extensions_client.show_extension(
                ext_alias)
            ext_details = ext_details['extension']

            self.assertIsNotNone(ext_details)
            self.assertIn('updated', ext_details.keys())
            self.assertIn('name', ext_details.keys())
            self.assertIn('description', ext_details.keys())
            self.assertIn('links', ext_details.keys())
            self.assertIn('alias', ext_details.keys())
            self.assertEqual(ext_details['name'], ext_name)
            self.assertEqual(ext_details['alias'], ext_alias)
            self.assertEqual(ext_details, ext)
        # Verify if expected extensions are present in the actual list
        # of extensions returned, but only for those that have been
        # enabled via configuration
        for e in expected_alias:
            if utils.is_extension_enabled(e, 'network'):
                self.assertIn(e, actual_alias)

 def resource_setup(cls):
     super(TestHealthMonitors, cls).resource_setup()
     if not utils.is_extension_enabled('lbaasv2', 'network'):
         msg = "lbaas extension not enabled."
         raise cls.skipException(msg)
     network_name = data_utils.rand_name('network-')
     cls.network = cls.create_network(network_name)
     cls.subnet = cls.create_subnet(cls.network)
     cls.load_balancer = cls._create_load_balancer(
         tenant_id=cls.subnet.get('tenant_id'),
         vip_subnet_id=cls.subnet.get('id'))
     cls.listener = cls._create_listener(
         loadbalancer_id=cls.load_balancer.get('id'),
         protocol=cls.listener_protocol, protocol_port=80)
     cls.pool = cls._create_pool(
         protocol=cls.pool_protocol, lb_algorithm='ROUND_ROBIN',
         listener_id=cls.listener.get('id'))

    def resource_setup(cls):
        super(LoadBalancersTestAdmin, cls).resource_setup()
        if not utils.is_extension_enabled('lbaasv2', 'network'):
            msg = "lbaas extension not enabled."
            raise cls.skipException(msg)
        network_name = data_utils.rand_name('network')
        cls.network = cls.create_network(network_name)
        cls.subnet = cls.create_subnet(cls.network)
        cls.load_balancer = cls.load_balancers_client.create_load_balancer(
            vip_subnet_id=cls.subnet['id'])
        cls._wait_for_load_balancer_status(cls.load_balancer['id'])

        cls.tenant = 'deffb4d7c0584e89a8ec99551565713c'
        cls.tenant_load_balancer = (
            cls.load_balancers_client.create_load_balancer(
                vip_subnet_id=cls.subnet['id'],
                tenant_id=cls.tenant))
        cls._wait_for_load_balancer_status(cls.tenant_load_balancer['id'])

    def resource_setup(cls):
        super(BgpSpeakerScenarioTestJSONBase, cls).resource_setup()
        if not utils.is_extension_enabled('bgp', 'network'):
            msg = "BGP Speaker extension is not enabled."
            raise cls.skipException(msg)

        cls.images = []
        cls.containers = []
        cls.r_ass = []
        cls.r_as_ip = []
        cls.bridges = []
        cls.admin_routerports = []
        cls.admin_floatingips = []
        cls.admin_routers = []
        cls.admin_router_ip = []
        cls.resource_setup_container()
        cls._setup_bgp_admin_client()
        cls._setup_bgp_non_admin_client()
        cls.lock = threading.Lock()

 def skip_checks(cls):
     super(TestSecurityGroupsBasicOps, cls).skip_checks()
     if CONF.network.port_vnic_type in ['direct', 'macvtap']:
         msg = ('Not currently supported when using vnic_type'
                ' direct or macvtap')
         raise cls.skipException(msg)
     if not (CONF.network.project_networks_reachable or
             CONF.network.public_network_id):
         msg = ('Either project_networks_reachable must be "true", or '
                'public_network_id must be defined.')
         raise cls.skipException(msg)
     if not utils.is_extension_enabled('security-group', 'network'):
         msg = "security-group extension not enabled."
         raise cls.skipException(msg)
     if CONF.network.shared_physical_network:
         msg = ('Deployment uses a shared physical network, security '
                'groups not supported')
         raise cls.skipException(msg)
     if not CONF.network_feature_enabled.floating_ips:
         raise cls.skipException("Floating ips are not available")

    def test_delete_external_networks_with_floating_ip(self):
        # Verifies external network can be deleted while still holding
        # (unassociated) floating IPs

        body = self.admin_networks_client.create_network(
            **{'router:external': True})
        external_network = body['network']
        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
                        self.admin_networks_client.delete_network,
                        external_network['id'])
        subnet = self.create_subnet(
            external_network, client=self.admin_subnets_client,
            enable_dhcp=False)
        body = self.admin_floating_ips_client.create_floatingip(
            floating_network_id=external_network['id'])
        created_floating_ip = body['floatingip']
        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
                        self.admin_floating_ips_client.delete_floatingip,
                        created_floating_ip['id'])
        if utils.is_extension_enabled('filter-validation', 'network'):
            floatingip_list = self.admin_floating_ips_client.list_floatingips(
                floating_network_id=external_network['id'])
        else:
            # NOTE(hongbin): This is for testing the backward-compatibility
            # of neutron API although the parameter is a wrong filter
            # for listing floating IPs.
            floatingip_list = self.admin_floating_ips_client.list_floatingips(
                invalid_filter=external_network['id'])
        self.assertIn(created_floating_ip['id'],
                      (f['id'] for f in floatingip_list['floatingips']))
        self.admin_networks_client.delete_network(external_network['id'])
        # Verifies floating ip is deleted
        floatingip_list = self.admin_floating_ips_client.list_floatingips()
        self.assertNotIn(created_floating_ip['id'],
                         (f['id'] for f in floatingip_list['floatingips']))
        # Verifies subnet is deleted
        subnet_list = self.admin_subnets_client.list_subnets()
        self.assertNotIn(subnet['id'],
                         (s['id'] for s in subnet_list))