def test_create_object_with_x_fresh_metadata(self):
        # create object with x_fresh_metadata
        object_name_base = data_utils.rand_name(name='TestObject')
        data = data_utils.arbitrary_string()
        metadata_1 = {'X-Object-Meta-test-meta': 'Meta'}
        self.object_client.create_object(self.container_name,
                                         object_name_base,
                                         data,
                                         metadata=metadata_1)
        object_name = data_utils.rand_name(name='TestObject')
        metadata_2 = {'X-Copy-From': '%s/%s' % (self.container_name,
                                                object_name_base),
                      'X-Fresh-Metadata': 'true'}
        resp, _ = self.object_client.create_object(
            self.container_name,
            object_name,
            '',
            metadata=metadata_2)
        self.assertHeaders(resp, 'Object', 'PUT')

        resp, body = self.object_client.get_object(self.container_name,
                                                   object_name)
        #Bug = 1417489
        #self.assertNotIn('x-object-meta-test-meta', resp)
        self.assertEqual(data, body)

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

        # List security groups and verify if created group is there in response
        list_body = self.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.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.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 resource_setup(cls):
     super(LoadBalancerTestJSON, cls).resource_setup()
     _params = cls.manager.default_params_with_timeout_values.copy()
     for p in _params.keys():
         if p in ['service', 'region', 'endpoint_type']:
             _params.pop(p)
     cls.lbv1_client = LBV1C.get_client(cls.manager)
     cls.network = cls.create_network()
     cls.name = cls.network['name']
     cls.subnet = cls.create_subnet(cls.network)
     cls.ext_net_id = CONF.network.public_network_id
     cls.router = cls.create_router(data_utils.rand_name('router-'),
                                    admin_state_up=True,
                                    external_network_id=cls.ext_net_id,
                                    router_type='exclusive')
     cls.create_router_interface(cls.router['id'], cls.subnet['id'])
     pool_name = data_utils.rand_name('pool-')
     vip_name = data_utils.rand_name('vip-')
     cls.pool = cls.lbv1_client.create_pool(
         pool_name, "ROUND_ROBIN", "HTTP", cls.subnet['id'])['pool']
     cls.vip = cls.lbv1_client.create_vip(cls.pool['id'],
                                          subnet_id=cls.subnet['id'],
                                          name=vip_name,
                                          protocol="HTTP",
                                          protocol_port=80)['vip']
     cls.member = cls.lbv1_client.create_member(
         80, cls.pool['id'], cls._ip_version)['member']
     cls.member_address = ("10.0.9.47" if cls._ip_version == 4
                           else "2015::beef")
     cls.health_monitor = cls.lbv1_client.create_health_monitor(
         delay=4, max_retries=3, type="TCP", timeout=1)['health_monitor']

 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 create_standard_security_group(self):
        name = data_utils.rand_name('sgName')
        desc = data_utils.rand_name('sgDesc')
        kwargs = {'GroupName': name, 'Description': desc}
        resp, data = self.vpc_client.CreateSecurityGroup(*[], **kwargs)
        self.assertEqual(200, resp.status_code, base.EC2ErrorConverter(data))
        self.addResourceCleanUp(self.vpc_client.DeleteSecurityGroup,
                                GroupName=name)
        time.sleep(2)

        kwargs = {
            'GroupName': name,
            'IpPermissions': [{
                'IpProtocol': 'icmp',
                'FromPort': -1,
                'ToPort': -1,
                'IpRanges': [{
                    'CidrIp': '0.0.0.0/0'
                }],
            }, {
                'IpProtocol': 'tcp',
                'FromPort': 22,
                'ToPort': 22,
                'IpRanges': [{
                    'CidrIp': '0.0.0.0/0'
                }],
            }]
        }
        resp, data = self.vpc_client.AuthorizeSecurityGroupIngress(*[], **kwargs)
        self.assertEqual(200, resp.status_code, base.EC2ErrorConverter(data))

        return name

 def test_create_update_delete_policy(self):
     # Test to update policy
     blob = data_utils.rand_name('BlobName')
     policy_type = data_utils.rand_name('PolicyType')
     policy = self.policy_client.create_policy(blob, policy_type)
     self.addCleanup(self._delete_policy, policy['id'])
     self.assertIn('id', policy)
     self.assertIn('type', policy)
     self.assertIn('blob', policy)
     self.assertIsNotNone(policy['id'])
     self.assertEqual(blob, policy['blob'])
     self.assertEqual(policy_type, policy['type'])
     # Update policy
     update_type = data_utils.rand_name('UpdatedPolicyType')
     data = self.policy_client.update_policy(
         policy['id'], type=update_type)
     self.assertIn('type', data)
     # Assertion for updated value with fetched value
     fetched_policy = self.policy_client.get_policy(policy['id'])
     self.assertIn('id', fetched_policy)
     self.assertIn('blob', fetched_policy)
     self.assertIn('type', fetched_policy)
     self.assertEqual(fetched_policy['id'], policy['id'])
     self.assertEqual(fetched_policy['blob'], policy['blob'])
     self.assertEqual(update_type, fetched_policy['type'])

    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 test_associate_user_to_project(self):
        # Associate a user to a project
        # Create a Project
        p_name = data_utils.rand_name('project')
        project = self.client.create_project(p_name)
        self.data.projects.append(project)

        # Create a User
        u_name = data_utils.rand_name('user')
        u_desc = u_name + 'description'
        u_email = u_name + '@testmail.tm'
        u_password = data_utils.rand_name('pass')
        user = self.client.create_user(
            u_name, description=u_desc, password=u_password,
            email=u_email, project_id=project['id'])
        # Delete the User at the end of this method
        self.addCleanup(self.client.delete_user, user['id'])

        # Get User To validate the user details
        new_user_get = self.client.get_user(user['id'])
        # Assert response body of GET
        self.assertEqual(u_name, new_user_get['name'])
        self.assertEqual(u_desc, new_user_get['description'])
        self.assertEqual(project['id'],
                         new_user_get['project_id'])
        self.assertEqual(u_email, new_user_get['email'])

    def test_update_image(self):
        # Updates an image by image_id

        # Create image
        image_name = data_utils.rand_name('image')
        body = self.client.create_image(name=image_name,
                                        container_format='bare',
                                        disk_format='iso',
                                        visibility='private')
        self.addCleanup(self.client.delete_image, body['id'])
        self.assertEqual('queued', body['status'])
        image_id = body['id']

        # Now try uploading an image file
        image_file = moves.cStringIO(data_utils.random_bytes())
        self.client.store_image(image_id, image_file)

        # Update Image
        new_image_name = data_utils.rand_name('new-image')
        body = self.client.update_image(image_id, [
            dict(replace='/name', value=new_image_name)])

        # Verifying updating

        body = self.client.get_image(image_id)
        self.assertEqual(image_id, body['id'])
        self.assertEqual(new_image_name, body['name'])

 def test_create_show_list_delete_l2gateway_connection(self):
     # Create a network
     name = data_utils.rand_name('network')
     net_body = self.admin_client.create_network(name=name)
     net_id = net_body['network']['id']
     self.addCleanup(self.admin_client.delete_network, net_id)
     # Create an L2Gateway
     gw_name = data_utils.rand_name('l2gw')
     devices = base_l2gw.get_l2gw_body(CONF.network.l2gw_switch)["devices"]
     l2_gw_body = self.admin_client.create_l2_gateway(
         name=gw_name, devices=devices)
     l2_gw_id = l2_gw_body['l2_gateway']['id']
     self.addCleanup(self.admin_client.delete_l2_gateway, l2_gw_id)
     # Create an L2Gateway Connection
     l2_gw_conn_body = self.admin_client.create_l2_gateway_connection(
         l2_gateway_id=l2_gw_id, network_id=net_id)
     l2_gw_conn_id = l2_gw_conn_body['l2_gateway_connection']['id']
     self.addCleanup(self.admin_client.delete_l2_gateway_connection,
                     l2_gw_conn_id)
     # Show details of created L2 Gateway connection
     show_body = self.admin_client.show_l2_gateway_connection(l2_gw_conn_id)
     l2_gw_conn = show_body['l2_gateway_connection']
     self.assertEqual(net_id, l2_gw_conn['network_id'])
     self.assertEqual(l2_gw_id, l2_gw_conn['l2_gateway_id'])
     # List L2Gateway Connections
     self.admin_client.list_l2_gateway_connections()

    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_create_l2gw_conn_with_segid_when_l2gw_created_without_segid(self):
     # Create an L2Gateway
     gw_name = data_utils.rand_name('l2gw')
     devices = base_l2gw.get_l2gw_body(CONF.network.l2gw_switch)["devices"]
     if devices[0]['interfaces'][0]['segmentation_id']:
         seg_id = devices[0]['interfaces'][0]['segmentation_id'][0]
         devices[0]['interfaces'][0].pop('segmentation_id')
     body = self.admin_client.create_l2_gateway(
         name=gw_name, devices=devices)
     l2_gateway = body['l2_gateway']
     l2_gw_id = l2_gateway['id']
     self.addCleanup(self.admin_client.delete_l2_gateway, l2_gw_id)
     # Create a network
     name = data_utils.rand_name('network')
     net_body = self.admin_client.create_network(name=name)
     net_id = net_body['network']['id']
     self.addCleanup(self.admin_client.delete_network, net_id)
     # Create an L2Gateway Connection
     l2_gw_conn_body = self.admin_client.create_l2_gateway_connection(
         l2_gateway_id=l2_gw_id, network_id=net_id,
         segmentation_id=seg_id)
     l2_gw_conn_id = l2_gw_conn_body['l2_gateway_connection']['id']
     l2_gw_seg_id = l2_gw_conn_body['l2_gateway_connection'][
         'segmentation_id']
     self.addCleanup(self.admin_client.delete_l2_gateway_connection,
                     l2_gw_conn_id)
     # Show details of created L2 Gateway connection
     show_body = self.admin_client.show_l2_gateway_connection(
         l2_gw_conn_id)
     l2_gw_conn = show_body['l2_gateway_connection']
     self.assertEqual(net_id, l2_gw_conn['network_id'])
     self.assertEqual(l2_gw_id, l2_gw_conn['l2_gateway_id'])
     self.assertEqual(str(l2_gw_seg_id),
                      str(l2_gw_conn['segmentation_id']))

    def _create_creds(self, suffix="", admin=False, roles=None):
        """Create random credentials under the following schema.

        If the name contains a '.' is the full class path of something, and
        we don't really care. If it isn't, it's probably a meaningful name,
        so use it.

        For logging purposes, -user and -tenant are long and redundant,
        don't use them. The user# will be sufficient to figure it out.
        """
        if '.' in self.name:
            root = ""
        else:
            root = self.name

        tenant_name = data_utils.rand_name(root) + suffix
        tenant_desc = tenant_name + "-desc"
        tenant = self._create_tenant(name=tenant_name,
                                     description=tenant_desc)

        username = data_utils.rand_name(root) + suffix
        email = data_utils.rand_name(root) + suffix + "@example.com"
        user = self._create_user(username, self.password,
                                 tenant, email)
        if admin:
            self._assign_user_role(tenant, user, CONF.identity.admin_role)
        # Add roles specified in config file
        for conf_role in CONF.auth.tempest_roles:
            self._assign_user_role(tenant, user, conf_role)
        # Add roles requested by caller
        if roles:
            for role in roles:
                self._assign_user_role(tenant, user, role)
        return self._get_credentials(user, tenant)

    def test_copy_object_2d_way(self):
        # create source object
        src_object_name = data_utils.rand_name(name='SrcObject')
        src_data = data_utils.arbitrary_string(size=len(src_object_name) * 2,
                                               base_text=src_object_name)
        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.arbitrary_string(size=len(dst_object_name) * 3,
                                               base_text=dst_object_name)
        resp, _ = self.object_client.create_object(self.container_name,
                                                   dst_object_name, dst_data)
        # copy source object to destination
        resp, _ = self.object_client.copy_object_2d_way(self.container_name,
                                                        src_object_name,
                                                        dst_object_name)
        self.assertHeaders(resp, 'Object', 'COPY')
        #Bug 1417469
        #self.assertEqual(
        #    resp['x-copied-from'],
        #    self.container_name + "/" + src_object_name)

        # check data
        self._check_copied_obj(dst_object_name, src_data)

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",
        "tls_disabled": False,
        "network_driver": None,
        "volume_driver": None,
        "docker_volume_size": 3,
        "labels": {},
        "fixed_network": "192.168.0.0/24",
        "dns_nameserver": "8.8.8.8",
        "flavor_id": data_utils.rand_name('bay'),
        "master_flavor_id": data_utils.rand_name('bay'),
        "external_network_id": str(data_utils.rand_uuid()),
        "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

    def test_tenant_update_desc(self):
        # Update description attribute of a tenant
        t_name = data_utils.rand_name(name='tenant')
        t_desc = data_utils.rand_name(name='desc')
        body = self.client.create_tenant(t_name, description=t_desc)
        tenant = body
        self.data.tenants.append(tenant)

        t_id = body['id']
        resp1_desc = body['description']

        t_desc2 = data_utils.rand_name(name='desc2')
        body = self.client.update_tenant(t_id, description=t_desc2)
        resp2_desc = body['description']
        self.assertNotEqual(resp1_desc, resp2_desc)

        body = self.client.get_tenant(t_id)
        resp3_desc = body['description']

        self.assertNotEqual(resp1_desc, resp3_desc)
        self.assertEqual(t_desc, resp1_desc)
        self.assertEqual(resp2_desc, resp3_desc)

        self.client.delete_tenant(t_id)
        self.data.tenants.remove(tenant)

    def test_list_users_for_tenant(self):
        # Return a list of all users for a tenant
        self.data.setup_test_tenant()
        user_ids = list()
        fetched_user_ids = list()
        alt_tenant_user1 = data_utils.rand_name('tenant_user1')
        user1 = self.client.create_user(alt_tenant_user1, 'password1',
                                        self.data.tenant['id'],
                                        '[email protected]')
        user_ids.append(user1['id'])
        self.data.users.append(user1)

        alt_tenant_user2 = data_utils.rand_name('tenant_user2')
        user2 = self.client.create_user(alt_tenant_user2, 'password2',
                                        self.data.tenant['id'],
                                        '[email protected]')
        user_ids.append(user2['id'])
        self.data.users.append(user2)
        # List of users for the respective tenant ID
        body = self.client.list_users_for_tenant(self.data.tenant['id'])
        for i in body:
            fetched_user_ids.append(i['id'])
        # verifying the user Id in the list
        missing_users =\
            [user for user in user_ids if user not in fetched_user_ids]
        self.assertEqual(0, len(missing_users),
                         "Failed to find user %s in fetched list" %
                         ', '.join(m_user for m_user in missing_users))

 def test_create_get_delete_token(self):
     # get a token by username and password
     user_name = data_utils.rand_name(name='user')
     user_password = data_utils.rand_name(name='pass')
     # first:create a tenant
     tenant_name = data_utils.rand_name(name='tenant')
     tenant = self.client.create_tenant(tenant_name)
     self.data.tenants.append(tenant)
     # second:create a user
     user = self.client.create_user(user_name, user_password,
                                    tenant['id'], '')
     self.data.users.append(user)
     # then get a token for the user
     body = self.token_client.auth(user_name,
                                   user_password,
                                   tenant['name'])
     self.assertEqual(body['token']['tenant']['name'],
                      tenant['name'])
     # Perform GET Token
     token_id = body['token']['id']
     token_details = self.client.get_token(token_id)
     self.assertEqual(token_id, token_details['token']['id'])
     self.assertEqual(user['id'], token_details['user']['id'])
     self.assertEqual(user_name, token_details['user']['name'])
     self.assertEqual(tenant['name'],
                      token_details['token']['tenant']['name'])
     # then delete the token
     self.client.delete_token(token_id)

    def create_standard_security_group(self):
        name = data_utils.rand_name('sgName')
        desc = data_utils.rand_name('sgDesc')
        kwargs = {'GroupName': name, 'Description': desc}
        self.client.create_security_group(*[], **kwargs)
        self.addResourceCleanUp(self.client.delete_security_group,
                                GroupName=name)
        time.sleep(2)

        kwargs = {
            'GroupName': name,
            'IpPermissions': [{
                'IpProtocol': 'icmp',
                'FromPort': -1,
                'ToPort': -1,
                'IpRanges': [{
                    'CidrIp': '0.0.0.0/0'
                }],
            }, {
                'IpProtocol': 'tcp',
                'FromPort': 22,
                'ToPort': 22,
                'IpRanges': [{
                    'CidrIp': '0.0.0.0/0'
                }],
            }]
        }
        self.client.authorize_security_group_ingress(*[], **kwargs)

        return name

 def test_VMs_when_hosted_on_different_network(self):
     network_name1 = data_utils.rand_name('first-network-')
     post_body = {'name': network_name1}
     body = self.admin_client.create_network(**post_body)
     network1 = body['network']
     subnet1 = self.create_subnet(network1, client=self.admin_client)
     group_create_body1 = self.admin_client.create_security_group()
     network_name2 = data_utils.rand_name('second-network-')
     post_body = {'name': network_name2}
     body = self.admin_client.create_network(**post_body)
     network2 = body['network']
     sub_cidr = netaddr.IPNetwork(CONF.network.tenant_network_cidr).next()
     subnet2 = self.create_subnet(network2, client=self.admin_client,
                                  cidr=sub_cidr)
     group_create_body2 = self.admin_client.create_security_group()
     name = data_utils.rand_name('router-')
     router = self.admin_client.create_router(
         name, external_gateway_info={
             "network_id": CONF.network.public_network_id},
         admin_state_up=True)
     self.addCleanup(self.admin_client.delete_router,
                     router['router']['id'])
     floating_ip_admin1, server_ip1 = \
         self._attach_subnet_to_router_fip(network1, subnet1,
                                           group_create_body1, router)
     floating_ip_admin2, server_ip2 = \
         self._attach_subnet_to_router_fip(network2, subnet2,
                                           group_create_body2, router)
     self.assertTrue(self._check_remote_connectivity(floating_ip_admin1,
                                                     server_ip2))
     self.assertTrue(self._check_remote_connectivity(floating_ip_admin2,
                                                     server_ip1))