summaryrefslogtreecommitdiffstats
path: root/test/lib/ansible_test/_internal/commands/coverage/analyze/targets/missing.py
blob: c1c77e750624aa36880b858bdccfd3d79dd801a2 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
"""Identify aggregated coverage in one file missing from another."""
from __future__ import annotations

import os
import typing as t

from .....encoding import (
    to_bytes,
)

from .....executor import (
    Delegate,
)

from .....provisioning import (
    prepare_profiles,
)

from . import (
    CoverageAnalyzeTargetsConfig,
    get_target_index,
    make_report,
    read_report,
    write_report,
)

from . import (
    TargetIndexes,
    IndexedPoints,
)


class CoverageAnalyzeTargetsMissingConfig(CoverageAnalyzeTargetsConfig):
    """Configuration for the `coverage analyze targets missing` command."""
    def __init__(self, args: t.Any) -> None:
        super().__init__(args)

        self.from_file: str = args.from_file
        self.to_file: str = args.to_file
        self.output_file: str = args.output_file

        self.only_gaps: bool = args.only_gaps
        self.only_exists: bool = args.only_exists


def command_coverage_analyze_targets_missing(args: CoverageAnalyzeTargetsMissingConfig) -> None:
    """Identify aggregated coverage in one file missing from another."""
    host_state = prepare_profiles(args)  # coverage analyze targets missing

    if args.delegate:
        raise Delegate(host_state=host_state)

    from_targets, from_path_arcs, from_path_lines = read_report(args.from_file)
    to_targets, to_path_arcs, to_path_lines = read_report(args.to_file)
    target_indexes: TargetIndexes = {}

    if args.only_gaps:
        arcs = find_gaps(from_path_arcs, from_targets, to_path_arcs, target_indexes, args.only_exists)
        lines = find_gaps(from_path_lines, from_targets, to_path_lines, target_indexes, args.only_exists)
    else:
        arcs = find_missing(from_path_arcs, from_targets, to_path_arcs, to_targets, target_indexes, args.only_exists)
        lines = find_missing(from_path_lines, from_targets, to_path_lines, to_targets, target_indexes, args.only_exists)

    report = make_report(target_indexes, arcs, lines)
    write_report(args, report, args.output_file)


def find_gaps(
    from_data: IndexedPoints,
    from_index: list[str],
    to_data: IndexedPoints,
    target_indexes: TargetIndexes,
    only_exists: bool,
) -> IndexedPoints:
    """Find gaps in coverage between the from and to data sets."""
    target_data: IndexedPoints = {}

    for from_path, from_points in from_data.items():
        if only_exists and not os.path.isfile(to_bytes(from_path)):
            continue

        to_points = to_data.get(from_path, {})

        gaps = set(from_points.keys()) - set(to_points.keys())

        if gaps:
            gap_points = dict((key, value) for key, value in from_points.items() if key in gaps)
            target_data[from_path] = dict((gap, set(get_target_index(from_index[i], target_indexes) for i in indexes)) for gap, indexes in gap_points.items())

    return target_data


def find_missing(
    from_data: IndexedPoints,
    from_index: list[str],
    to_data: IndexedPoints,
    to_index: list[str],
    target_indexes: TargetIndexes,
    only_exists: bool,
) -> IndexedPoints:
    """Find coverage in from_data not present in to_data (arcs or lines)."""
    target_data: IndexedPoints = {}

    for from_path, from_points in from_data.items():
        if only_exists and not os.path.isfile(to_bytes(from_path)):
            continue

        to_points = to_data.get(from_path, {})

        for from_point, from_target_indexes in from_points.items():
            to_target_indexes = to_points.get(from_point, set())

            remaining_targets = set(from_index[i] for i in from_target_indexes) - set(to_index[i] for i in to_target_indexes)

            if remaining_targets:
                target_index = target_data.setdefault(from_path, {}).setdefault(from_point, set())
                target_index.update(get_target_index(name, target_indexes) for name in remaining_targets)

    return target_data