def test_list_show_extensions(self):
        # List available extensions for the tenant
        expected_alias = ['security-group', 'binding', 'quotas',
                          'agent', 'router', 'external-net',
                          'allowed-address-pairs', 'extra_dhcp_opt']
        expected_alias = [ext for ext in expected_alias if
                          test.is_extension_enabled(ext, 'network')]
        actual_alias = list()
        extensions = self.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.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('namespace', 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 test.is_extension_enabled(e, 'network'):
                self.assertIn(e, actual_alias)

 def skip_checks(cls):
     super(LoadBalancerTestJSON, cls).skip_checks()
     if not test.is_extension_enabled('lbaas', 'network'):
         msg = "lbaas extension not enabled."
         raise cls.skipException(msg)
     if not test.is_extension_enabled('nsxv-router-type', 'network'):
         msg = "nsxv-router-type extension is not enabled"
         raise cls.skipException(msg)

 def skip_checks(cls):
     super(L2GatewayConnectionTest, cls).skip_checks()
     if not test.is_extension_enabled('l2-gateway', 'network'):
         msg = "l2-gateway extension not enabled."
         raise cls.skipException(msg)
     if not test.is_extension_enabled('l2-gateway-connection',
                                      'network'):
         msg = "l2-gateway-connection extension is not enabled"
         raise cls.skipException(msg)
     # skip test if CONF session:l2gw does not have the following opts
     cls.getattr_or_skip_test("device_one_vlan")
     cls.getattr_or_skip_test("vlan_subnet_ipv4_dict")

 def test_show_network(self):
     # Verify the details of a network
     body = self.client.show_network(self.network['id'])
     network = body['network']
     fields = ['id', 'name']
     if test.is_extension_enabled('net-mtu', 'network'):
         fields.append('mtu')
     for key in fields:
         self.assertEqual(network[key], self.network[key])
     project_id = self.client.tenant_id
     self.assertEqual(project_id, network['tenant_id'])
     if test.is_extension_enabled('project-id', 'network'):
         self.assertEqual(project_id, network['project_id'])

    def setUpClass(cls):
        super(FlavorsExtraSpecsNegativeTestJSON, cls).setUpClass()
        if not test.is_extension_enabled('FlavorExtraData', 'compute'):
            msg = "FlavorExtraData extension not enabled."
            raise cls.skipException(msg)

        cls.client = cls.os_adm.flavors_client
        flavor_name = data_utils.rand_name('test_flavor')
        ram = 512
        vcpus = 1
        disk = 10
        ephemeral = 10
        cls.new_flavor_id = data_utils.rand_int_id(start=1000)
        swap = 1024
        rxtx = 1
        # Create a flavor
        resp, cls.flavor = cls.client.create_flavor(
            flavor_name,
            ram,
            vcpus,
            disk,
            cls.new_flavor_id,
            ephemeral=ephemeral,
            swap=swap,
            rxtx=rxtx)

 def skip_checks(cls):
     super(L2GatewayTest, cls).skip_checks()
     if not test.is_extension_enabled('l2-gateway', 'network'):
         msg = "l2-gateway extension not enabled."
         raise cls.skipException(msg)
     # if CONF attr device_on_vlan not defined, SKIP entire test suite
     cls.getattr_or_skip_test("device_one_vlan")

    def resource_setup(cls):
        super(L7PolicyRulesTestJSON, cls).resource_setup()
        if not test.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']

        # Create listener for tests
        cls.create_listener_kwargs = {'loadbalancer_id': cls.load_balancer_id,
                                      'protocol': "HTTP",
                                      'protocol_port': "80"}
        cls.listener = (
            cls._create_listener(**cls.create_listener_kwargs))
        cls.listener_id = cls.listener['id']

        # Create pool for tests
        cls.create_pool_kwargs = {'listener_id': cls.listener_id,
                                  'protocol': "HTTP",
                                  'lb_algorithm': "ROUND_ROBIN"}
        cls.pool = (
            cls._create_pool(**cls.create_pool_kwargs))
        cls.pool_id = cls.pool['id']

        cls.partition = 'Project_' + cls.subnet['tenant_id']
        cls.vs_name = 'Project_' + cls.listener_id
        cls.bigip = BigIpClient()

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

        cls.ext_net_id = CONF.network.public_network_id

 def verify_flavor_response_extension(flavor):
     # check some extensions for the flavor create/show/detail response
     self.assertEqual(flavor['swap'], 0)
     if test.is_extension_enabled("os-flavor-rxtx", "compute_v3"):
         self.assertEqual(int(flavor['os-flavor-rxtx:rxtx_factor']), 1)
     self.assertEqual(int(flavor['ephemeral']), 0)
     self.assertEqual(flavor['flavor-access:is_public'], True)

    def resource_setup(cls):
        super(TestFlavorsJson, cls).resource_setup()
        if not test.is_extension_enabled('flavors', 'network'):
            msg = "flavors extension not enabled."
            raise cls.skipException(msg)

        # Use flavors service type as know this is loaded
        service_type = "FLAVORS"
        description_flavor = "flavor is created by tempest"
        name_flavor = "Best flavor created by tempest"

        # The check above will pass if api_extensions=all, which does
        # not mean flavors extension itself is present.
        try:
            cls.flavor = cls.create_flavor(name_flavor, description_flavor,
                                           service_type)
        except lib_exc.NotFound:
            msg = "flavors plugin not enabled."
            raise cls.skipException(msg)

        description_sp = "service profile created by tempest"
        # Drivers are supported as is an empty driver field.  Use an
        # empty field for now since otherwise driver is validated against the
        # servicetype configuration which may differ in test scenarios.
        driver = ""
        metainfo = '{"data": "value"}'
        cls.service_profile = cls.create_service_profile(
            description=description_sp, metainfo=metainfo, driver=driver)

 def skip_checks(cls):
     super(TestProviderDistributedRouterTenantNetwork,
           cls).skip_checks()
     for ext in ['dvr', 'nsxv-router-type']:
         if not test.is_extension_enabled(ext, 'network'):
             msg = "%s extension not enabled." % ext
             raise cls.skipException(msg)

 def skip_checks(cls):
     """
     Skip running test if we do not meet criteria to run the tests.
     """
     super(BaseL2GatewayTest, cls).skip_checks()
     if not test.is_extension_enabled("l2-gateway", "network"):
         raise cls.skipException("l2-gateway extension not enabled.")

    def resource_setup(cls):
        """Setup client fixtures for test suite."""
        super(MemberStatusTestJSON, cls).resource_setup()
        if not test.is_extension_enabled('lbaas', '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']

        # create listener
        listener_kwargs = {'loadbalancer_id': cls.load_balancer_id,
                           'protocol': 'HTTP',
                           'protocol_port': 80}
        cls.listener = cls._create_listener(**listener_kwargs)

        # get an RPC client for calling into driver
        cls.client = cls.plugin_rpc.get_client()
        cls.context = cls.plugin_rpc.get_context()

 def resource_setup(cls):
     super(FWaaSExtensionTestJSON, cls).resource_setup()
     if not test.is_extension_enabled("fwaas", "network"):
         msg = "FWaaS Extension not enabled."
         raise cls.skipException(msg)
     cls.fw_rule = cls.create_firewall_rule("allow", "tcp")
     cls.fw_policy = cls.create_firewall_policy()

 def setUpClass(cls):
     super(FWaaSExtensionTestJSON, cls).setUpClass()
     if not test.is_extension_enabled('fwaas', 'network'):
         msg = "FWaaS Extension not enabled."
         raise cls.skipException(msg)
     cls.fw_rule = cls.create_firewall_rule("allow", "tcp")
     cls.fw_policy = cls.create_firewall_policy()

 def resource_setup(cls):
     super(L3AgentSchedulerTestJSON, cls).resource_setup()
     body = cls.admin_client.list_agents()
     agents = body['agents']
     for agent in agents:
         # TODO(armax): falling back on default _agent_mode can be
         # dropped as soon as Icehouse is dropped.
         agent_mode = (
             agent['configurations'].get('agent_mode', cls._agent_mode))
         if agent['agent_type'] == AGENT_TYPE and agent_mode in AGENT_MODES:
             cls.agent = agent
             break
     else:
         msg = "L3 Agent Scheduler enabled in conf, but L3 Agent not found"
         raise exceptions.InvalidConfiguration(msg)
     cls.router = cls.create_router(data_utils.rand_name('router'))
     # NOTE(armax): If DVR is an available extension, and the created router
     # is indeed a distributed one, more resources need to be provisioned
     # in order to bind the router to the L3 agent.
     # That said, let's preserve the existing test logic, where the extra
     # query and setup steps are only required if the extension is available
     # and only if the router's default type is distributed.
     if test.is_extension_enabled('dvr', 'network'):
         is_dvr_router = cls.admin_client.show_router(
             cls.router['id'])['router'].get('distributed', False)
         if is_dvr_router:
             cls.network = cls.create_network()
             cls.create_subnet(cls.network)
             cls.port = cls.create_port(cls.network)
             cls.client.add_router_interface_with_port_id(
                 cls.router['id'], cls.port['id'])

 def setUpClass(cls):
     super(TestNetworkBasicOps, cls).setUpClass()
     for ext in ['router', 'security-group']:
         if not test.is_extension_enabled(ext, 'network'):
             msg = "%s extension not enabled." % ext
             raise cls.skipException(msg)
     cls.check_preconditions()

 def setUpClass(cls):
     super(RoutersTest, cls).setUpClass()
     if not test.is_extension_enabled('router', 'network'):
         msg = "router extension not enabled."
         raise cls.skipException(msg)
     admin_manager = clients.AdminManager()
     cls.identity_admin_client = admin_manager.identity_client

def get_tenant_cleanup_services():
    tenant_services = []
    # TODO(gmann): Tempest should provide some plugin hook for cleanup
    # script extension to plugin tests also.
    if IS_NOVA:
        tenant_services.append(ServerService)
        tenant_services.append(KeyPairService)
        tenant_services.append(SecurityGroupService)
        tenant_services.append(ServerGroupService)
        if not IS_NEUTRON:
            tenant_services.append(FloatingIpService)
        tenant_services.append(NovaQuotaService)
    if IS_HEAT:
        tenant_services.append(StackService)
    if IS_NEUTRON:
        tenant_services.append(NetworkFloatingIpService)
        if test.is_extension_enabled("metering", "network"):
            tenant_services.append(NetworkMeteringLabelRuleService)
            tenant_services.append(NetworkMeteringLabelService)
        tenant_services.append(NetworkRouterService)
        tenant_services.append(NetworkPortService)
        tenant_services.append(NetworkSubnetService)
        tenant_services.append(NetworkService)
        tenant_services.append(NetworkSecGroupService)
    if IS_CINDER:
        tenant_services.append(SnapshotService)
        tenant_services.append(VolumeService)
        tenant_services.append(VolumeQuotaService)
    return tenant_services

def get_tenant_cleanup_services():
    tenant_services = []
    if IS_CEILOMETER:
        tenant_services.append(TelemetryAlarmService)
    if IS_NOVA:
        tenant_services.append(ServerService)
        tenant_services.append(KeyPairService)
        tenant_services.append(SecurityGroupService)
        tenant_services.append(ServerGroupService)
        if not IS_NEUTRON:
            tenant_services.append(FloatingIpService)
        tenant_services.append(NovaQuotaService)
    if IS_HEAT:
        tenant_services.append(StackService)
    if IS_NEUTRON:
        tenant_services.append(NetworkFloatingIpService)
        if test.is_extension_enabled('metering', 'network'):
            tenant_services.append(NetworkMeteringLabelRuleService)
            tenant_services.append(NetworkMeteringLabelService)
        tenant_services.append(NetworkRouterService)
        tenant_services.append(NetworkPortService)
        tenant_services.append(NetworkSubnetService)
        tenant_services.append(NetworkService)
        tenant_services.append(NetworkSecGroupService)
    if IS_CINDER:
        tenant_services.append(SnapshotService)
        tenant_services.append(VolumeService)
        tenant_services.append(VolumeQuotaService)
    return tenant_services