Best Python code snippet using lisa_python
legacy_runner.py
Source:legacy_runner.py
...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():632 if process_exiting:633 break634 process_exiting = True635 time.sleep(5)636 # Handle cases which aborted in deployment stage637 for result in case_states.results:...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!