def test_update_consistency_group_v2_4(self):

        # Get consistency_group
        consistency_group = self.shares_v2_client.get_consistency_group(
            self.consistency_group['id'], version='2.4')
        self.assertEqual(self.cg_name, consistency_group["name"])
        self.assertEqual(self.cg_desc, consistency_group["description"])

        # Update consistency_group
        new_name = data_utils.rand_name("tempest-new-name")
        new_desc = data_utils.rand_name("tempest-new-description")
        updated = self.shares_v2_client.update_consistency_group(
            consistency_group["id"],
            name=new_name,
            description=new_desc,
            version='2.4'
        )
        self.assertEqual(new_name, updated["name"])
        self.assertEqual(new_desc, updated["description"])

        # Get consistency_group
        consistency_group = self.shares_v2_client.get_consistency_group(
            self.consistency_group['id'], version='2.4')
        self.assertEqual(new_name, consistency_group["name"])
        self.assertEqual(new_desc, consistency_group["description"])

    def test_check_simple_image_attributes(self):
        name = data_utils.rand_name('image')
        desc = data_utils.rand_name('desc for image')
        image_id, image_clean = self._create_image(name, desc)

        data = self.client.describe_image_attribute(
            ImageId=image_id, Attribute='kernel')
        self.assertIn('KernelId', data)

        data = self.client.describe_image_attribute(
            ImageId=image_id, Attribute='ramdisk')
        self.assertIn('RamdiskId', data)

        # description
        data = self.client.describe_image_attribute(
            ImageId=image_id, Attribute='description')
        self.assertIn('Description', data)
        self.assertIn('Value', data['Description'])
        self.assertEqual(desc, data['Description']['Value'])

        def _modify_description(**kwargs):
            self.client.modify_image_attribute(ImageId=image_id, **kwargs)
            data = self.client.describe_image_attribute(
                ImageId=image_id, Attribute='description')
            self.assertEqual(new_desc, data['Description']['Value'])

        new_desc = data_utils.rand_name('new desc')
        _modify_description(Attribute='description', Value=new_desc)
        _modify_description(Description={'Value': new_desc})

        data = self.client.deregister_image(ImageId=image_id)
        self.cancelResourceCleanUp(image_clean)

    def _check_create_cluster_template(self):
        ng_template_name = data_utils.rand_name('sahara-ng-template')
        ng_template = self.create_node_group_template(ng_template_name,
                                                      **self.worker_template)

        full_cluster_template = self.cluster_template.copy()
        full_cluster_template['node_groups'] = [
            {
                'name': 'master-node',
                'flavor_id': TEMPEST_CONF.compute.flavor_ref,
                'node_processes': ['namenode'],
                'count': 1
            },
            {
                'name': 'worker-node',
                'node_group_template_id': ng_template.id,
                'count': 3
            }
        ]

        template_name = data_utils.rand_name('sahara-cluster-template')

        # create cluster template
        resp_body = self.create_cluster_template(template_name,
                                                 **full_cluster_template)

        # check that template created successfully
        self.assertEqual(template_name, resp_body.name)
        self.assertDictContainsSubset(self.cluster_template,
                                      resp_body.__dict__)

        return resp_body.id, template_name

 def test_add_multiple_router_interfaces(self):
     network_name = data_utils.rand_name(self.__class__.__name__)
     network01 = self.networks_client.create_network(
         name=network_name)['network']
     self.addCleanup(self.networks_client.delete_network,
                     network01['id'])
     network_name = data_utils.rand_name(self.__class__.__name__)
     network02 = self.networks_client.create_network(
         name=network_name)['network']
     self.addCleanup(self.networks_client.delete_network,
                     network02['id'])
     subnet01 = self.create_subnet(network01)
     self.addCleanup(self.subnets_client.delete_subnet, subnet01['id'])
     sub02_cidr = self.cidr.next()
     subnet02 = self.create_subnet(network02, cidr=sub02_cidr)
     self.addCleanup(self.subnets_client.delete_subnet, subnet02['id'])
     router = self.create_router()
     self.addCleanup(self.delete_router, router)
     interface01 = self._add_router_interface_with_subnet_id(router['id'],
                                                             subnet01['id'])
     self._verify_router_interface(router['id'], subnet01['id'],
                                   interface01['port_id'])
     interface02 = self._add_router_interface_with_subnet_id(router['id'],
                                                             subnet02['id'])
     self._verify_router_interface(router['id'], subnet02['id'],
                                   interface02['port_id'])

    def test_group_type_create_list_show(self):
        # Create/list/show group type.
        name = data_utils.rand_name(self.__class__.__name__ + '-group-type')
        description = data_utils.rand_name("group-type-description")
        group_specs = {"consistent_group_snapshot_enabled": "<is> False"}
        params = {'name': name,
                  'description': description,
                  'group_specs': group_specs,
                  'is_public': True}
        body = self.create_group_type(**params)
        self.assertIn('name', body)
        err_msg = ("The created group_type %(var)s is not equal to the "
                   "requested %(var)s")
        self.assertEqual(name, body['name'], err_msg % {"var": "name"})
        self.assertEqual(description, body['description'],
                         err_msg % {"var": "description"})

        group_list = (
            self.admin_group_types_client.list_group_types()['group_types'])
        self.assertIsInstance(group_list, list)
        self.assertNotEmpty(group_list)

        fetched_group_type = self.admin_group_types_client.show_group_type(
            body['id'])['group_type']
        for key in params.keys():
            self.assertEqual(params[key], fetched_group_type[key],
                             '%s of the fetched group_type is different '
                             'from the created group_type' % key)

    def test_tag_security_group(self):
        cidr = '10.1.0.0/16'
        data = self.client.create_vpc(CidrBlock=cidr)
        vpc_id = data['Vpc']['VpcId']
        dv_clean = self.addResourceCleanUp(
            self.client.delete_vpc, VpcId=vpc_id)

        name = data_utils.rand_name('sgName')
        desc = data_utils.rand_name('sgDesc')
        data = self.client.create_security_group(VpcId=vpc_id,
                                                 GroupName=name,
                                                 Description=desc)
        group_id = data['GroupId']
        res_clean = self.addResourceCleanUp(self.client.delete_security_group,
                                            GroupId=group_id)
        time.sleep(2)

        def describe_func(*args, **kwargs):
            data = self.client.describe_security_groups(*args, **kwargs)
            self.assertEqual(1, len(data['SecurityGroups']))
            self.assertEqual(group_id,
                             data['SecurityGroups'][0]['GroupId'])

        self._test_tag_resource(group_id, 'security-group', describe_func)

        self.client.delete_security_group(GroupId=group_id)
        self.cancelResourceCleanUp(res_clean)

        self.client.delete_vpc(VpcId=vpc_id)
        self.cancelResourceCleanUp(dv_clean)
        self.get_vpc_waiter().wait_delete(vpc_id)

 def _create_dummy_user(self, add_clean_up=True):
     username = data_utils.rand_name('TestUser')
     password = data_utils.rand_name('password')
     email = data_utils.rand_name() + '@example.com'
     description = data_utils.rand_name('description')
     raw_output = self.openstack(
         'user create '
         '--domain %(domain)s '
         '--project %(project)s '
         '--project-domain %(project_domain)s '
         '--password %(password)s '
         '--email %(email)s '
         '--description %(description)s '
         '--enable '
         '%(name)s' % {'domain': self.domain_name,
                       'project': self.project_name,
                       'project_domain': self.domain_name,
                       'email': email,
                       'password': password,
                       'description': description,
                       'name': username})
     if add_clean_up:
         self.addCleanup(
             self.openstack,
             'user delete %s' % self.parse_show_as_object(raw_output)['id'])
     items = self.parse_show(raw_output)
     self.assert_show_fields(items, self.USER_FIELDS)
     return username

    def test_create_update_and_delete_domain_config_groups_and_opts(self):
        domain, _ = self._create_domain_and_config(self.custom_config)

        # Check that updating configuration groups work.
        new_driver = data_utils.rand_name('driver')
        new_limit = data_utils.rand_int_id(0, 100)
        new_group_config = {'identity': {'driver': new_driver,
                                         'list_limit': new_limit}}

        updated_config = self.client.update_domain_group_config(
            domain['id'], 'identity', **new_group_config)['config']

        self.assertEqual(new_driver, updated_config['identity']['driver'])
        self.assertEqual(new_limit, updated_config['identity']['list_limit'])

        # Check that updating individual configuration group options work.
        new_driver = data_utils.rand_name('driver')

        updated_config = self.client.update_domain_group_option_config(
            domain['id'], 'identity', 'driver', driver=new_driver)['config']

        self.assertEqual(new_driver, updated_config['identity']['driver'])

        # Check that deleting individual configuration group options work.
        self.client.delete_domain_group_option_config(
            domain['id'], 'identity', 'driver')
        self.assertRaises(lib_exc.NotFound,
                          self.client.show_domain_group_option_config,
                          domain['id'], 'identity', 'driver')

        # Check that deleting configuration groups work.
        self.client.delete_domain_group_config(domain['id'], 'identity')
        self.assertRaises(lib_exc.NotFound,
                          self.client.show_domain_group_config,
                          domain['id'], 'identity')

