How to use _add_case_metadata method in lisa

Best Python code snippet using lisa_python

testsuite.py

Source:testsuite.py

...369        self.name = func.__name__370        self.full_name = func.__qualname__371        self.qualname = func.__qualname__372        self._func = func373        _add_case_metadata(self)374        @wraps(self._func)375        def wrapper(*args: Any, **kwargs: Any) -> None:376            parameters: Dict[str, Any] = {}377            for name in kwargs.keys():378                if name in func.__annotations__:379                    parameters[name] = kwargs[name]380            func(*args, **parameters)381        return wrapper382    @property383    def owner(self) -> str:384        if self._owner:385            return self._owner386        return self.suite.owner387class TestCaseRuntimeData:388    def __init__(self, metadata: TestCaseMetadata):389        self.metadata = metadata390        # all runtime setting fields391        self.select_action: str = ""392        self.times: int = 1393        self.retry: int = 0394        self.use_new_environment: bool = metadata.use_new_environment395        self.ignore_failure: bool = False396        self.environment_name: str = ""397    def __getattr__(self, key: str) -> Any:398        # return attributes of metadata for convenient399        assert self.metadata400        return getattr(self.metadata, key)401    def __repr__(self) -> str:402        return (403            f"name: {self.metadata.name}, "404            f"action: {self.select_action}, "405            f"times: {self.times}, retry: {self.retry}, "406            f"new_env: {self.use_new_environment}, "407            f"ignore_failure: {self.ignore_failure}, "408            f"env_name: {self.environment_name}"409        )410    def clone(self) -> TestCaseRuntimeData:411        cloned = TestCaseRuntimeData(self.metadata)412        fields = [413            constants.TESTCASE_SELECT_ACTION,414            constants.TESTCASE_TIMES,415            constants.TESTCASE_RETRY,416            constants.TESTCASE_USE_NEW_ENVIRONMENT,417            constants.TESTCASE_IGNORE_FAILURE,418            constants.ENVIRONMENT,419        ]420        set_filtered_fields(self, cloned, fields)421        return cloned422class TestSuite:423    def __init__(424        self,425        metadata: TestSuiteMetadata,426    ) -> None:427        super().__init__()428        self._metadata = metadata429        self._should_stop = False430        self.__log = get_logger("suite", metadata.name)431    def before_suite(self, log: Logger, **kwargs: Any) -> None:432        ...433    def after_suite(self, log: Logger, **kwargs: Any) -> None:434        ...435    def before_case(self, log: Logger, **kwargs: Any) -> None:436        ...437    def after_case(self, log: Logger, **kwargs: Any) -> None:438        ...439    def start(440        self,441        environment: Environment,442        # test cases to run, must be a test method in this class.443        case_results: List[TestResult],444        # case accessible variables445        case_variables: Dict[str, Any],446    ) -> None:447        suite_error_message = ""448        # set the environment is not new, when it's used by any suite.449        environment.is_new = False450        test_kwargs = {451            "environment": environment,452            "log": self.__log,453            "node": environment.default_node,454            # copy to prevent the data is changed and effect other cases.455            "variables": copy.copy(case_variables),456        }457        #  replace to case's logger temporarily458        suite_log = self.__log459        (460            is_suite_continue,461            suite_error_message,462            suite_error_stacktrace,463        ) = self.__suite_method(464            self.before_suite, test_kwargs=test_kwargs, log=suite_log465        )466        for case_result in case_results:467            case_name = case_result.runtime_data.name468            case_result.environment = environment469            case_log = get_logger("case", case_name, parent=self.__log)470            case_log_path = self.__create_case_log_path(case_name)471            case_part_path = self.__get_test_part_path(case_log_path)472            case_working_path = self.__get_case_working_path(case_part_path)473            case_unique_name = case_log_path.name474            case_log_file = case_log_path / f"{case_log_path.name}.log"475            case_log_handler = create_file_handler(case_log_file, case_log)476            add_handler(case_log_handler, environment.log)477            case_kwargs = test_kwargs.copy()478            case_kwargs.update({"case_name": case_unique_name})479            case_kwargs.update({"log": case_log})480            case_kwargs.update({"log_path": case_log_path})481            case_kwargs.update({"working_path": case_working_path})482            case_kwargs.update({"part_path": case_part_path})483            case_kwargs.update({"result": case_result})484            case_log.info(485                f"test case '{case_result.runtime_data.full_name}' is running"486            )487            is_continue: bool = is_suite_continue488            total_timer = create_timer()489            case_result.log_file = case_log_file.relative_to(490                constants.RUN_LOCAL_LOG_PATH491            ).as_posix()492            case_result.set_status(TestStatus.RUNNING, "")493            case_timeout = case_result.runtime_data.metadata.timeout494            if is_continue:495                is_continue = self.__before_case(496                    case_result=case_result,497                    timeout=case_timeout,498                    test_kwargs=case_kwargs,499                    log=case_log,500                )501            else:502                case_result.stacktrace = suite_error_stacktrace503                case_result.set_status(TestStatus.SKIPPED, suite_error_message)504            if is_continue:505                self.__run_case(506                    case_result=case_result,507                    timeout=case_timeout,508                    test_kwargs=case_kwargs,509                    log=case_log,510                )511            self.__after_case(512                case_result=case_result,513                timeout=case_timeout,514                test_kwargs=case_kwargs,515                log=case_log,516            )517            case_log.info(518                f"result: {case_result.status.name}, " f"elapsed: {total_timer}"519            )520            remove_handler(case_log_handler, case_log)521            remove_handler(case_log_handler, environment.log)522            if case_log_handler:523                case_log_handler.close()524            if self._should_stop:525                suite_log.info("received stop message, stop run")526                break527        self.__suite_method(self.after_suite, test_kwargs=test_kwargs, log=suite_log)528    def stop(self) -> None:529        self._should_stop = True530    def __create_case_log_path(self, case_name: str) -> Path:531        while True:532            path = (533                constants.RUN_LOCAL_LOG_PATH534                / "tests"535                / f"{get_datetime_path()}-{case_name}"536            )537            if not path.exists():538                break539            sleep(0.1)540        # avoid to create folder for UT541        if not is_unittest():542            path.mkdir(parents=True)543        return path544    def __get_test_part_path(self, log_path: Path) -> Path:545        if is_unittest():546            return Path()547        return Path(log_path.parts[-2]) / log_path.parts[-1]548    def __get_case_working_path(self, test_part_path: Path) -> Path:549        if is_unittest():550            return Path()551        # The working path should be the same name as log_path, so it's easy to552        # associated. Unlike the log path, the working path won't be created, because553        # it's not used in most cases. So it doesn't need to be created too. The554        # test case should create it, when it's used.555        working_path = constants.RUN_LOCAL_WORKING_PATH / test_part_path556        return working_path557    def __suite_method(558        self, method: Callable[..., Any], test_kwargs: Dict[str, Any], log: Logger559    ) -> Tuple[bool, str, Optional[str]]:560        result: bool = True561        message: str = ""562        timer = create_timer()563        method_name = method.__name__564        stacktrace: Optional[str] = None565        try:566            self.__call_with_retry_and_timeout(567                method,568                retries=0,569                timeout=3600,570                log=log,571                test_kwargs=test_kwargs,572            )573        except Exception as identifier:574            result = False575            message = f"{method_name}: {identifier}"576            stacktrace = traceback.format_exc()577        log.debug(f"{method_name} end in {timer}")578        return result, message, stacktrace579    def __before_case(580        self,581        case_result: TestResult,582        timeout: int,583        test_kwargs: Dict[str, Any],584        log: Logger,585    ) -> bool:586        result: bool = True587        timer = create_timer()588        try:589            self.__call_with_retry_and_timeout(590                self.before_case,591                retries=case_result.runtime_data.retry,592                timeout=timeout,593                log=log,594                test_kwargs=test_kwargs,595            )596        except Exception as identifier:597            log.error("before_case: ", exc_info=identifier)598            case_result.stacktrace = traceback.format_exc()599            case_result.set_status(TestStatus.SKIPPED, f"before_case: {identifier}")600            result = False601        log.debug(f"before_case end in {timer}")602        return result603    def __after_case(604        self,605        case_result: TestResult,606        timeout: int,607        test_kwargs: Dict[str, Any],608        log: Logger,609    ) -> None:610        timer = create_timer()611        try:612            self.__call_with_retry_and_timeout(613                self.after_case,614                retries=case_result.runtime_data.retry,615                timeout=timeout,616                log=log,617                test_kwargs=test_kwargs,618            )619        except Exception as identifier:620            # after case doesn't impact test case result.621            log.error("after_case failed", exc_info=identifier)622        log.debug(f"after_case end in {timer}")623    def __run_case(624        self,625        case_result: TestResult,626        timeout: int,627        test_kwargs: Dict[str, Any],628        log: Logger,629    ) -> None:630        timer = create_timer()631        case_name = case_result.runtime_data.name632        test_method = getattr(self, case_name)633        try:634            self.__call_with_retry_and_timeout(635                test_method,636                retries=case_result.runtime_data.retry,637                timeout=timeout,638                log=log,639                test_kwargs=test_kwargs,640            )641            case_result.set_status(TestStatus.PASSED, "")642        except Exception as identifier:643            case_result.handle_exception(exception=identifier, log=log)644        log.debug(f"case end in {timer}")645    def __call_with_retry_and_timeout(646        self,647        method: Callable[..., Any],648        retries: int,649        timeout: int,650        log: Logger,651        test_kwargs: Dict[str, Any],652    ) -> None:653        try:654            retry_call(655                func_timeout,656                fkwargs={657                    "timeout": timeout,658                    "func": method,659                    "kwargs": test_kwargs,660                },661                exceptions=Exception,662                tries=retries + 1,663                logger=log,664            )665        except FunctionTimedOut:666            # FunctionTimedOut is a special exception. If it's not captured667            # explicitly, it will make the whole program exit.668            raise TimeoutError(f"time out in {timeout} seconds.")669def get_suites_metadata() -> Dict[str, TestSuiteMetadata]:670    return _all_suites671def get_cases_metadata() -> Dict[str, TestCaseMetadata]:672    return _all_cases673def _add_suite_metadata(metadata: TestSuiteMetadata) -> None:674    key = metadata.test_class.__name__675    exist_metadata = _all_suites.get(key)676    if exist_metadata is None:677        _all_suites[key] = metadata678    else:679        raise LisaException(680            f"duplicate test class name: {key}, "681            f"new: [{metadata}], exists: [{exist_metadata}]"682        )683    class_prefix = f"{key}."684    for test_case in _all_cases.values():685        if test_case.full_name.startswith(class_prefix):686            _add_case_to_suite(metadata, test_case)687    log = get_logger("init", "test")688    log.info(689        f"registered test suite '{key}' "690        f"with test cases: '{', '.join([case.name for case in metadata.cases])}'"691    )692def _add_case_metadata(metadata: TestCaseMetadata) -> None:693    qualname = metadata.qualname694    if _all_cases.get(qualname) is None:695        _all_cases[qualname] = metadata696    else:697        raise LisaException(698            f"found duplicate test class name: {qualname}. "699            "Check there is no duplicate test class name, "700            "and not import by extension twice."701        )702    # this should be None in current observation.703    # the methods are loaded prior to test class704    # in case logic is changed, so keep this logic705    #   to make two collection consistent.706    class_name = qualname.split(".")[0]...

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.