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
|
"""Test metadata for passing data to delegated tests."""
from __future__ import annotations
import typing as t
from .util import (
display,
generate_name,
)
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) -> None:
"""Initialize metadata."""
self.changes: dict[str, tuple[tuple[int, int], ...]] = {}
self.cloud_config: t.Optional[dict[str, dict[str, t.Union[int, str, bool]]]] = None
self.change_description: t.Optional[ChangeDescription] = None
self.ci_provider: t.Optional[str] = None
self.session_id = generate_name()
def populate_changes(self, diff: t.Optional[list[str]]) -> None:
"""Populate the changeset using the given diff."""
patches = parse_diff(diff)
patches: list[FileDiff] = sorted(patches, key=lambda k: k.new.path)
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) -> dict[str, t.Any]:
"""Return a dictionary representation of the metadata."""
return dict(
changes=self.changes,
cloud_config=self.cloud_config,
ci_provider=self.ci_provider,
change_description=self.change_description.to_dict(),
session_id=self.session_id,
)
def to_file(self, path: str) -> None:
"""Write the metadata to the specified file."""
data = self.to_dict()
display.info('>>> Metadata: %s\n%s' % (path, data), verbosity=3)
write_json_file(path, data)
@staticmethod
def from_file(path: str) -> Metadata:
"""Return metadata loaded from the specified file."""
data = read_json_file(path)
return Metadata.from_dict(data)
@staticmethod
def from_dict(data: dict[str, t.Any]) -> Metadata:
"""Return metadata loaded from the specified dictionary."""
metadata = Metadata()
metadata.changes = data['changes']
metadata.cloud_config = data['cloud_config']
metadata.ci_provider = data['ci_provider']
metadata.change_description = ChangeDescription.from_dict(data['change_description'])
metadata.session_id = data['session_id']
return metadata
class ChangeDescription:
"""Description of changes."""
def __init__(self) -> None:
self.command: str = ''
self.changed_paths: list[str] = []
self.deleted_paths: list[str] = []
self.regular_command_targets: dict[str, list[str]] = {}
self.focused_command_targets: dict[str, list[str]] = {}
self.no_integration_paths: list[str] = []
@property
def targets(self) -> t.Optional[list[str]]:
"""Optional list of target names."""
return self.regular_command_targets.get(self.command)
@property
def focused_targets(self) -> t.Optional[list[str]]:
"""Optional list of focused target names."""
return self.focused_command_targets.get(self.command)
def to_dict(self) -> dict[str, t.Any]:
"""Return a dictionary representation of the change description."""
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: dict[str, t.Any]) -> ChangeDescription:
"""Return a change description loaded from the given dictionary."""
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
|
