KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _sanitize_creds method in tempest

Best Python code snippet using tempest_python

preprov_creds.py

Source:preprov_creds.py

...209            useable_hashes = [x for x in hashes if x not in admin_hashes]210        else:211            useable_hashes = hashes212        return useable_hashes213    def _sanitize_creds(self, creds):214        temp_creds = creds.copy()215        temp_creds.pop('password')216        return temp_creds217    def _get_creds(self, roles=None):218        useable_hashes = self._get_match_hash_list(roles)219        if len(useable_hashes) == 0:220            msg = 'No users configured for type/roles %s' % roles221            raise lib_exc.InvalidCredentials(msg)222        free_hash = self._get_free_hash(useable_hashes)223        clean_creds = self._sanitize_creds(224            self.hash_dict['creds'][free_hash])225        LOG.info('%s allocated creds:\n%s', self.name, clean_creds)226        return self._wrap_creds_with_network(free_hash)227    @lockutils.synchronized('test_accounts_io', external=True)228    def remove_hash(self, hash_string):229        hash_path = os.path.join(self.accounts_dir, hash_string)230        if not os.path.isfile(hash_path):231            LOG.warning('Expected an account lock file %s to remove, but '232                        'one did not exist', hash_path)233        else:234            os.remove(hash_path)235            if not os.listdir(self.accounts_dir):236                os.rmdir(self.accounts_dir)237    def get_hash(self, creds):238        for _hash in self.hash_dict['creds']:239            # Comparing on the attributes that are expected in the YAML240            init_attributes = creds.get_init_attributes()241            # Only use the attributes initially used to calculate the hash242            init_attributes = [x for x in init_attributes if243                               x in self.HASH_CRED_FIELDS]244            hash_attributes = self.hash_dict['creds'][_hash].copy()245            # NOTE(andreaf) Not all fields may be available on all credentials246            # so defaulting to None for that case.247            if all([getattr(creds, k, None) == hash_attributes.get(k, None) for248                   k in init_attributes]):249                return _hash250        raise AttributeError('Invalid credentials %s' % creds)251    def remove_credentials(self, creds):252        _hash = self.get_hash(creds)253        clean_creds = self._sanitize_creds(self.hash_dict['creds'][_hash])254        self.remove_hash(_hash)255        LOG.info("%s returned allocated creds:\n%s", self.name, clean_creds)256    def get_primary_creds(self):257        if self._creds.get('primary'):258            return self._creds.get('primary')259        net_creds = self._get_creds()260        self._creds['primary'] = net_creds261        return net_creds262    def get_alt_creds(self):263        if self._creds.get('alt'):264            return self._creds.get('alt')265        net_creds = self._get_creds()266        self._creds['alt'] = net_creds267        return net_creds268    def get_creds_by_roles(self, roles, force_new=False):269        roles = list(set(roles))270        exist_creds = self._creds.get(six.text_type(roles).encode(271            'utf-8'), None)272        # The force kwarg is used to allocate an additional set of creds with273        # the same role list. The index used for the previously allocation274        # in the _creds dict will be moved.275        if exist_creds and not force_new:276            return exist_creds277        elif exist_creds and force_new:278            # NOTE(andreaf) In py3.x encode returns bytes, and b'' is bytes279            # In py2.7 encode returns strings, and b'' is still string280            new_index = six.text_type(roles).encode('utf-8') + b'-' + \281                six.text_type(len(self._creds)).encode('utf-8')282            self._creds[new_index] = exist_creds283        net_creds = self._get_creds(roles=roles)284        self._creds[six.text_type(roles).encode('utf-8')] = net_creds285        return net_creds286    def clear_creds(self):287        for creds in self._creds.values():288            self.remove_credentials(creds)289    def get_admin_creds(self):290        return self.get_creds_by_roles([self.admin_role])291    def is_role_available(self, role):292        if self.hash_dict['roles'].get(role):293            return True294        return False295    def admin_available(self):296        return self.is_role_available(self.admin_role)297    def _wrap_creds_with_network(self, hash):298        creds_dict = self.hash_dict['creds'][hash]299        # Make sure a domain scope if defined for users in case of V3300        # Make sure a tenant is available in case of V2301        creds_dict = self._extend_credentials(creds_dict)302        # This just builds a Credentials object, it does not validate303        # nor fill  with missing fields.304        credential = auth.get_credentials(305            auth_url=None, fill_in=False,306            identity_version=self.identity_version, **creds_dict)307        net_creds = cred_provider.TestResources(credential)308        net_clients = clients.Manager(credentials=credential)309        compute_network_client = net_clients.compute_networks_client310        net_name = self.hash_dict['networks'].get(hash, None)311        try:312            network = fixed_network.get_network_from_name(313                net_name, compute_network_client)314        except exceptions.InvalidTestResource:315            network = {}316        net_creds.set_resources(network=network)317        return net_creds318    def _extend_credentials(self, creds_dict):319        # Add or remove credential domain fields to fit the identity version320        domain_fields = set(x for x in auth.KeystoneV3Credentials.ATTRIBUTES321                            if 'domain' in x)322        msg = 'Assuming they are valid in the default domain.'323        if self.identity_version == 'v3':324            if not domain_fields.intersection(set(creds_dict.keys())):325                msg = 'Using credentials %s for v3 API calls. ' + msg326                LOG.warning(msg, self._sanitize_creds(creds_dict))327                creds_dict['domain_name'] = self.credentials_domain328        if self.identity_version == 'v2':329            if domain_fields.intersection(set(creds_dict.keys())):330                msg = 'Using credentials %s for v2 API calls. ' + msg331                LOG.warning(msg, self._sanitize_creds(creds_dict))332            # Remove all valid domain attributes333            for attr in domain_fields.intersection(set(creds_dict.keys())):334                creds_dict.pop(attr)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.