def cluster_template_data(**kwargs):
    """Generates random cluster_template data

    Keypair and image id cannot be random for the cluster_template to be valid
    due to validations for the presence of keypair and image id prior to
    cluster_template creation.

    :param keypair_id: keypair name
    :param image_id: image id or name
    :returns: ClusterTemplateEntity with generated data
    """

    data = {
        "name": data_utils.rand_name('cluster'),
        "coe": "swarm",
        "tls_disabled": False,
        "network_driver": None,
        "volume_driver": None,
        "docker_volume_size": 3,
        "labels": {},
        "public": False,
        "dns_nameserver": "8.8.8.8",
        "flavor_id": data_utils.rand_name('cluster'),
        "master_flavor_id": data_utils.rand_name('cluster'),
        "external_network_id": config.Config.nic_id,
        "keypair_id": data_utils.rand_name('cluster'),
        "image_id": data_utils.rand_name('cluster')
    }

    data.update(kwargs)
    model = cluster_template_model.ClusterTemplateEntity.from_dict(data)

    return model

 def resource_setup(cls):
     super(EndPointsTestJSON, cls).resource_setup()
     cls.service_ids = list()
     s_name = data_utils.rand_name('service')
     s_type = data_utils.rand_name('type')
     s_description = data_utils.rand_name('description')
     service_data = cls.services_client.create_service(
         name=s_name, type=s_type,
         description=s_description)['OS-KSADM:service']
     cls.service_id = service_data['id']
     cls.service_ids.append(cls.service_id)
     # Create endpoints so as to use for LIST and GET test cases
     cls.setup_endpoints = list()
     for _ in range(2):
         region = data_utils.rand_name('region')
         url = data_utils.rand_url()
         endpoint = cls.endpoints_client.create_endpoint(
             service_id=cls.service_id,
             region=region,
             publicurl=url,
             adminurl=url,
             internalurl=url)['endpoint']
         # list_endpoints() will return 'enabled' field
         endpoint['enabled'] = True
         cls.setup_endpoints.append(endpoint)

    def _get_updated_quotas(self):
        # Verify that GET shows the updated quota set of project
        project_name = data_utils.rand_name('cpu_quota_project')
        project_desc = project_name + '-desc'
        project = identity.identity_utils(self.os_admin).create_project(
            name=project_name, description=project_desc)
        project_id = project['id']
        self.addCleanup(identity.identity_utils(self.os_admin).delete_project,
                        project_id)

        self.adm_client.update_quota_set(project_id, ram='5120')
        # Call show_quota_set with detail=true to cover the
        # get_quota_set_details response schema for microversion tests
        quota_set = self.adm_client.show_quota_set(
            project_id, detail=True)['quota_set']
        self.assertEqual(5120, quota_set['ram']['limit'])

        # Verify that GET shows the updated quota set of user
        user_name = data_utils.rand_name('cpu_quota_user')
        password = data_utils.rand_password()
        email = user_name + '@testmail.tm'
        user = identity.identity_utils(self.os_admin).create_user(
            username=user_name, password=password, project=project,
            email=email)
        user_id = user['id']
        self.addCleanup(identity.identity_utils(self.os_admin).delete_user,
                        user_id)

        self.adm_client.update_quota_set(project_id,
                                         user_id=user_id,
                                         ram='2048')
        quota_set = self.adm_client.show_quota_set(
            project_id, user_id=user_id)['quota_set']
        self.assertEqual(2048, quota_set['ram'])

 def test_to_verify_communication_between_two_vms_in_diff_network(self):
     net_id = self.network['id']
     name = data_utils.rand_name('server-smoke')
     group_create_body = self._create_custom_security_group()
     serverid = self._create_server_with_sec_group(
         name, net_id, group_create_body['security_group']['id'])
     self.assertTrue(self.verify_portgroup(self.network['id'], serverid))
     deviceport = self.ports_client.list_ports(device_id=serverid)
     body = self._associate_floating_ips(
         port_id=deviceport['ports'][0]['id'])
     fip1 = body['floatingip']['floating_ip_address']
     network2 = self.create_network()
     sub_cidr = netaddr.IPNetwork(CONF.network.project_network_cidr).next()
     subnet2 = self.create_subnet(network2, cidr=sub_cidr)
     router2 = self.create_router(data_utils.rand_name('router2-'),
                                  external_network_id=self.ext_net_id,
                                  admin_state_up="true")
     self.create_router_interface(router2['id'], subnet2['id'])
     serverid2 = self._create_server_with_sec_group(
         name, network2['id'], group_create_body['security_group']['id'])
     deviceport2 = self.ports_client.list_ports(device_id=serverid2)
     body = self._associate_floating_ips(
         port_id=deviceport2['ports'][0]['id'])
     fip2 = body['floatingip']['floating_ip_address']
     self.assertTrue(self.verify_portgroup(self.network['id'], serverid))
     self.assertTrue(self.verify_portgroup(network2['id'], serverid2))
     self.assertTrue(self._check_remote_connectivity(fip1, fip2))

 def test_creation_of_VM_attach_to_user_created_port(self):
     group_create_body = self._create_custom_security_group()
     network = self.create_network()
     subnet = self.create_subnet(network)
     router = self.create_router(data_utils.rand_name('router-'),
                                 external_network_id=self.ext_net_id,
                                 admin_state_up="true")
     self.create_router_interface(router['id'], subnet['id'])
     post_body = {
         "name": data_utils.rand_name('port-'),
         "security_groups": [group_create_body['security_group']['id']],
         "network_id": network['id'],
         "admin_state_up": True}
     port = self.ports_client.create_port(**post_body)
     self.addCleanup(self.ports_client.delete_port, port['port']['id'])
     name = data_utils.rand_name('server-smoke')
     group_create_body, _ = self._create_security_group()
     serverid = self._create_server_user_created_port(
         name, port['port']['id'])
     self.assertTrue(self.verify_portgroup(network['id'], serverid))
     deviceport = self.ports_client.list_ports(device_id=serverid)
     body = self._associate_floating_ips(
         port_id=deviceport['ports'][0]['id'])
     floatingiptoreach = body['floatingip']['floating_ip_address']
     self._check_public_network_connectivity(floatingiptoreach)

 def test_create_update_get_delete_region(self):
     # Create region
     r_description = data_utils.rand_name('description')
     region = self.client.create_region(
         description=r_description,
         parent_region_id=self.setup_regions[0]['id'])['region']
     # This test will delete the region as part of the validation
     # procedure, so it needs a different cleanup method that
     # would be useful in case the tests fails at any point before
     # reaching the deletion part.
     self.addCleanup(test_utils.call_and_ignore_notfound_exc,
                     self.client.delete_region, region['id'])
     self.assertEqual(r_description, region['description'])
     self.assertEqual(self.setup_regions[0]['id'],
                      region['parent_region_id'])
     # Update region with new description and parent ID
     r_alt_description = data_utils.rand_name('description')
     region = self.client.update_region(
         region['id'],
         description=r_alt_description,
         parent_region_id=self.setup_regions[1]['id'])['region']
     self.assertEqual(r_alt_description, region['description'])
     self.assertEqual(self.setup_regions[1]['id'],
                      region['parent_region_id'])
     # Get the details of region
     region = self.client.show_region(region['id'])['region']
     self.assertEqual(r_alt_description, region['description'])
     self.assertEqual(self.setup_regions[1]['id'],
                      region['parent_region_id'])
     # Delete the region
     self.client.delete_region(region['id'])
     body = self.client.list_regions()['regions']
     regions_list = [r['id'] for r in body]
     self.assertNotIn(region['id'], regions_list)

 def resource_setup(cls):
     super(InheritsV3TestJSON, cls).resource_setup()
     u_name = data_utils.rand_name('user-')
     u_desc = '%s description' % u_name
     u_email = '%[email protected]' % u_name
     u_password = data_utils.rand_password()
     cls.domain = cls.create_domain()
     cls.project = cls.projects_client.create_project(
         data_utils.rand_name('project-'),
         description=data_utils.rand_name('project-desc-'),
         domain_id=cls.domain['id'])['project']
     cls.addClassResourceCleanup(cls.projects_client.delete_project,
                                 cls.project['id'])
     cls.group = cls.groups_client.create_group(
         name=data_utils.rand_name('group-'), project_id=cls.project['id'],
         domain_id=cls.domain['id'])['group']
     cls.addClassResourceCleanup(cls.groups_client.delete_group,
                                 cls.group['id'])
     if not CONF.identity_feature_enabled.immutable_user_source:
         cls.user = cls.users_client.create_user(
             name=u_name,
             description=u_desc,
             password=u_password,
             email=u_email,
             project_id=cls.project['id'],
             domain_id=cls.domain['id']
         )['user']
         cls.addClassResourceCleanup(cls.users_client.delete_user,
                                     cls.user['id'])

    def test_create_list_update_show_delete_security_group(self):
        group_create_body, _ = self._create_security_group()

        # List security groups and verify if created group is there in response
        list_body = self.security_groups_client.list_security_groups()
        secgroup_list = list()
        for secgroup in list_body['security_groups']:
            secgroup_list.append(secgroup['id'])
        self.assertIn(group_create_body['security_group']['id'], secgroup_list)
        # Update the security group
        new_name = data_utils.rand_name('security-')
        new_description = data_utils.rand_name('security-description')
        update_body = self.security_groups_client.update_security_group(
            group_create_body['security_group']['id'],
            name=new_name,
            description=new_description)
        # Verify if security group is updated
        self.assertEqual(update_body['security_group']['name'], new_name)
        self.assertEqual(update_body['security_group']['description'],
                         new_description)
        # Show details of the updated security group
        show_body = self.security_groups_client.show_security_group(
            group_create_body['security_group']['id'])
        self.assertEqual(show_body['security_group']['name'], new_name)
        self.assertEqual(show_body['security_group']['description'],
                         new_description)

 def test_application_credential_create_with_options(self):
     name = data_utils.rand_name('name')
     secret = data_utils.rand_name('secret')
     description = data_utils.rand_name('description')
     tomorrow = (datetime.datetime.utcnow() +
                 datetime.timedelta(days=1)).strftime('%Y-%m-%dT%H:%M:%S%z')
     role1, role2 = self._create_role_assignments()
     raw_output = self.openstack('application credential create %(name)s'
                                 ' --secret %(secret)s'
                                 ' --description %(description)s'
                                 ' --expiration %(tomorrow)s'
                                 ' --role %(role1)s'
                                 ' --role %(role2)s'
                                 ' --unrestricted'
                                 % {'name': name,
                                    'secret': secret,
                                    'description': description,
                                    'tomorrow': tomorrow,
                                    'role1': role1,
                                    'role2': role2})
     self.addCleanup(
         self.openstack,
         'application credential delete %(name)s' % {'name': name})
     items = self.parse_show(raw_output)
     self.assert_show_fields(items, self.APPLICATION_CREDENTIAL_FIELDS)

    def test_copy_object_in_same_container(self):
        # create source object
        src_object_name = data_utils.rand_name(name='SrcObject')
        src_data = data_utils.random_bytes(size=len(src_object_name) * 2)
        resp, _ = self.object_client.create_object(self.container_name,
                                                   src_object_name,
                                                   src_data)
        # create destination object
        dst_object_name = data_utils.rand_name(name='DstObject')
        dst_data = data_utils.random_bytes(size=len(dst_object_name) * 3)
        resp, _ = self.object_client.create_object(self.container_name,
                                                   dst_object_name,
                                                   dst_data)
        # copy source object to destination
        headers = {}
        headers['X-Copy-From'] = "%s/%s" % (str(self.container_name),
                                            str(src_object_name))
        resp, body = self.object_client.create_object(self.container_name,
                                                      dst_object_name,
                                                      data=None,
                                                      headers=headers)
        self.assertHeaders(resp, 'Object', 'PUT')

        # check data
        resp, body = self.object_client.get_object(self.container_name,
                                                   dst_object_name)
        self.assertEqual(body, src_data)

    def setUpClass(cls):
        # prepare v3 env
        os.environ['OS_IDENTITY_API_VERSION'] = '3'
        auth_url = os.environ.get('OS_AUTH_URL')
        auth_url = auth_url.replace('v2.0', 'v3')
        os.environ['OS_AUTH_URL'] = auth_url

        # create dummy domain
        cls.domain_name = data_utils.rand_name('TestDomain')
        cls.domain_description = data_utils.rand_name('description')
        cls.openstack(
            'domain create '
            '--description %(description)s '
            '--enable '
            '%(name)s' % {'description': cls.domain_description,
                          'name': cls.domain_name})

        # create dummy project
        cls.project_name = data_utils.rand_name('TestProject')
        cls.project_description = data_utils.rand_name('description')
        cls.openstack(
            'project create '
            '--domain %(domain)s '
            '--description %(description)s '
            '--enable '
            '%(name)s' % {'domain': cls.domain_name,
                          'description': cls.project_description,
                          'name': cls.project_name})

def baymodel_data(**kwargs):
    """Generates random baymodel data

    Keypair and image id cannot be random for the baymodel to be valid due to
    validations for the presence of keypair and image id prior to baymodel
    creation.

    :param keypair_id: keypair name
    :param image_id: image id or name
    :returns: BayModelEntity with generated data
    """

    data = {
        "name": data_utils.rand_name('bay'),
        "coe": "swarm-mode",
        "tls_disabled": False,
        "network_driver": None,
        "volume_driver": None,
        "labels": {},
        "public": False,
        "dns_nameserver": "8.8.8.8",
        "flavor_id": data_utils.rand_name('bay'),
        "master_flavor_id": data_utils.rand_name('bay'),
        "external_network_id": config.Config.nic_id,
        "keypair_id": data_utils.rand_name('bay'),
        "image_id": data_utils.rand_name('bay')
    }

    data.update(kwargs)
    model = baymodel_model.BayModelEntity.from_dict(data)

    return model