def test_rand_password_with_len_2(self):
     actual = data_utils.rand_password(2)
     self.assertIsInstance(actual, str)
     self.assertEqual(len(actual), 3)
     self.assertRegex(actual, "[[email protected]#$%^&*_=+]{3}")
     actual2 = data_utils.rand_password(2)
     self.assertNotEqual(actual, actual2)

 def test_rand_password_with_len(self):
     actual = data_utils.rand_password(8)
     self.assertIsInstance(actual, str)
     self.assertEqual(len(actual), 8)
     self.assertRegexpMatches(actual, "[[email protected]#$%^&*_=+]{8}")
     actual2 = data_utils.rand_password(8)
     self.assertNotEqual(actual, actual2)

    def test_password_history_check_self_service_api(self):
        old_pass = self.creds.password
        new_pass1 = data_utils.rand_password()
        new_pass2 = data_utils.rand_password()

        self.addCleanup(self._restore_password, old_pass, new_pass2)

        # Update password
        self._update_password(original_password=old_pass, password=new_pass1)

        if CONF.identity.user_unique_last_password_count > 1:
            # Can not reuse a previously set password
            self.assertRaises(exceptions.BadRequest,
                              self.non_admin_users_client.update_user_password,
                              self.user_id,
                              password=new_pass1,
                              original_password=new_pass1)

            self.assertRaises(exceptions.BadRequest,
                              self.non_admin_users_client.update_user_password,
                              self.user_id,
                              password=old_pass,
                              original_password=new_pass1)

        # A different password can be set
        self._update_password(original_password=new_pass1, password=new_pass2)

 def test_rand_password_with_len_2(self):
     actual = data_utils.rand_password(2)
     self.assertIsInstance(actual, str)
     self.assertEqual(len(actual), 3)
     self.assertRegex(actual, "[[email protected]#%^&*_=+]{3}")
     actual2 = data_utils.rand_password(2)
     # NOTE(masayukig): Originally, we checked that the acutal and actual2
     # are different each other. But only 3 letters can be the same value
     # in a very rare case. So, we just check the length here, too,
     # just in case.
     self.assertEqual(len(actual2), 3)

    def test_user_update(self):
        # Test case to check if updating of user attributes is successful.
        # Creating first 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(
            name=u_name, description=u_desc, password=u_password,
            email=u_email, enabled=False)['user']
        # Delete the User at the end of this method
        self.addCleanup(self.users_client.delete_user, user['id'])

        # Creating second project for updation
        project = self.setup_test_project()

        # Updating user details with new values
        update_kwargs = {'name': data_utils.rand_name('user2'),
                         'description': data_utils.rand_name('desc2'),
                         'project_id': project['id'],
                         'email': '[email protected]',
                         'enabled': False}
        updated_user = self.users_client.update_user(
            user['id'], **update_kwargs)['user']
        for field in update_kwargs:
            self.assertEqual(update_kwargs[field], updated_user[field])

        # GET by id after updating
        new_user_get = self.users_client.show_user(user['id'])['user']
        # Assert response body of GET after updation
        for field in update_kwargs:
            self.assertEqual(update_kwargs[field], new_user_get[field])

    def recreate_server(cls, server_id, validatable=False, **kwargs):
        """Destroy an existing class level server and creates a new one

        Some test classes use a test server that can be used by multiple
        tests. This is done to optimise runtime and test load.
        If something goes wrong with the test server, it can be rebuilt
        using this helper.

        This helper can also be used for the initial provisioning if no
        server_id is specified.

        :param server_id: UUID of the server to be rebuilt. If None is
            specified, a new server is provisioned.
        :param validatable: whether to the server needs to be
            validatable. When True, validation resources are acquired via
            the `get_class_validation_resources` helper.
        :param kwargs: extra paramaters are passed through to the
            `create_test_server` call.
        :return: the UUID of the created server.
        """
        if server_id:
            cls.delete_server(server_id)

        cls.password = data_utils.rand_password()
        server = cls.create_test_server(
            validatable,
            validation_resources=cls.get_class_validation_resources(
                cls.os_primary),
            wait_until='ACTIVE',
            adminPass=cls.password,
            **kwargs)
        return server['id']

 def resource_setup(cls):
     super(RolesV3TestJSON, cls).resource_setup()
     cls.roles = list()
     for _ in range(3):
         role_name = data_utils.rand_name(name='role')
         role = cls.roles_client.create_role(name=role_name)['role']
         cls.roles.append(role)
     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.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.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(
         name=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 _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_user_account_lockout(self):
        if (CONF.identity.user_lockout_failure_attempts <= 0 or
                CONF.identity.user_lockout_duration <= 0):
            raise self.skipException(
                "Both CONF.identity.user_lockout_failure_attempts and "
                "CONF.identity.user_lockout_duration should be greater than "
                "zero to test this feature")

        password = self.creds.password

        # First, we login using the correct credentials
        self.non_admin_token.auth(user_id=self.user_id, password=password)

        # Lock user account by using the wrong password to login
        bad_password = data_utils.rand_password()
        for _ in range(CONF.identity.user_lockout_failure_attempts):
            self.assertRaises(exceptions.Unauthorized,
                              self.non_admin_token.auth,
                              user_id=self.user_id,
                              password=bad_password)

        # The user account must be locked, so now it is not possible to login
        # even using the correct password
        self.assertRaises(exceptions.Unauthorized,
                          self.non_admin_token.auth,
                          user_id=self.user_id,
                          password=password)

        # If we wait the required time, the user account will be unlocked
        time.sleep(CONF.identity.user_lockout_duration + 1)
        self.non_admin_token.auth(user_id=self.user_id, password=password)

 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 create_trustor_and_roles(self):
        # create a project that trusts will be granted on
        trustor_project_name = data_utils.rand_name(name='project')
        project = self.projects_client.create_project(
            trustor_project_name,
            domain_id=CONF.identity.default_domain_id)['project']
        self.trustor_project_id = project['id']
        self.assertIsNotNone(self.trustor_project_id)

        # Create a trustor User
        trustor_username = data_utils.rand_name('user')
        u_desc = trustor_username + 'description'
        u_email = trustor_username + '@testmail.xx'
        trustor_password = data_utils.rand_password()
        user = self.users_client.create_user(
            name=trustor_username,
            description=u_desc,
            password=trustor_password,
            email=u_email,
            project_id=self.trustor_project_id,
            domain_id=CONF.identity.default_domain_id)['user']
        self.trustor_user_id = user['id']

        # And two roles, one we'll delegate and one we won't
        self.delegated_role = data_utils.rand_name('DelegatedRole')
        self.not_delegated_role = data_utils.rand_name('NotDelegatedRole')

        role = self.roles_client.create_role(name=self.delegated_role)['role']
        self.delegated_role_id = role['id']

        role = self.roles_client.create_role(
            name=self.not_delegated_role)['role']
        self.not_delegated_role_id = role['id']

        # Assign roles to trustor
        self.roles_client.create_user_role_on_project(
            self.trustor_project_id,
            self.trustor_user_id,
            self.delegated_role_id)
        self.roles_client.create_user_role_on_project(
            self.trustor_project_id,
            self.trustor_user_id,
            self.not_delegated_role_id)

        # Get trustee user ID, use the demo user
        trustee_username = self.non_admin_client.user
        self.trustee_user_id = self.get_user_by_name(trustee_username)['id']
        self.assertIsNotNone(self.trustee_user_id)

        # Initialize a new client with the trustor credentials
        creds = common_creds.get_credentials(
            identity_version='v3',
            username=trustor_username,
            password=trustor_password,
            user_domain_id=CONF.identity.default_domain_id,
            tenant_name=trustor_project_name,
            project_domain_id=CONF.identity.default_domain_id,
            domain_id=CONF.identity.default_domain_id)
        os = clients.Manager(credentials=creds)
        self.trustor_client = os.trusts_client

 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_server_with_personality(self):
     file_contents = 'This is a test file.'
     file_path = '/test.txt'
     personality = [{'path': file_path,
                     'contents': base64.encode_as_text(file_contents)}]
     password = data_utils.rand_password()
     validation_resources = self.get_test_validation_resources(
         self.os_primary)
     created_server = self.create_test_server(
         personality=personality, adminPass=password, wait_until='ACTIVE',
         validatable=True,
         validation_resources=validation_resources)
     self.addCleanup(waiters.wait_for_server_termination,
                     self.servers_client, created_server['id'])
     self.addCleanup(test_utils.call_and_ignore_notfound_exc,
                     self.servers_client.delete_server,
                     created_server['id'])
     server = self.client.show_server(created_server['id'])['server']
     if CONF.validation.run_validation:
         linux_client = remote_client.RemoteClient(
             self.get_server_ip(server, validation_resources),
             self.ssh_user, password,
             validation_resources['keypair']['private_key'],
             server=server,
             servers_client=self.client)
         self.assertEqual(file_contents,
                          linux_client.exec_command(
                              'sudo cat %s' % file_path))

 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_creds(self, admin=False, roles=None):
        """Create credentials with random name.

        Creates project and user. When admin flag is True create user
        with admin role. Assign user with additional roles (for example
        _member_) and roles requested by caller.

        :param admin: Flag if to assign to the user admin role
        :type admin: bool
        :param roles: Roles to assign for the user
        :type roles: list
        :return: Readonly Credentials with network resources
        """
        root = self.name

        project_name = data_utils.rand_name(root, prefix=self.resource_prefix)
        project_desc = project_name + "-desc"
        project = self.creds_client.create_project(
            name=project_name, description=project_desc)

        # NOTE(andreaf) User and project can be distinguished from the context,
        # having the same ID in both makes it easier to match them and debug.
        username = project_name
        user_password = data_utils.rand_password()
        email = data_utils.rand_name(
            root, prefix=self.resource_prefix) + "@example.com"
        user = self.creds_client.create_user(
            username, user_password, project, email)
        role_assigned = False
        if admin:
            self.creds_client.assign_user_role(user, project, self.admin_role)
            role_assigned = True
            if (self.identity_version == 'v3' and
                    self.identity_admin_domain_scope):
                self.creds_client.assign_user_role_on_domain(
                    user, self.identity_admin_role)
        # Add roles specified in config file
        for conf_role in self.extra_roles:
            self.creds_client.assign_user_role(user, project, conf_role)
            role_assigned = True
        # Add roles requested by caller
        if roles:
            for role in roles:
                self.creds_client.assign_user_role(user, project, role)
                role_assigned = True
        # NOTE(mtreinish) For a user to have access to a project with v3 auth
        # it must beassigned a role on the project. So we need to ensure that
        # our newly created user has a role on the newly created project.
        if self.identity_version == 'v3' and not role_assigned:
            try:
                self.creds_client.create_user_role('Member')
            except lib_exc.Conflict:
                LOG.warning('Member role already exists, ignoring conflict.')
            self.creds_client.assign_user_role(user, project, 'Member')

        creds = self.creds_client.get_credentials(user, project, user_password)
        return cred_provider.TestResources(creds)

 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,
                       name=u_name, password=u_password,
                       email=u_email,
                       domain_id=data_utils.rand_uuid_hex())

    def resource_setup(cls):
        super(ServerRescueNegativeTestJSON, cls).resource_setup()
        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_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=CONF.identity.default_domain_id)

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

        password = data_utils.rand_password()
        server = cls.create_test_server(adminPass=password,
                                        wait_until='ACTIVE')
        cls.servers_client.rescue_server(server['id'], adminPass=password)
        waiters.wait_for_server_status(cls.servers_client, server['id'],
                                       'RESCUE')
        cls.rescued_server_id = server['id']

 def test_create_user_with_duplicate_name(self):
     # Duplicate user should not be created
     password = data_utils.rand_password()
     user = self.setup_test_user(password)
     tenant = self.tenants_client.show_tenant(user['tenantId'])['tenant']
     self.assertRaises(lib_exc.Conflict, self.users_client.create_user,
                       name=user['name'],
                       password=password,
                       tenantId=tenant['id'],
                       email=user['email'])