analysis.py
1 """Analysis module.""" 2 3 from __future__ import annotations 4 5 import sys 6 from typing import TYPE_CHECKING 7 8 from tap.tracker import Tracker 9 10 from archan.enums import ResultCode 11 from archan.logging import Logger 12 from archan.printing import PrintableNameMixin, PrintableResultMixin 13 14 if TYPE_CHECKING: 15 from archan.checkers import Checker 16 from archan.config import Config 17 from archan.providers import Provider 18 19 logger = Logger.get_logger(__name__) 20 21 22 class Analysis: 23 """Analysis class. 24 25 An instance of Analysis contains a Config object. 26 Providers are first run to generate the data, then 27 these data are all checked against every checker. 28 """ 29 30 def __init__(self, config: Config): 31 """Initialization method. 32 33 Parameters: 34 config: The configuration object to use for analysis. 35 """ 36 self.config = config 37 self.results: list[Result] = [] 38 39 @staticmethod 40 def _get_checker_result( 41 group: AnalysisGroup, 42 checker: Checker, 43 provider: Provider | None = None, 44 nd: str = "", 45 ) -> Result: 46 logger.info(f"Run {nd}checker {checker.identifier or checker.name}") 47 checker.run(provider.data if provider else None) 48 return Result(group, provider, checker, *checker.result) 49 50 def run(self, verbose: bool = True) -> None: # noqa: FBT001, FBT002 51 """Run the analysis. 52 53 Generate data from each provider, then check these data with every 54 checker, and store the analysis results. 55 56 Parameters: 57 verbose: Whether to immediately print the results or not. 58 """ 59 self.results.clear() 60 61 for analysis_group in self.config.analysis_groups: 62 if analysis_group.providers: 63 for provider in analysis_group.providers: 64 logger.info(f"Run provider {provider.identifier}") 65 provider.run() 66 for checker in analysis_group.checkers: 67 result = self._get_checker_result(analysis_group, checker, provider) 68 self.results.append(result) 69 analysis_group.results.append(result) 70 if verbose: 71 result.print() 72 else: 73 for checker in analysis_group.checkers: 74 result = self._get_checker_result(analysis_group, checker, nd="no-data-") 75 self.results.append(result) 76 analysis_group.results.append(result) 77 if verbose: 78 result.print() 79 80 def print_results(self) -> None: 81 """Print analysis results as text on standard output.""" 82 for result in self.results: 83 result.print() 84 85 def output_tap(self) -> None: 86 """Output analysis results in TAP format.""" 87 tracker = Tracker(streaming=True, stream=sys.stdout) 88 for group in self.config.analysis_groups: 89 n_providers = len(group.providers) 90 n_checkers = len(group.checkers) 91 if not group.providers and group.checkers: 92 test_suite = group.name 93 description_lambda = lambda result: result.checker.name # noqa: E731 94 elif not group.checkers: 95 logger.warning("Invalid analysis group (no checkers), skipping") 96 continue 97 elif n_providers > n_checkers: 98 test_suite = group.checkers[0].name 99 description_lambda = lambda result: result.provider.name # noqa: E731 100 else: 101 test_suite = group.providers[0].name 102 description_lambda = lambda result: result.checker.name # noqa: E731 103 104 for result in group.results: 105 description = description_lambda(result) 106 if result.code == ResultCode.PASSED: 107 tracker.add_ok(test_suite, description) 108 elif result.code == ResultCode.IGNORED: 109 tracker.add_ok(test_suite, description + " (ALLOWED FAILURE)") 110 elif result.code == ResultCode.NOT_IMPLEMENTED: 111 tracker.add_not_ok(test_suite, description, "TODO implement the test") 112 elif result.code == ResultCode.FAILED: 113 message = "\n message: ".join(result.messages.split("\n")) 114 tracker.add_not_ok( 115 test_suite, 116 description, 117 diagnostics=f" ---\n message: {message}\n hint: {result.checker.hint}\n ...", 118 ) 119 120 def output_json(self) -> None: 121 """Output analysis results in JSON format.""" 122 123 @property 124 def successful(self) -> bool: 125 """Property to tell if the run was successful: no failures. 126 127 Returns: 128 True if the run was successful. 129 """ 130 return all(result.code != ResultCode.FAILED for result in self.results) 131 132 133 class AnalysisGroup(PrintableNameMixin): 134 """Placeholder for groups of providers and checkers.""" 135 136 def __init__( 137 self, 138 name: str | None = None, 139 description: str | None = None, 140 providers: list | None = None, 141 checkers: list | None = None, 142 ): 143 """Initialization method. 144 145 Parameters: 146 name: The group name. 147 description: The group description. 148 providers: The list of providers. 149 checkers: The list of checkers. 150 """ 151 self.name = name 152 self.description = description 153 self.providers = providers or [] 154 self.checkers = checkers or [] 155 self.results: list[Result] = [] 156 157 158 class Result(PrintableResultMixin): 159 """Placeholder for analysis results.""" 160 161 def __init__(self, group: AnalysisGroup, provider: Provider, checker: Checker, code: int, messages: str): 162 """Initialization method. 163 164 Parameters: 165 group: Parent group. 166 provider: Parent Provider. 167 checker: Parent Checker. 168 code: Constant from Checker class. 169 messages: Messages string. 170 """ 171 self.group = group 172 self.provider = provider 173 self.checker = checker 174 self.code = code 175 self.messages = messages