diff options
Diffstat (limited to 'test/lib/ansible_test/_internal/commands/coverage/analyze/targets/combine.py')
-rw-r--r-- | test/lib/ansible_test/_internal/commands/coverage/analyze/targets/combine.py | 74 |
1 files changed, 74 insertions, 0 deletions
diff --git a/test/lib/ansible_test/_internal/commands/coverage/analyze/targets/combine.py b/test/lib/ansible_test/_internal/commands/coverage/analyze/targets/combine.py new file mode 100644 index 0000000..e3782ce --- /dev/null +++ b/test/lib/ansible_test/_internal/commands/coverage/analyze/targets/combine.py @@ -0,0 +1,74 @@ +"""Combine integration test target code coverage reports.""" +from __future__ import annotations +import typing as t + +from .....executor import ( + Delegate, +) + +from .....provisioning import ( + prepare_profiles, +) + +from . import ( + CoverageAnalyzeTargetsConfig, + get_target_index, + make_report, + read_report, + write_report, +) + +from . import ( + Arcs, + IndexedPoints, + Lines, + TargetIndexes, +) + + +class CoverageAnalyzeTargetsCombineConfig(CoverageAnalyzeTargetsConfig): + """Configuration for the `coverage analyze targets combine` command.""" + def __init__(self, args: t.Any) -> None: + super().__init__(args) + + self.input_files: list[str] = args.input_file + self.output_file: str = args.output_file + + +def command_coverage_analyze_targets_combine(args: CoverageAnalyzeTargetsCombineConfig) -> None: + """Combine integration test target code coverage reports.""" + host_state = prepare_profiles(args) # coverage analyze targets combine + + if args.delegate: + raise Delegate(host_state=host_state) + + combined_target_indexes: TargetIndexes = {} + combined_path_arcs: Arcs = {} + combined_path_lines: Lines = {} + + for report_path in args.input_files: + covered_targets, covered_path_arcs, covered_path_lines = read_report(report_path) + + merge_indexes(covered_path_arcs, covered_targets, combined_path_arcs, combined_target_indexes) + merge_indexes(covered_path_lines, covered_targets, combined_path_lines, combined_target_indexes) + + report = make_report(combined_target_indexes, combined_path_arcs, combined_path_lines) + + write_report(args, report, args.output_file) + + +def merge_indexes( + source_data: IndexedPoints, + source_index: list[str], + combined_data: IndexedPoints, + combined_index: TargetIndexes, +) -> None: + """Merge indexes from the source into the combined data set (arcs or lines).""" + for covered_path, covered_points in source_data.items(): + combined_points = combined_data.setdefault(covered_path, {}) + + for covered_point, covered_target_indexes in covered_points.items(): + combined_point = combined_points.setdefault(covered_point, set()) + + for covered_target_index in covered_target_indexes: + combined_point.add(get_target_index(source_index[covered_target_index], combined_index)) |