def test_update_user_password(self):
     # Creating User to check password updation
     u_name = data_utils.rand_name("user")
     original_password = data_utils.rand_password()
     user = self.client.create_user(u_name, password=original_password)["user"]
     # Delete the User at the end all test methods
     self.addCleanup(self.client.delete_user, user["id"])
     # Update user with new password
     new_password = data_utils.rand_password()
     self.client.update_user_password(user["id"], password=new_password, original_password=original_password)
     # TODO(lbragstad): Sleeping after the response status has been checked
     # and the body loaded as JSON allows requests to fail-fast. The sleep
     # is necessary because keystone will err on the side of security and
     # invalidate tokens within a small margin of error (within the same
     # wall clock second) after a revocation event is issued (such as a
     # password change). Remove this once keystone and Fernet support
     # sub-second precision, see bug 1517697 for more details.
     time.sleep(1)
     resp = self.token.auth(user_id=user["id"], password=new_password).response
     subject_token = resp["x-subject-token"]
     # Perform GET Token to verify and confirm password is updated
     token_details = self.client.show_token(subject_token)["token"]
     self.assertEqual(resp["x-subject-token"], subject_token)
     self.assertEqual(token_details["user"]["id"], user["id"])
     self.assertEqual(token_details["user"]["name"], u_name)

 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()
     password1 = data_utils.rand_password()
     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']
     user_ids.append(user1['id'])
     self.data.users.append(user1)
     password2 = data_utils.rand_password()
     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']
     user_ids.append(user2['id'])
     self.data.users.append(user2)
     # List of users for the respective tenant ID
     body = (self.tenants_client.list_tenant_users(self.data.tenant['id'])
             ['users'])
     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_list_users_for_tenant(self):
     # Return a list of all users for a tenant
     tenant = self.setup_test_tenant()
     user_ids = list()
     fetched_user_ids = list()
     password1 = data_utils.rand_password()
     alt_tenant_user1 = data_utils.rand_name('tenant_user1')
     user1 = self.users_client.create_user(name=alt_tenant_user1,
                                           password=password1,
                                           tenantId=tenant['id'],
                                           email='[email protected]')['user']
     user_ids.append(user1['id'])
     # Delete the User at the end of the test
     self.addCleanup(self.users_client.delete_user, user1['id'])
     password2 = data_utils.rand_password()
     alt_tenant_user2 = data_utils.rand_name('tenant_user2')
     user2 = self.users_client.create_user(name=alt_tenant_user2,
                                           password=password2,
                                           tenantId=tenant['id'],
                                           email='[email protected]')['user']
     user_ids.append(user2['id'])
     # Delete the User at the end of the test
     self.addCleanup(self.users_client.delete_user, user2['id'])
     # List of users for the respective tenant ID
     body = (self.tenants_client.list_tenant_users(tenant['id'])
             ['users'])
     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 resource_setup(cls):
        super(ServerRescueNegativeTestJSON, cls).resource_setup()
        cls.device = CONF.compute.volume_device_name
        cls.password = data_utils.rand_password()
        rescue_password = data_utils.rand_password()
        # Server for negative tests
        server = cls.create_test_server(adminPass=cls.password, wait_until="BUILD")
        resc_server = cls.create_test_server(adminPass=rescue_password, wait_until="ACTIVE")
        cls.server_id = server["id"]
        cls.rescue_id = resc_server["id"]

        cls.servers_client.rescue_server(cls.rescue_id, adminPass=rescue_password)
        waiters.wait_for_server_status(cls.servers_client, cls.rescue_id, "RESCUE")
        waiters.wait_for_server_status(cls.servers_client, cls.server_id, "ACTIVE")

 def test_authentication_with_invalid_username(self):
     # Non-existent user's token should not get authenticated
     password = data_utils.rand_password()
     user = self.setup_test_user(password)
     tenant = self.tenants_client.show_tenant(user['tenantId'])['tenant']
     self.assertRaises(lib_exc.Unauthorized, self.token_client.auth,
                       'junkuser123', password, tenant['name'])

 def resource_setup(cls):
     super(RolesV3TestJSON, cls).resource_setup()
     for _ in range(3):
         role_name = data_utils.rand_name(name='role')
         role = cls.roles_client.create_role(name=role_name)['role']
         cls.data.roles.append(role)
     cls.fetched_role_ids = list()
     u_name = data_utils.rand_name('user')
     u_desc = '%s description' % u_name
     u_email = '%[email protected]' % u_name
     cls.u_password = data_utils.rand_password()
     cls.domain = cls.domains_client.create_domain(
         data_utils.rand_name('domain'),
         description=data_utils.rand_name('domain-desc'))['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.group_body = cls.groups_client.create_group(
         name=data_utils.rand_name('Group'), project_id=cls.project['id'],
         domain_id=cls.domain['id'])['group']
     cls.user_body = cls.users_client.create_user(
         u_name, description=u_desc, password=cls.u_password,
         email=u_email, project_id=cls.project['id'],
         domain_id=cls.domain['id'])['user']
     cls.role = cls.roles_client.create_role(
         name=data_utils.rand_name('Role'))['role']

 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_password()
     # first:create a tenant
     tenant_name = data_utils.rand_name(name='tenant')
     tenant = self.tenants_client.create_tenant(tenant_name)['tenant']
     self.data.tenants.append(tenant)
     # second:create a user
     user = self.users_client.create_user(name=user_name,
                                          password=user_password,
                                          tenantId=tenant['id'],
                                          email='')['user']
     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.show_token(token_id)['access']
     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 test_list_users_with_roles_for_tenant(self):
        # Return list of users on tenant when roles are assigned to users
        user = self.setup_test_user()
        tenant = self.tenants_client.show_tenant(user['tenantId'])['tenant']
        role = self.setup_test_role()
        # Assigning roles to two users
        user_ids = list()
        fetched_user_ids = list()
        user_ids.append(user['id'])
        role = self.roles_client.create_user_role_on_project(
            tenant['id'], user['id'], role['id'])['role']

        alt_user2 = data_utils.rand_name('second_user')
        alt_password2 = data_utils.rand_password()
        second_user = self.users_client.create_user(
            name=alt_user2,
            password=alt_password2,
            tenantId=tenant['id'],
            email='[email protected]')['user']
        user_ids.append(second_user['id'])
        # Delete the User at the end of the test
        self.addCleanup(self.users_client.delete_user, second_user['id'])
        role = self.roles_client.create_user_role_on_project(
            tenant['id'], second_user['id'], role['id'])['role']
        # List of users with roles for the respective tenant ID
        body = (self.tenants_client.list_tenant_users(tenant['id'])['users'])
        for i in body:
            fetched_user_ids.append(i['id'])
        # verifying the user Id in the list
        missing_users = [missing_user for missing_user in user_ids
                         if missing_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_associate_user_to_project(self):
        # Associate a user to a project
        # Create a Project
        p_name = data_utils.rand_name('project')
        project = self.projects_client.create_project(p_name)['project']
        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_password()
        user = self.users_client.create_user(
            u_name, description=u_desc, password=u_password,
            email=u_email, project_id=project['id'])['user']
        # Delete the User at the end of this method
        self.addCleanup(self.users_client.delete_user, user['id'])

        # Get User To validate the user details
        new_user_get = self.users_client.show_user(user['id'])['user']
        # 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_list_users_with_roles_for_tenant(self):
        # Return list of users on tenant when roles are assigned to users
        self.data.setup_test_user()
        self.data.setup_test_role()
        user = self.get_user_by_name(self.data.user['name'])
        tenant = self.get_tenant_by_name(self.data.tenant['name'])
        role = self.get_role_by_name(self.data.role['name'])
        # Assigning roles to two users
        user_ids = list()
        fetched_user_ids = list()
        user_ids.append(user['id'])
        role = self.roles_client.assign_user_role(tenant['id'], user['id'],
                                                  role['id'])['role']

        alt_user2 = data_utils.rand_name('second_user')
        alt_password2 = data_utils.rand_password()
        second_user = self.users_client.create_user(alt_user2, alt_password2,
                                                    self.data.tenant['id'],
                                                    '[email protected]')['user']
        user_ids.append(second_user['id'])
        self.data.users.append(second_user)
        role = self.roles_client.assign_user_role(tenant['id'],
                                                  second_user['id'],
                                                  role['id'])['role']
        # List of users with roles for the respective tenant ID
        body = (self.tenants_client.list_tenant_users(self.data.tenant['id'])
                ['users'])
        for i in body:
            fetched_user_ids.append(i['id'])
        # verifying the user Id in the list
        missing_users = [missing_user for missing_user in user_ids
                         if missing_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_group_users_add_list_delete(self):
        name = data_utils.rand_name('Group')
        group = self.groups_client.create_group(
            name=name, domain_id=self.data.domain['id'])['group']
        self.addCleanup(self.groups_client.delete_group, group['id'])
        # add user into group
        users = []
        for i in range(3):
            name = data_utils.rand_name('User')
            password = data_utils.rand_password()
            user = self.users_client.create_user(name, password)['user']
            users.append(user)
            self.addCleanup(self.users_client.delete_user, user['id'])
            self.groups_client.add_group_user(group['id'], user['id'])

        # list users in group
        group_users = self.groups_client.list_group_users(group['id'])['users']
        self.assertEqual(sorted(users), sorted(group_users))
        # check and delete user in group
        for user in users:
            self.groups_client.check_group_user_existence(
                group['id'], user['id'])
            self.groups_client.delete_group_user(group['id'], user['id'])
        group_users = self.groups_client.list_group_users(group['id'])['users']
        self.assertEqual(len(group_users), 0)

 def _create_test_user(self, **kwargs):
     username = data_utils.rand_name('test_user')
     self.user_password = data_utils.rand_password()
     self.user = self.users_client.create_user(
         username, password=self.user_password,
         email=username + '@testmail.tm', **kwargs)['user']
     self.users.append(self.user)

    def test_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 = self.identity_utils.create_project(name=project_name,
                                                     description=project_desc)
        project_id = project['id']
        self.addCleanup(self.identity_utils.delete_project, project_id)

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

        # 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 = self.identity_utils.create_user(username=user_name,
                                               password=password,
                                               project=project,
                                               email=email)
        if 'user' in user:
            user = user['user']
        user_id = user['id']
        self.addCleanup(self.identity_utils.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_rescope_token(self):
        """An unscoped token can be requested

        That token can be used to request a scoped token.
        """

        # Create a user.
        user_name = data_utils.rand_name(name='user')
        user_password = data_utils.rand_password()
        tenant_id = None  # No default tenant so will get unscoped token.
        email = ''
        user = self.users_client.create_user(name=user_name,
                                             password=user_password,
                                             tenantId=tenant_id,
                                             email=email)['user']
        self.data.users.append(user)

        # Create a couple tenants.
        tenant1_name = data_utils.rand_name(name='tenant')
        tenant1 = self.tenants_client.create_tenant(
            name=tenant1_name)['tenant']
        self.data.tenants.append(tenant1)

        tenant2_name = data_utils.rand_name(name='tenant')
        tenant2 = self.tenants_client.create_tenant(
            name=tenant2_name)['tenant']
        self.data.tenants.append(tenant2)

        # Create a role
        role_name = data_utils.rand_name(name='role')
        role = self.roles_client.create_role(name=role_name)['role']
        self.data.roles.append(role)

        # Grant the user the role on the tenants.
        self.roles_client.create_user_role_on_project(tenant1['id'],
                                                      user['id'],
                                                      role['id'])

        self.roles_client.create_user_role_on_project(tenant2['id'],
                                                      user['id'],
                                                      role['id'])

        # Get an unscoped token.
        body = self.token_client.auth(user_name, user_password)

        token_id = body['token']['id']

        # Use the unscoped token to get a token scoped to tenant1
        body = self.token_client.auth_token(token_id,
                                            tenant=tenant1_name)

        scoped_token_id = body['token']['id']

        # Revoke the scoped token
        self.client.delete_token(scoped_token_id)

        # Use the unscoped token to get a token scoped to tenant2
        body = self.token_client.auth_token(token_id,
                                            tenant=tenant2_name)

 def test_authentication_with_invalid_tenant(self):
     # User's token for an invalid tenant should not be authenticated
     password = data_utils.rand_password()
     user = self.setup_test_user(password)
     self.assertRaises(lib_exc.Unauthorized, self.token_client.auth,
                       user['name'],
                       password,
                       'junktenant1234')

 def _create_test_user(self, **kwargs):
     self.test_user = data_utils.rand_name('test_user')
     self.test_password = data_utils.rand_password()
     self.test_email = self.test_user + '@testmail.tm'
     self.user = self.users_client.create_user(
         self.test_user, password=self.test_password,
         email=self.test_email, **kwargs)['user']
     self.users.append(self.user)

 def _create_test_user(self, **kwargs):
     if kwargs['password'] is None:
         user_password = data_utils.rand_password()
         kwargs['password'] = user_password
     user = self.users_client.create_user(**kwargs)['user']
     # Delete the user at the end of the test
     self.addCleanup(self.users_client.delete_user, user['id'])
     return user

 def test_authentication_for_disabled_user(self):
     # Attempt to authenticate for disabled user should fail
     password = data_utils.rand_password()
     user = self.setup_test_user(password)
     self.disable_user(user['name'], user['domain_id'])
     self.assertRaises(lib_exc.Unauthorized, self.token.auth,
                       username=user['name'],
                       password=password,
                       user_domain_id='default')

 def resource_setup(cls):
     cls.set_validation_resources()
     super(HybridCreateAwsServersTestJSON, cls).resource_setup()
     cls.meta = {'hello': 'world'}
     cls.accessIPv4 = '1.1.1.1'
     cls.accessIPv6 = '0000:0000:0000:0000:0000:babe:220.12.22.2'
     cls.name = data_utils.rand_name('server')
     cls.password = data_utils.rand_password()
     disk_config = cls.disk_config

 def test_create_user_for_non_existent_domain(self):
     # Attempt to create a user in a non-existent domain should fail
     u_name = data_utils.rand_name('user')
     u_email = u_name + '@testmail.tm'
     u_password = data_utils.rand_password()
     self.assertRaises(lib_exc.NotFound, self.users_client.create_user,
                       u_name, u_password,
                       email=u_email,
                       domain_id=data_utils.rand_uuid_hex())