def _get_unused_flavor_id(self):
     flavor_id = data_utils.rand_int_id(start=1000)
     while True:
         try:
             self.flavors_client.get_flavor_details(flavor_id)
         except lib_exc.NotFound:
             break
         flavor_id = data_utils.rand_int_id(start=1000)
     return flavor_id

    def test_list_non_public_flavor(self):
        # Create a flavor with os-flavor-access:is_public false.
        # The flavor should not be present in list_details as the
        # tenant is not automatically added access list.
        # This operation requires the user to have 'admin' role
        flavor_name = data_utils.rand_name(self.flavor_name_prefix)
        new_flavor_id = data_utils.rand_int_id(start=1000)

        # Create the flavor
        flavor = self.client.create_flavor(flavor_name,
                                           self.ram, self.vcpus,
                                           self.disk,
                                           new_flavor_id,
                                           is_public="False")
        self.addCleanup(self.flavor_clean_up, flavor['id'])
        # Verify flavor is retrieved
        flag = False
        flavors = self.client.list_flavors_with_detail()
        for flavor in flavors:
            if flavor['name'] == flavor_name:
                flag = True
        self.assertFalse(flag)

        # Verify flavor is not retrieved with other user
        flag = False
        flavors = self.user_client.list_flavors_with_detail()
        for flavor in flavors:
            if flavor['name'] == flavor_name:
                flag = True
        self.assertFalse(flag)

    def test_flavor_access_add_remove(self):
        # Test to add and remove flavor access to a given tenant.
        flavor_name = data_utils.rand_name(self.flavor_name_prefix)
        new_flavor_id = data_utils.rand_int_id(start=1000)
        new_flavor = self.client.create_flavor(flavor_name,
                                               self.ram, self.vcpus,
                                               self.disk,
                                               new_flavor_id,
                                               is_public='False')
        self.addCleanup(self.client.delete_flavor, new_flavor['id'])
        # Add flavor access to a tenant.
        resp_body = {
            "tenant_id": str(self.tenant_id),
            "flavor_id": str(new_flavor['id']),
        }
        add_body = \
            self.client.add_flavor_access(new_flavor['id'], self.tenant_id)
        self.assertIn(resp_body, add_body)

        # The flavor is present in list.
        flavors = self.flavors_client.list_flavors_with_detail()
        self.assertIn(new_flavor['id'], map(lambda x: x['id'], flavors))

        # Remove flavor access from a tenant.
        remove_body = \
            self.client.remove_flavor_access(new_flavor['id'], self.tenant_id)
        self.assertNotIn(resp_body, remove_body)

        # The flavor is not present in list.
        flavors = self.flavors_client.list_flavors_with_detail()
        self.assertNotIn(new_flavor['id'], map(lambda x: x['id'], flavors))

 def test_volume_list_details_by_name(self):
     volume = self.volume_list[data_utils.rand_int_id(0, 2)]
     params = {self.name: volume[self.name]}
     fetched_vol = self.client.list_volumes(detail=True, params=params)
     self.assertEqual(1, len(fetched_vol), str(fetched_vol))
     self.assertEqual(fetched_vol[0][self.name],
                      volume[self.name])

 def test_volumes_list_by_availability_zone(self):
     volume = self.volume_list[data_utils.rand_int_id(0, 2)]
     zone = volume['availability_zone']
     params = {'availability_zone': zone}
     fetched_list = self.client.list_volumes(params=params)
     self._list_by_param_value_and_assert(params)
     self.assertVolumesIn(fetched_list, self.volume_list,
                          fields=self.VOLUME_FIELDS)

 def test_volumes_list_details_by_availability_zone(self):
     volume = self.volume_list[data_utils.rand_int_id(0, 2)]
     zone = volume['availability_zone']
     params = {'availability_zone': zone}
     fetched_list = self.client.list_volumes(detail=True, params=params)
     for volume in fetched_list:
         self.assertEqual(zone, volume['availability_zone'])
     self.assertVolumesIn(fetched_list, self.volume_list)

 def test_get_nonexistent_floating_ip_details(self):
     # Negative test:Should not be able to GET the details
     # of non-existent floating IP
     # Creating a non-existent floatingIP id
     if CONF.service_available.neutron:
         non_exist_id = str(uuid.uuid4())
     else:
         non_exist_id = data_utils.rand_int_id(start=999)
     self.assertRaises(lib_exc.NotFound,
                       self.client.show_floating_ip, non_exist_id)

    def test_list_container_contents_with_limit(self):
        # get container contents list using limit param
        container_name = self._create_container()
        object_name = self._create_object(container_name)

        params = {'limit': data_utils.rand_int_id(1, 10000)}
        resp, object_list = self.container_client.list_container_contents(
            container_name,
            params=params)
        self.assertHeaders(resp, 'Container', 'GET')
        self.assertEqual(object_name, object_list.strip('\n'))

 def test_get_queue_stats(self):
     # Retrieve random queue
     queue_name = self.queues[data_utils.rand_int_id(0,
                                                     len(self.queues) - 1)]
     # Get Queue Stats for a newly created Queue
     _, body = self.get_queue_stats(queue_name)
     msgs = body['messages']
     for element in ('free', 'claimed', 'total'):
         self.assertEqual(0, msgs[element])
     for element in ('oldest', 'newest'):
         self.assertNotIn(element, msgs)

    def test_is_public_string_variations(self):
        flavor_id_not_public = data_utils.rand_int_id(start=1000)
        flavor_name_not_public = data_utils.rand_name(self.flavor_name_prefix)
        flavor_id_public = data_utils.rand_int_id(start=1000)
        flavor_name_public = data_utils.rand_name(self.flavor_name_prefix)

        # Create a non public flavor
        flavor = self.client.create_flavor(flavor_name_not_public,
                                           self.ram, self.vcpus,
                                           self.disk,
                                           flavor_id_not_public,
                                           is_public="False")
        self.addCleanup(self.flavor_clean_up, flavor['id'])

        # Create a public flavor
        flavor = self.client.create_flavor(flavor_name_public,
                                           self.ram, self.vcpus,
                                           self.disk,
                                           flavor_id_public,
                                           is_public="True")
        self.addCleanup(self.flavor_clean_up, flavor['id'])

        def _flavor_lookup(flavors, flavor_name):
            for flavor in flavors:
                if flavor['name'] == flavor_name:
                    return flavor
            return None

        def _test_string_variations(variations, flavor_name):
            for string in variations:
                params = {'is_public': string}
                flavors = self.client.list_flavors_with_detail(params)
                flavor = _flavor_lookup(flavors, flavor_name)
                self.assertIsNotNone(flavor)

        _test_string_variations(['f', 'false', 'no', '0'],
                                flavor_name_not_public)

        _test_string_variations(['t', 'true', 'yes', '1'],
                                flavor_name_public)

 def test_create_security_group_rule_with_invalid_to_port(self):
     # Negative test: Creation of Security Group rule should FAIL
     # with invalid to_port
     # Creating a Security Group to add rule to it
     sg = self.create_security_group()
     # Adding rules to the created Security Group
     parent_group_id = sg['id']
     ip_protocol = 'tcp'
     from_port = 22
     to_port = data_utils.rand_int_id(start=65536)
     self.assertRaises(lib_exc.BadRequest,
                       self.client.create_security_group_rule,
                       parent_group_id, ip_protocol, from_port, to_port)

 def test_flavor_access_list_with_private_flavor(self):
     # Test to make sure that list flavor access on a newly created
     # private flavor will return an empty access list
     flavor_name = data_utils.rand_name(self.flavor_name_prefix)
     new_flavor_id = data_utils.rand_int_id(start=1000)
     new_flavor = self.client.create_flavor(flavor_name,
                                            self.ram, self.vcpus,
                                            self.disk,
                                            new_flavor_id,
                                            is_public='False')
     self.addCleanup(self.client.delete_flavor, new_flavor['id'])
     flavor_access = self.client.list_flavor_access(new_flavor_id)
     self.assertEqual(len(flavor_access), 0, str(flavor_access))

        def create_flavor_without_extra_specs():
            flavor_no_eph_disk_name = data_utils.rand_name("no_eph_flavor")
            flavor_no_eph_disk_id = data_utils.rand_int_id(start=1000)

            ram = 64
            vcpus = 1
            disk = 0

            # Create a flavor without extra specs
            flavor = self.flavor_client.create_flavor(flavor_no_eph_disk_name, ram, vcpus, disk, flavor_no_eph_disk_id)
            self.addCleanup(flavor_clean_up, flavor["id"])

            return flavor["id"]

 def test_flavor_access_list_with_public_flavor(self):
     # Test to list flavor access with exceptions by querying public flavor
     flavor_name = data_utils.rand_name(self.flavor_name_prefix)
     new_flavor_id = data_utils.rand_int_id(start=1000)
     new_flavor = self.client.create_flavor(flavor_name,
                                            self.ram, self.vcpus,
                                            self.disk,
                                            new_flavor_id,
                                            is_public='True')
     self.addCleanup(self.client.delete_flavor, new_flavor['id'])
     self.assertRaises(lib_exc.NotFound,
                       self.client.list_flavor_access,
                       new_flavor_id)

 def _generate_a_non_existent_security_group_id(self):
     security_group_id = []
     body = self.client.list_security_groups()
     for i in range(len(body)):
         security_group_id.append(body[i]['id'])
     # Generate a non-existent security group id
     while True:
         non_exist_id = data_utils.rand_int_id(start=999)
         if self.neutron_available:
             non_exist_id = data_utils.rand_uuid()
         if non_exist_id not in security_group_id:
             break
     return non_exist_id

    def populate_smoke_table(self, table_name, keycount, count_per_key):
        """
        Put [keycont*count_per_key] autogenerated items to the table.

        In result, [keycount] unique hash key values
        and [count_per_key] items for each has key value are generated.

        For example, to generate some number of items for the only hash key,
        set keycount=1 and count_per_key=needed_number_of_items.
        """
        new_items = []
        for _ in range(keycount):
            forum = 'forum%s' % data_utils.rand_int_id()
            for i in range(count_per_key):
                item = self.put_smoke_item(
                    table_name, forum=forum, subject='subject%s' % i,
                    message=data_utils.rand_name(self.table_prefix),
                    last_posted_by=data_utils.rand_uuid(),
                    replies=str(data_utils.rand_int_id())
                )
                new_items.append(item)
        return new_items

 def test_flavor_non_admin_add(self):
     # Test to add flavor access as a user without admin privileges.
     flavor_name = data_utils.rand_name(self.flavor_name_prefix)
     new_flavor_id = data_utils.rand_int_id(start=1000)
     new_flavor = self.client.create_flavor(flavor_name,
                                            self.ram, self.vcpus,
                                            self.disk,
                                            new_flavor_id,
                                            is_public='False')
     self.addCleanup(self.client.delete_flavor, new_flavor['id'])
     self.assertRaises(lib_exc.Forbidden,
                       self.flavors_client.add_flavor_access,
                       new_flavor['id'],
                       self.tenant_id)

    def test_create_flavor_using_string_ram(self):
        flavor_name = data_utils.rand_name(self.flavor_name_prefix)
        new_flavor_id = data_utils.rand_int_id(start=1000)

        ram = "1024"
        flavor = self.client.create_flavor(flavor_name,
                                           ram, self.vcpus,
                                           self.disk,
                                           new_flavor_id)
        self.addCleanup(self.flavor_clean_up, flavor['id'])
        self.assertEqual(flavor['name'], flavor_name)
        self.assertEqual(flavor['vcpus'], self.vcpus)
        self.assertEqual(flavor['disk'], self.disk)
        self.assertEqual(flavor['ram'], int(ram))
        self.assertEqual(int(flavor['id']), new_flavor_id)

    def resource_setup(cls):
        super(FlavorsExtraSpecsNegativeTestJSON, cls).resource_setup()

        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
        cls.flavor = cls.client.create_flavor(
            flavor_name, ram, vcpus, disk, cls.new_flavor_id, ephemeral=ephemeral, swap=swap, rxtx=rxtx
        )

    def test_remove_flavor_access_not_found(self):
        # Create a new flavor.
        flavor_name = data_utils.rand_name(self.flavor_name_prefix)
        new_flavor_id = data_utils.rand_int_id(start=1000)
        new_flavor = self.client.create_flavor(flavor_name,
                                               self.ram, self.vcpus,
                                               self.disk,
                                               new_flavor_id,
                                               is_public='False')
        self.addCleanup(self.client.delete_flavor, new_flavor['id'])

        # An exception should be raised when flavor access is not found
        self.assertRaises(lib_exc.NotFound,
                          self.client.remove_flavor_access,
                          new_flavor['id'],
                          str(uuid.uuid4()))