Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _get_idempotent_id method in tempest

Best Python code snippet using tempest_python

check_uuid.py

Source:check_uuid.py

...93                    if not module_name.startswith(UNIT_TESTS_EXCLUDE):94                        modules.append(module_name)95        return modules96    @staticmethod97    def _get_idempotent_id(test_node):98        """99        Return key-value dict with all metadata from @test.idempotent_id100        decorators for test method101        """102        idempotent_id = None103        for decorator in test_node.decorator_list:104            if (hasattr(decorator, 'func') and105                hasattr(decorator.func, 'attr') and106                decorator.func.attr == DECORATOR_NAME and107                hasattr(decorator.func, 'value') and108                decorator.func.value.id == DECORATOR_MODULE):109                for arg in decorator.args:110                    idempotent_id = ast.literal_eval(arg)111        return idempotent_id112    @staticmethod113    def _is_decorator(line):114        return line.strip().startswith('@')115    @staticmethod116    def _is_def(line):117        return line.strip().startswith('def ')118    def _add_uuid_to_test(self, patcher, test_node, source_path):119        with open(source_path) as src:120            src_lines = src.read().split('\n')121        lineno = test_node.lineno122        insert_position = lineno123        while True:124            if (self._is_def(src_lines[lineno - 1]) or125                    (self._is_decorator(src_lines[lineno - 1]) and126                        (DECORATOR_TEMPLATE.split('(')[0] <=127                            src_lines[lineno - 1].strip().split('(')[0]))):128                insert_position = lineno129                break130            lineno += 1131        patcher.add_patch(132            source_path,133            ' ' * test_node.col_offset + DECORATOR_TEMPLATE % uuid.uuid4(),134            insert_position135        )136    @staticmethod137    def _is_test_case(module, node):138        if (node.__class__ is ast.ClassDef and139                hasattr(module, node.name) and140                inspect.isclass(getattr(module, node.name))):141            return issubclass(getattr(module, node.name), unittest.TestCase)142    @staticmethod143    def _is_test_method(node):144        return (node.__class__ is ast.FunctionDef145                and node.name.startswith('test_'))146    @staticmethod147    def _next_node(body, node):148        if body.index(node) < len(body):149            return body[body.index(node) + 1]150    @staticmethod151    def _import_name(node):152        if type(node) == ast.Import:153            return node.names[0].name154        elif type(node) == ast.ImportFrom:155            return '%s.%s' % (node.module, node.names[0].name)156    def _add_import_for_test_uuid(self, patcher, src_parsed, source_path):157        with open(source_path) as f:158            src_lines = f.read().split('\n')159        line_no = 0160        tempest_imports = [node for node in src_parsed.body161                           if self._import_name(node) and162                           'tempest.' in self._import_name(node)]163        if not tempest_imports:164            import_snippet = '\n'.join(('', IMPORT_LINE, ''))165        else:166            for node in tempest_imports:167                if self._import_name(node) < DECORATOR_IMPORT:168                    continue169                else:170                    line_no = node.lineno171                    import_snippet = IMPORT_LINE172                    break173            else:174                line_no = tempest_imports[-1].lineno175                while True:176                    if (not src_lines[line_no - 1] or177                            getattr(self._next_node(src_parsed.body,178                                                    tempest_imports[-1]),179                                    'lineno') == line_no or180                            line_no == len(src_lines)):181                        break182                    line_no += 1183                import_snippet = '\n'.join((IMPORT_LINE, ''))184        patcher.add_patch(source_path, import_snippet, line_no)185    def get_tests(self):186        """Get test methods with sources from base package with metadata"""187        tests = {}188        for module_name in self._modules_search():189            tests[module_name] = {}190            module = importlib.import_module(module_name)191            source_path = '.'.join(192                (os.path.splitext(module.__file__)[0], 'py')193            )194            with open(source_path, 'r') as f:195                source = f.read()196            tests[module_name]['source_path'] = source_path197            tests[module_name]['tests'] = {}198            source_parsed = ast.parse(source)199            tests[module_name]['ast'] = source_parsed200            tests[module_name]['import_valid'] = (201                hasattr(module, DECORATOR_MODULE) and202                inspect.ismodule(getattr(module, DECORATOR_MODULE))203            )204            test_cases = (node for node in source_parsed.body205                          if self._is_test_case(module, node))206            for node in test_cases:207                for subnode in filter(self._is_test_method, node.body):208                        test_name = '%s.%s' % (node.name, subnode.name)209                        tests[module_name]['tests'][test_name] = subnode210        return tests211    @staticmethod212    def _filter_tests(function, tests):213        """Filter tests with condition 'function(test_node) == True'"""214        result = {}215        for module_name in tests:216            for test_name in tests[module_name]['tests']:217                if function(module_name, test_name, tests):218                    if module_name not in result:219                        result[module_name] = {220                            'ast': tests[module_name]['ast'],221                            'source_path': tests[module_name]['source_path'],222                            'import_valid': tests[module_name]['import_valid'],223                            'tests': {}224                        }225                    result[module_name]['tests'][test_name] = \226                        tests[module_name]['tests'][test_name]227        return result228    def find_untagged(self, tests):229        """Filter all tests without uuid in metadata"""230        def check_uuid_in_meta(module_name, test_name, tests):231            idempotent_id = self._get_idempotent_id(232                tests[module_name]['tests'][test_name])233            return not idempotent_id234        return self._filter_tests(check_uuid_in_meta, tests)235    def report_collisions(self, tests):236        """Reports collisions if there are any. Returns true if237        collisions exist.238        """239        uuids = {}240        def report(module_name, test_name, tests):241            test_uuid = self._get_idempotent_id(242                tests[module_name]['tests'][test_name])243            if not test_uuid:244                return245            if test_uuid in uuids:246                error_str = "%s:%s\n uuid %s collision: %s<->%s\n%s:%s" % (247                    tests[module_name]['source_path'],248                    tests[module_name]['tests'][test_name].lineno,249                    test_uuid,250                    test_name,251                    uuids[test_uuid]['test_name'],252                    uuids[test_uuid]['source_path'],253                    uuids[test_uuid]['test_node'].lineno,254                )255                print(error_str)...

common.py

Source:common.py

...73                                     request_field=args[0])74def logrpc(f):75    def tab(what):76        return '\t' + '\n\t'.join(filter(None, str(what).split('\n')))77    def _get_idempotent_id(request):78        for param in ('name', 'id', 'volume_id', 'snapshot_id'):79            try:80                return ' ' + getattr(request, param)81            except AttributeError:82                pass83        return ''84    def _get_response_id(response):85        id_str = ' (id =%s)'86        for resource in ('volume', 'snapshot'):87            try:88                return id_str % _get_idempotent_id(getattr(response, resource))89            except AttributeError:90                pass91        return ''92    @functools.wraps(f)93    def dolog(self, request, context):94        # Set this thread's context95        context_utils.RequestContext(96            overwrite=True,97            user_id=CONF.EMBER_CONFIG['user_id'],98            project_id=CONF.EMBER_CONFIG['project_id'],99            project_name=CONF.PROJECT_NAME)100        start = datetime.utcnow()101        LOG.info('=> GRPC %s%s' % (f.__name__, _get_idempotent_id(request)))102        if request.ListFields():103            msg = ' params:\n%s' % tab(request)104        else:105            msg = 'out params'106        LOG.debug('With%s' % msg)107        try:108            result = f(self, request, context)109        except Exception as exc:110            end = datetime.utcnow()111            if context._state.code:112                code = str(context._state.code)[11:]113                details = context._state.details114                tback = ''115            else:...

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.