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

How to use discover_running_cases method in lisa

Best Python code snippet using lisa_python

legacy_runner.py

Source:legacy_runner.py

...449        if all_cases:450            # expand for test matrix451            all_cases = all_cases * int(count / len(all_cases))452        return all_cases453    def discover_running_cases(self) -> List[Dict[str, str]]:454        cases: List[Dict[str, str]] = []455        for line in self._line_iter():456            case_match = self.CASE_RUNNING.match(line)457            if case_match:458                name = case_match["name"]459                current_case: Dict[str, str] = {460                    key: value for key, value in case_match.groupdict().items() if value461                }462            image_match = self.CASE_IMAGE_LOCATION.match(line)463            location = ""464            if image_match:465                location = image_match["location"]466                current_case.update(467                    {468                        key: value469                        for key, value in image_match.groupdict().items()470                        if value471                    }472                )473                # marketplace_image for ARMImage, vhd_image for OsVHD in legacy run474                if "marketplace_image" in current_case.keys():475                    current_case["image"] = current_case.pop("marketplace_image")476                elif "vhd_image" in current_case.keys():477                    current_case["image"] = current_case.pop("vhd_image")478                else:479                    raise LisaException(480                        "Can't get ARMImage or OsVHD from legacy run "481                        "when parsing running cases"482                    )483                # In sequence run, there is no vm size log line.484                # So, when image and location is found, the case can be added.485                cases.append(current_case)486            vmsize_match = self.CASE_VMSIZE.match(line)487            if vmsize_match:488                temp_name = vmsize_match["name"]489                temp_location = vmsize_match["location"]490                assert name == temp_name, (491                    f"cannot match location between logs. "492                    f"current case is: '{name}', "493                    f"name in vmsize is: '{temp_name}'. {line}"494                )495                if location:496                    assert location == temp_location, (497                        f"cannot match location between logs. "498                        f"setup config is: '{location}', "499                        f"location in vmsize is: '{temp_location}'. {line}"500                    )501                current_case.update(502                    {503                        key: value504                        for key, value in vmsize_match.groupdict().items()505                        if value506                    }507                )508        return cases509    def discover_completed_cases(self) -> List[Dict[str, str]]:510        cases: List[Dict[str, str]] = []511        for line in self._line_iter():512            case_match = self.CASE_COMPLETED.match(line)513            if case_match:514                current_case = {515                    key: value for key, value in case_match.groupdict().items() if value516                }517                # marketplace_image for ARMImage, vhd_image for OsVHD in legacy run518                if "marketplace_image" in current_case.keys():519                    current_case["image"] = current_case.pop("marketplace_image")520                elif "vhd_image" in current_case.keys():521                    current_case["image"] = current_case.pop("vhd_image")522                else:523                    raise LisaException(524                        "Can't get ARMImage or OsVHD from legacy run "525                        "when parsing completed cases"526                    )527                cases.append(current_case)528        return cases529    @retry(tries=30, jitter=(1, 2))530    def _read_log(self) -> str:531        """532        V2 opens log file frequently to write content, copying may be failed due to533        conflict. So retry to make it more stable.534        """535        # refer from http://thepythoncorner.com/dev/how-to-open-file-without-locking-it/536        # that's cool!537        handle = win32file.CreateFile(538            self._runner_log_path,539            win32file.GENERIC_READ,540            win32file.FILE_SHARE_DELETE541            | win32file.FILE_SHARE_READ542            | win32file.FILE_SHARE_WRITE,543            None,544            win32file.OPEN_EXISTING,545            0,546            None,547        )548        # detach the handle549        detached_handle = handle.Detach()550        content = ""551        # get a file descriptor associated to the handle552        if not TYPE_CHECKING:  # FIXME: if you have a better solution553            # for mypy checks on Linux, change this554            file_descriptor = msvcrt.open_osfhandle(detached_handle, os.O_RDONLY)555            # open the file descriptor556            with open(file_descriptor) as file:557                content = file.read()558        return content559    def _line_iter(self) -> Iterable[str]:560        content = self._read_log()561        iterator = self.LOG_LINE.finditer(content)562        for match in iterator:563            yield match["message"]564def _find_matched_files(565    working_dir: Path, pattern: Pattern[str], log: Logger566) -> List[LogParser]:567    """568    return file parsers for matched files569    """570    log_file_pattern = str(working_dir / "TestResults/**/*.log")571    results: List[LogParser] = []572    for file_path in glob.glob(log_file_pattern, recursive=True):573        matched = pattern.findall(file_path)574        if matched:575            results.append(LogParser(file_path, log))576    return results577def _track_progress(578    process: Process, working_dir: Path, log: Logger, runner: LegacyRunner, id_: str579) -> None:580    # discovered all cases581    all_cases: List[Dict[str, str]] = []582    process_exiting: bool = False583    while True:584        check_cancelled()585        root_parsers = _find_matched_files(586            working_dir=working_dir, pattern=ROOT_LOG_FILE_PATTERN, log=log587        )588        assert len(root_parsers) <= 1, "found multiple root parsers. It's unexpected."589        if root_parsers:590            root_parser = root_parsers[0]591            all_cases = root_parser.discover_cases()592        # check if any case is running, it means all cases are collected593        running_parsers = _find_matched_files(594            working_dir=working_dir, pattern=LOG_FILE_PATTERN, log=log595        )596        if any(parser.discover_running_cases() for parser in running_parsers):597            log.info(f"found {len(all_cases)} cases: {[x['name'] for x in all_cases]}")598            break599        # try one more time, after process is exited.600        if not process.is_running():601            if process_exiting:602                break603            process_exiting = True604        time.sleep(5)605    process_exiting = False606    case_states = ResultStateManager(id_=id_, log=log)607    # loop to check running and completed results608    while True:609        check_cancelled()610        running_cases: List[Dict[str, str]] = []611        completed_cases: List[Dict[str, str]] = []612        # discover running cases613        running_parsers = _find_matched_files(614            working_dir=working_dir, pattern=LOG_FILE_PATTERN, log=log615        )616        for parser in running_parsers:617            running_cases.extend(parser.discover_running_cases())618        # discover completed cases619        completed_parsers = _find_matched_files(620            working_dir=working_dir, pattern=CASE_LOG_FILE_PATTERN, log=log621        )622        for parser in completed_parsers:623            completed_cases.extend(parser.discover_completed_cases())624        # merge all dict to result status625        case_states.set_states(626            all_cases=all_cases,627            running_cases=running_cases,628            completed_cases=completed_cases,629        )630        # try one more time, after process is exited.631        if not process.is_running():...

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.