summaryrefslogtreecommitdiff
path: root/test/lib/ansible_test/_internal/metadata.py
blob: 389836e634e4a3df3474c0a6db5c072acd605615 (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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
"""Test metadata for passing data to delegated tests."""
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type

from . import types as t

from .util import (
    display,
    is_shippable,
)

from .io import (
    write_json_file,
    read_json_file,
)

from .diff import (
    parse_diff,
    FileDiff,
)


class Metadata:
    """Metadata object for passing data to delegated tests."""
    def __init__(self):
        """Initialize metadata."""
        self.changes = {}  # type: t.Dict[str, t.Tuple[t.Tuple[int, int]]]
        self.cloud_config = None  # type: t.Optional[t.Dict[str, str]]
        self.instance_config = None  # type: t.Optional[t.List[t.Dict[str, str]]]
        self.change_description = None  # type: t.Optional[ChangeDescription]

        if is_shippable():
            self.ci_provider = 'shippable'
        else:
            self.ci_provider = ''

    def populate_changes(self, diff):
        """
        :type diff: list[str] | None
        """
        patches = parse_diff(diff)
        patches = sorted(patches, key=lambda k: k.new.path)  # type: t.List[FileDiff]

        self.changes = dict((patch.new.path, tuple(patch.new.ranges)) for patch in patches)

        renames = [patch.old.path for patch in patches if patch.old.path != patch.new.path and patch.old.exists and patch.new.exists]
        deletes = [patch.old.path for patch in patches if not patch.new.exists]

        # make sure old paths which were renamed or deleted are registered in changes
        for path in renames + deletes:
            if path in self.changes:
                # old path was replaced with another file
                continue

            # failed tests involving deleted files should be using line 0 since there is no content remaining
            self.changes[path] = ((0, 0),)

    def to_dict(self):
        """
        :rtype: dict[str, any]
        """
        return dict(
            changes=self.changes,
            cloud_config=self.cloud_config,
            instance_config=self.instance_config,
            ci_provider=self.ci_provider,
            change_description=self.change_description.to_dict(),
        )

    def to_file(self, path):
        """
        :type path: path
        """
        data = self.to_dict()

        display.info('>>> Metadata: %s\n%s' % (path, data), verbosity=3)

        write_json_file(path, data)

    @staticmethod
    def from_file(path):
        """
        :type path: str
        :rtype: Metadata
        """
        data = read_json_file(path)
        return Metadata.from_dict(data)

    @staticmethod
    def from_dict(data):
        """
        :type data: dict[str, any]
        :rtype: Metadata
        """
        metadata = Metadata()
        metadata.changes = data['changes']
        metadata.cloud_config = data['cloud_config']
        metadata.instance_config = data['instance_config']
        metadata.ci_provider = data['ci_provider']
        metadata.change_description = ChangeDescription.from_dict(data['change_description'])

        return metadata


class ChangeDescription:
    """Description of changes."""
    def __init__(self):
        self.command = ''  # type: str
        self.changed_paths = []  # type: t.List[str]
        self.deleted_paths = []  # type: t.List[str]
        self.regular_command_targets = {}  # type: t.Dict[str, t.List[str]]
        self.focused_command_targets = {}  # type: t.Dict[str, t.List[str]]
        self.no_integration_paths = []  # type: t.List[str]

    @property
    def targets(self):
        """
        :rtype: list[str] | None
        """
        return self.regular_command_targets.get(self.command)

    @property
    def focused_targets(self):
        """
        :rtype: list[str] | None
        """
        return self.focused_command_targets.get(self.command)

    def to_dict(self):
        """
        :rtype: dict[str, any]
        """
        return dict(
            command=self.command,
            changed_paths=self.changed_paths,
            deleted_paths=self.deleted_paths,
            regular_command_targets=self.regular_command_targets,
            focused_command_targets=self.focused_command_targets,
            no_integration_paths=self.no_integration_paths,
        )

    @staticmethod
    def from_dict(data):
        """
        :param data: dict[str, any]
        :rtype: ChangeDescription
        """
        changes = ChangeDescription()
        changes.command = data['command']
        changes.changed_paths = data['changed_paths']
        changes.deleted_paths = data['deleted_paths']
        changes.regular_command_targets = data['regular_command_targets']
        changes.focused_command_targets = data['focused_command_targets']
        changes.no_integration_paths = data['no_integration_paths']

        return changes