Coverage for codexa/client/executor.py: 71%

1import io 

2import logging 

3from collections import defaultdict 

4from contextlib import redirect_stderr, redirect_stdout 

5from dataclasses import asdict, dataclass 

6from pathlib import Path 

7from typing import Any, Dict, List, Optional, Tuple 

8 

9import pytest 

10from _pytest.reports import CollectReport 

11 

12logger = logging.getLogger(__name__) 

13 

14 

15TestMap = Dict[str, Dict[str, List[str]]] 

16 

17 

18@dataclass(frozen=True) 

19class TestcaseMetadata: 

20 """Test case metadata model.""" 

21 

22 node_id: str 

23 name: str 

24 file: str 

25 line_number: int 

26 keywords: List[str] 

27 module: Optional[str] = None 

28 cls: Optional[str] = None 

29 function: Optional[str] = None 

30 

31 def to_dict(self) -> Dict[str, Any]: 

32 """Convert to dictionary map structure.""" 

33 return asdict(self) 

34 

35 

36class TestExecutor: 

37 """Class for executing tests.""" 

38 

39 def __init__(self, test_ids: List[str]): 

40 self.__test_ids = test_ids 

41 

42 def run(self, verbose: bool = False) -> Tuple[int, str, str]: 

43 """Run the Pytest tests located at the specified path. 

44 

45 Args: 

46 path (str): _description_ 

47 

48 Returns: 

49 Tuple[str, str, int]: _description_ 

50 """ 

51 stdout = io.StringIO() 

52 stderr = io.StringIO() 

53 

54 pytest_base_args = [] 

55 if verbose: 

56 pytest_base_args.append("-vv") 

57 

58 # Redirect both stdout and stderr during the test run 

59 with redirect_stdout(stdout), redirect_stderr(stderr): 

60 exit_code = pytest.main([*pytest_base_args, *self.__test_ids]) 

61 

62 # Retrieve output 

63 shell_output = stdout.getvalue() 

64 error_output = stderr.getvalue() 

65 

66 return exit_code, shell_output, error_output 

67 

68 @staticmethod 

69 def __collect_ids(paths: List[str]) -> List[TestcaseMetadata]: 

70 args = ["--collect-only", "-q", "-p", "no:warnings"] 

71 if paths: 

72 args.extend([str(Path(p).resolve()) for p in paths]) 

73 

74 # Collect test data 

75 collected = [] 

76 

77 class CollectorPlugin: 

78 def pytest_collectreport(self, report: CollectReport): 

79 if report.failed: 

80 logger.error(report.longrepr) 

81 

82 def pytest_itemcollected(self, item: Any): 

83 file, line_number, _ = item.location 

84 entry = TestcaseMetadata( 

85 node_id=item.nodeid, 

86 name=item.name, 

87 file=file, 

88 line_number=line_number, 

89 keywords=list(item.keywords), 

90 module=item.module.__name__ if item.module else None, 

91 cls=item.cls.__name__ if item.cls else None, 

92 function=getattr(item.function, "__name__", None), 

93 ) 

94 collected.append(entry) 

95 

96 stdout = io.StringIO() 

97 stderr = io.StringIO() 

98 with redirect_stdout(stdout), redirect_stderr(stderr): 

99 pytest.main(args, plugins=[CollectorPlugin()]) 

100 return collected 

101 

102 def collect_all_tests(self) -> List[str]: 

103 """Collect all available Pytest node IDs. 

104 

105 Args: 

106 paths (List[str], optional): List of base paths, defaults to None. 

107 

108 Returns: 

109 List[str]: List of executable node IDs 

110 """ 

111 entries = self.__collect_ids(self.__test_ids) 

112 return [entry.node_id for entry in entries] 

113 

114 def get_structured_test_tree(self) -> Dict[str, Any]: 

115 """Generate a mapping of available tests. 

116 

117 Args: 

118 paths (List[str], optional): List of base paths, defaults to None. 

119 

120 Returns: 

121 TestTree: Test entries map 

122 """ 

123 entries = self.__collect_ids(self.__test_ids) 

124 test_map = defaultdict(lambda: defaultdict(list)) 

125 

126 for entry in entries: 

127 module_path = entry.file 

128 class_name = entry.cls 

129 func_name = entry.name 

130 

131 test_map[module_path][class_name].append(func_name) 

132 

133 return test_map