def create_server(cls):
     tenant_network = cls.get_tenant_network()
     body, server = compute.create_test_server(
         cls.os, tenant_network=tenant_network, name=data_utils.rand_name("ceilometer-instance"), wait_until="ACTIVE"
     )
     cls.server_ids.append(body["id"])
     return body

    def test_reassign_port_between_servers(self):
        """Tests the following:

        1. Create a port in Neutron.
        2. Create two servers in Nova.
        3. Attach the port to the first server.
        4. Detach the port from the first server.
        5. Attach the port to the second server.
        6. Detach the port from the second server.
        """
        network = self.get_tenant_network()
        network_id = network['id']
        port = self.ports_client.create_port(network_id=network_id)
        port_id = port['port']['id']
        self.addCleanup(self.ports_client.delete_port, port_id)

        # create two servers
        _, servers = compute.create_test_server(
            self.os_primary, tenant_network=network,
            wait_until='ACTIVE', min_count=2)
        # add our cleanups for the servers since we bypassed the base class
        for server in servers:
            self.addCleanup(self.delete_server, server['id'])

        for server in servers:
            # attach the port to the server
            iface = self.interfaces_client.create_interface(
                server['id'], port_id=port_id)['interfaceAttachment']
            self._check_interface(iface, server_id=server['id'],
                                  port_id=port_id)

            # detach the port from the server; this is a cast in the compute
            # API so we have to poll the port until the device_id is unset.
            self.interfaces_client.delete_interface(server['id'], port_id)
            self.wait_for_port_detach(port_id)

    def create_test_server(cls, validatable=False, volume_backed=False,
                           **kwargs):
        """Wrapper utility that returns a test server.

        This wrapper utility calls the common create test server and
        returns a test server. The purpose of this wrapper is to minimize
        the impact on the code of the tests already using this
        function.

        :param validatable: Whether the server will be pingable or sshable.
        :param volume_backed: Whether the instance is volume backed or not.
        """
        if 'name' not in kwargs:
            kwargs['name'] = data_utils.rand_name(cls.__name__ + "-server")
        tenant_network = cls.get_tenant_network()
        body, servers = compute.create_test_server(
            cls.os,
            validatable,
            validation_resources=cls.validation_resources,
            tenant_network=tenant_network,
            volume_backed=volume_backed,
            **kwargs)

        cls.servers.extend(servers)

        return body

    def create_test_server(cls, validatable=False, volume_backed=False,
                           **kwargs):
        """Wrapper utility that returns a test server.

        This wrapper utility calls the common create test server and
        returns a test server. The purpose of this wrapper is to minimize
        the impact on the code of the tests already using this
        function.

        :param validatable: Whether the server will be pingable or sshable.
        :param volume_backed: Whether the instance is volume backed or not.
        """
        if 'name' not in kwargs:
            kwargs['name'] = data_utils.rand_name(cls.__name__ + "-server")

        request_version = api_version_request.APIVersionRequest(
            cls.request_microversion)
        v2_37_version = api_version_request.APIVersionRequest('2.37')

        # NOTE(snikitin): since microversion v2.37 'networks' field is required
        if request_version >= v2_37_version and 'networks' not in kwargs:
            kwargs['networks'] = 'none'

        tenant_network = cls.get_tenant_network()
        body, servers = compute.create_test_server(
            cls.os_primary,
            validatable,
            validation_resources=cls.validation_resources,
            tenant_network=tenant_network,
            volume_backed=volume_backed,
            **kwargs)

        cls.servers.extend(servers)

        return body

 def create_server(cls, name, **kwargs):
     tenant_network = cls.get_tenant_network()
     body, _ = compute.create_test_server(
         cls.os,
         tenant_network=tenant_network,
         name=name,
         **kwargs)
     return body

 def test_server_create_no_allocate(self):
     """Tests that no networking is allocated for the server."""
     # create the server with no networking
     server, _ = compute.create_test_server(self.os, networks="none", wait_until="ACTIVE")
     self.addCleanup(self.delete_server, server["id"])
     # get the server ips
     addresses = self.servers_client.list_addresses(server["id"])["addresses"]
     # assert that there is no networking
     self.assertEqual({}, addresses)

 def create_server(cls, validatable=False, volume_backed=False, **kwargs):
     tenant_network = cls.get_tenant_network()
     body, servers = compute.create_test_server(
         cls.os,
         validatable,
         validation_resources=cls.validation_resources,
         tenant_network=tenant_network,
         **kwargs)
     return body

 def create_server(cls):
     tenant_network = cls.get_tenant_network()
     body, server = compute.create_test_server(
         cls.os_primary,
         tenant_network=tenant_network,
         name=data_utils.rand_name('ceilometer-instance'),
         wait_until='ACTIVE')
     cls.server_ids.append(body['id'])
     return body

 def create_server(cls, name, **kwargs):
     tenant_network = cls.get_tenant_network()
     kwargs['availability_zone'] = CONF.volume.get('availability_zone')
     body, _ = compute.create_test_server(
         cls.os,
         tenant_network=tenant_network,
         name=name,
         **kwargs)
     return body

 def test_server_create_no_allocate(self):
     """Tests that no networking is allocated for the server."""
     # create the server with no networking
     server, _ = compute.create_test_server(
         self.os_primary, networks='none', wait_until='ACTIVE')
     self.addCleanup(self.delete_server, server['id'])
     # get the server ips
     addresses = self.servers_client.list_addresses(
         server['id'])['addresses']
     # assert that there is no networking
     self.assertEqual({}, addresses)

    def create_test_server(self, wait_until=None):
        flavors = self.flavors_client.list_flavors()['flavors']
        images = self.image_client.list_images()['images']
        tenant_network = self.get_tenant_network()
        body, servers = compute.create_test_server(
            self.os,
            wait_until=wait_until,
            image_id=images[0]['id'],
            flavor=flavors[0]['id'],
            tenant_network=tenant_network)

        server = self.os_primary.servers_client.show_server(body['id'])[
            'server']
        return server, flavors[0]

    def create_server(self, **kwargs):
        name = kwargs.get("name", data_utils.rand_name(self.__class__.__name__ + "-instance"))

        tenant_network = self.get_tenant_network()
        body, _ = compute.create_test_server(self.os, tenant_network=tenant_network, name=name, **kwargs)

        self.addCleanup(
            test_utils.call_and_ignore_notfound_exc,
            waiters.wait_for_server_termination,
            self.servers_client,
            body["id"],
        )
        self.addCleanup(test_utils.call_and_ignore_notfound_exc, self.servers_client.delete_server, body["id"])
        return body

 def test_multiple_create(self):
     tenant_network = self.get_tenant_network()
     body, servers = compute.create_test_server(
         self.os_primary,
         wait_until='ACTIVE',
         min_count=2,
         tenant_network=tenant_network)
     for server in servers:
         self.addCleanup(self.servers_client.delete_server, server['id'])
     # NOTE(maurosr): do status response check and also make sure that
     # reservation_id is not in the response body when the request send
     # contains return_reservation_id=False
     self.assertNotIn('reservation_id', body)
     self.assertEqual(2, len(servers))

    def create_test_server(cls, validatable=False, volume_backed=False,
                           validation_resources=None, **kwargs):
        """Wrapper utility that returns a test server.

        This wrapper utility calls the common create test server and
        returns a test server. The purpose of this wrapper is to minimize
        the impact on the code of the tests already using this
        function.

        :param validatable: Whether the server will be pingable or sshable.
        :param volume_backed: Whether the instance is volume backed or not.
        :param validation_resources: Dictionary of validation resources as
            returned by `get_class_validation_resources`.
        :param kwargs: Extra arguments are passed down to the
            `compute.create_test_server` call.
        """
        if 'name' not in kwargs:
            kwargs['name'] = data_utils.rand_name(cls.__name__ + "-server")

        request_version = api_version_request.APIVersionRequest(
            cls.request_microversion)
        v2_37_version = api_version_request.APIVersionRequest('2.37')

        tenant_network = cls.get_tenant_network()
        # NOTE(snikitin): since microversion v2.37 'networks' field is required
        if (request_version >= v2_37_version and 'networks' not in kwargs and
            not tenant_network):
            kwargs['networks'] = 'none'

        body, servers = compute.create_test_server(
            cls.os_primary,
            validatable,
            validation_resources=validation_resources,
            tenant_network=tenant_network,
            volume_backed=volume_backed,
            **kwargs)

        # For each server schedule wait and delete, so we first delete all
        # and then wait for all
        for server in servers:
            cls.addClassResourceCleanup(waiters.wait_for_server_termination,
                                        cls.servers_client, server['id'])
        for server in servers:
            cls.addClassResourceCleanup(
                test_utils.call_and_ignore_notfound_exc,
                cls.servers_client.delete_server, server['id'])

        return body

    def create_test_server(cls, validatable=False, **kwargs):
        """Wrapper utility that returns a test server.

        This wrapper utility calls the common create test server and
        returns a test server. The purpose of this wrapper is to minimize
        the impact on the code of the tests already using this
        function.
        """
        tenant_network = cls.get_tenant_network()
        body, servers = compute.create_test_server(
            cls.os, validatable, validation_resources=cls.validation_resources, tenant_network=tenant_network, **kwargs
        )

        cls.servers.extend(servers)

        return body

    def create_server(self, wait_until='ACTIVE', **kwargs):
        name = kwargs.pop(
            'name',
            data_utils.rand_name(self.__class__.__name__ + '-instance'))

        tenant_network = self.get_tenant_network()
        body, _ = compute.create_test_server(
            self.os_primary,
            tenant_network=tenant_network,
            name=name,
            wait_until=wait_until,
            **kwargs)

        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
                        waiters.wait_for_server_termination,
                        self.servers_client, body['id'])
        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
                        self.servers_client.delete_server, body['id'])
        return body

 def test_list_servers_filter_by_exist_host(self):
     # Filter the list of servers by existent host
     name = data_utils.rand_name(self.__class__.__name__ + "-server")
     network = self.get_tenant_network()
     network_kwargs = fixed_network.set_networks_kwarg(network)
     # We need to create the server as an admin, so we can't use
     # self.create_test_server() here as this method creates the server
     # in the "primary" (i.e non-admin) tenant.
     test_server, _ = compute.create_test_server(self.os_adm, wait_until="ACTIVE", name=name, **network_kwargs)
     self.addCleanup(self.client.delete_server, test_server["id"])
     server = self.client.show_server(test_server["id"])["server"]
     self.assertEqual(server["status"], "ACTIVE")
     hostname = server[self._host_key]
     params = {"host": hostname}
     body = self.client.list_servers(**params)
     servers = body["servers"]
     nonexistent_params = {"host": "nonexistent_host"}
     nonexistent_body = self.client.list_servers(**nonexistent_params)
     nonexistent_servers = nonexistent_body["servers"]
     self.assertIn(test_server["id"], map(lambda x: x["id"], servers))
     self.assertNotIn(test_server["id"], map(lambda x: x["id"], nonexistent_servers))

 def _create_validatable_server():
     _, servers = compute.create_test_server(
         self.os_primary, tenant_network=network,
         wait_until='ACTIVE', validatable=True,
         validation_resources=validation_resources)
     return servers[0]

    def create_server(self, name=None, image_id=None, flavor=None,
                      validatable=False, wait_until='ACTIVE',
                      clients=None, **kwargs):
        """Wrapper utility that returns a test server.

        This wrapper utility calls the common create test server and
        returns a test server. The purpose of this wrapper is to minimize
        the impact on the code of the tests already using this
        function.
        """

        # NOTE(jlanoux): As a first step, ssh checks in the scenario
        # tests need to be run regardless of the run_validation and
        # validatable parameters and thus until the ssh validation job
        # becomes voting in CI. The test resources management and IP
        # association are taken care of in the scenario tests.
        # Therefore, the validatable parameter is set to false in all
        # those tests. In this way create_server just return a standard
        # server and the scenario tests always perform ssh checks.

        # Needed for the cross_tenant_traffic test:
        if clients is None:
            clients = self.os_primary

        if name is None:
            name = data_utils.rand_name(self.__class__.__name__ + "-server")

        vnic_type = CONF.network.port_vnic_type

        # If vnic_type is configured create port for
        # every network
        if vnic_type:
            ports = []

            create_port_body = {'binding:vnic_type': vnic_type,
                                'namestart': 'port-smoke'}
            if kwargs:
                # Convert security group names to security group ids
                # to pass to create_port
                if 'security_groups' in kwargs:
                    security_groups = \
                        clients.security_groups_client.list_security_groups(
                        ).get('security_groups')
                    sec_dict = dict([(s['name'], s['id'])
                                    for s in security_groups])

                    sec_groups_names = [s['name'] for s in kwargs.pop(
                        'security_groups')]
                    security_groups_ids = [sec_dict[s]
                                           for s in sec_groups_names]

                    if security_groups_ids:
                        create_port_body[
                            'security_groups'] = security_groups_ids
                networks = kwargs.pop('networks', [])
            else:
                networks = []

            # If there are no networks passed to us we look up
            # for the project's private networks and create a port.
            # The same behaviour as we would expect when passing
            # the call to the clients with no networks
            if not networks:
                networks = clients.networks_client.list_networks(
                    **{'router:external': False, 'fields': 'id'})['networks']

            # It's net['uuid'] if networks come from kwargs
            # and net['id'] if they come from
            # clients.networks_client.list_networks
            for net in networks:
                net_id = net.get('uuid', net.get('id'))
                if 'port' not in net:
                    port = self._create_port(network_id=net_id,
                                             client=clients.ports_client,
                                             **create_port_body)
                    ports.append({'port': port['id']})
                else:
                    ports.append({'port': net['port']})
            if ports:
                kwargs['networks'] = ports
            self.ports = ports

        tenant_network = self.get_tenant_network()

        body, servers = compute.create_test_server(
            clients,
            tenant_network=tenant_network,
            wait_until=wait_until,
            name=name, flavor=flavor,
            image_id=image_id, **kwargs)

        self.addCleanup(waiters.wait_for_server_termination,
                        clients.servers_client, body['id'])
        self.addCleanup(test_utils.call_and_ignore_notfound_exc,
                        clients.servers_client.delete_server, body['id'])
        server = clients.servers_client.show_server(body['id'])['server']
        return server