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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
|
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
from __future__ import absolute_import, print_function, unicode_literals
import unittest
import datetime
import mock
import os
from mozunit import main
from taskgraph.util.parameterization import (
resolve_timestamps,
resolve_task_references,
)
class TestTimestamps(unittest.TestCase):
def test_no_change(self):
now = datetime.datetime(2018, 1, 1)
input = {
"key": "value",
"numeric": 10,
"list": ["a", True, False, None],
}
self.assertEqual(resolve_timestamps(now, input), input)
def test_buried_replacement(self):
now = datetime.datetime(2018, 1, 1)
input = {"key": [{"key2": [{"relative-datestamp": "1 day"}]}]}
self.assertEqual(
resolve_timestamps(now, input),
{"key": [{"key2": ["2018-01-02T00:00:00Z"]}]},
)
def test_appears_with_other_keys(self):
now = datetime.datetime(2018, 1, 1)
input = [{"relative-datestamp": "1 day", "another-key": True}]
self.assertEqual(
resolve_timestamps(now, input),
[{"relative-datestamp": "1 day", "another-key": True}],
)
class TestTaskRefs(unittest.TestCase):
def do(self, input, output):
taskid_for_edge_name = {"edge%d" % n: "tid%d" % n for n in range(1, 4)}
self.assertEqual(
resolve_task_references(
"subject",
input,
"tid-self",
"tid-decision",
taskid_for_edge_name,
),
output,
)
def test_no_change(self):
"resolve_task_references does nothing when there are no task references"
self.do(
{"in-a-list": ["stuff", {"property": "<edge1>"}]},
{"in-a-list": ["stuff", {"property": "<edge1>"}]},
)
def test_in_list(self):
"resolve_task_references resolves task references in a list"
self.do(
{"in-a-list": ["stuff", {"task-reference": "<edge1>"}]},
{"in-a-list": ["stuff", "tid1"]},
)
def test_in_dict(self):
"resolve_task_references resolves task references in a dict"
self.do(
{"in-a-dict": {"stuff": {"task-reference": "<edge2>"}}},
{"in-a-dict": {"stuff": "tid2"}},
)
def test_multiple(self):
"resolve_task_references resolves multiple references in the same string"
self.do(
{"multiple": {"task-reference": "stuff <edge1> stuff <edge2> after"}},
{"multiple": "stuff tid1 stuff tid2 after"},
)
def test_embedded(self):
"resolve_task_references resolves ebmedded references"
self.do(
{"embedded": {"task-reference": "stuff before <edge3> stuff after"}},
{"embedded": "stuff before tid3 stuff after"},
)
def test_escaping(self):
"resolve_task_references resolves escapes in task references"
self.do({"escape": {"task-reference": "<<><edge3>>"}}, {"escape": "<tid3>"})
def test_multikey(self):
"resolve_task_references is ignored when there is another key in the dict"
self.do(
{"escape": {"task-reference": "<edge3>", "another-key": True}},
{"escape": {"task-reference": "<edge3>", "another-key": True}},
)
def test_self(self):
"resolve_task_references resolves `self` to the provided task id"
self.do({"escape": {"task-reference": "<self>"}}, {"escape": "tid-self"})
def test_decision(self):
"resolve_task_references resolves `decision` to the provided decision task id"
self.do(
{"escape": {"task-reference": "<decision>"}}, {"escape": "tid-decision"}
)
def test_invalid(self):
"resolve_task_references raises a KeyError on reference to an invalid task"
self.assertRaisesRegexp(
KeyError,
"task 'subject' has no dependency named 'no-such'",
lambda: resolve_task_references(
"subject",
{"task-reference": "<no-such>"},
"tid-self",
"tid-decision",
{},
),
)
class TestArtifactRefs(unittest.TestCase):
def do(self, input, output):
taskid_for_edge_name = {"edge%d" % n: "tid%d" % n for n in range(1, 4)}
with mock.patch.dict(
os.environ, {"TASKCLUSTER_ROOT_URL": "https://tc-tests.localhost"}
):
self.assertEqual(
resolve_task_references(
"subject", input, "tid-self", "tid-decision", taskid_for_edge_name
),
output,
)
def test_in_list(self):
"resolve_task_references resolves artifact references in a list"
self.do(
{"in-a-list": ["stuff", {"artifact-reference": "<edge1/public/foo/bar>"}]},
{
"in-a-list": [
"stuff",
"https://tc-tests.localhost/api/queue/v1"
"/task/tid1/artifacts/public/foo/bar",
]
},
)
def test_in_dict(self):
"resolve_task_references resolves artifact references in a dict"
self.do(
{"in-a-dict": {"stuff": {"artifact-reference": "<edge2/public/bar/foo>"}}},
{
"in-a-dict": {
"stuff": "https://tc-tests.localhost/api/queue/v1"
"/task/tid2/artifacts/public/bar/foo"
}
},
)
def test_in_string(self):
"resolve_task_references resolves artifact references embedded in a string"
self.do(
{
"stuff": {
"artifact-reference": "<edge1/public/filename> and <edge2/public/bar>"
}
},
{
"stuff": "https://tc-tests.localhost/api/queue/v1"
"/task/tid1/artifacts/public/filename and "
"https://tc-tests.localhost/api/queue/v1/task/tid2/artifacts/public/bar"
},
)
def test_self(self):
"resolve_task_references raises KeyError on artifact references to `self`"
self.assertRaisesRegexp(
KeyError,
"task 'subject' can't reference artifacts of self",
lambda: resolve_task_references(
"subject",
{"artifact-reference": "<self/public/artifact>"},
"tid-self",
"tid-decision",
{},
),
)
def test_decision(self):
"resolve_task_references resolves `decision` to the provided decision task id"
self.do(
{"stuff": {"artifact-reference": "<decision/public/artifact>"}},
{
"stuff": "https://tc-tests.localhost/api/queue/v1/task/tid-decision/"
"artifacts/public/artifact"
},
)
def test_invalid(self):
"resolve_task_references raises a KeyError on reference to an invalid task"
self.assertRaisesRegexp(
KeyError,
"task 'subject' has no dependency named 'no-such'",
lambda: resolve_task_references(
"subject",
{"artifact-reference": "<no-such/public/artifact>"},
"tid-self",
"tid-decision",
{},
),
)
def test_badly_formed(self):
"resolve_task_references ignores badly-formatted artifact references"
for inv in ["<edge1>", "edge1/foo>", "<edge1>/foo", "<edge1>foo"]:
resolved = resolve_task_references(
"subject", {"artifact-reference": inv}, "tid-self", "tid-decision", {}
)
self.assertEqual(resolved, inv)
if __name__ == "__main__":
main()
|
