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
|
# 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/.
"""
Utilities for generating a decision task from :file:`.taskcluster.yml`.
"""
import os
import jsone
import slugid
import yaml
from .templates import merge
from .time import current_json_time
from .vcs import find_hg_revision_push_info
def make_decision_task(params, root, context, head_rev=None):
"""Generate a basic decision task, based on the root .taskcluster.yml"""
with open(os.path.join(root, ".taskcluster.yml"), "rb") as f:
taskcluster_yml = yaml.safe_load(f)
if not head_rev:
head_rev = params["head_rev"]
if params["repository_type"] == "hg":
pushlog = find_hg_revision_push_info(params["repository_url"], head_rev)
hg_push_context = {
"pushlog_id": pushlog["pushid"],
"pushdate": pushlog["pushdate"],
"owner": pushlog["user"],
}
else:
hg_push_context = {}
slugids = {}
def as_slugid(name):
# https://github.com/taskcluster/json-e/issues/164
name = name[0]
if name not in slugids:
slugids[name] = slugid.nice()
return slugids[name]
# provide a similar JSON-e context to what mozilla-taskcluster provides:
# https://docs.taskcluster.net/reference/integrations/mozilla-taskcluster/docs/taskcluster-yml
# but with a different tasks_for and an extra `cron` section
context = merge(
{
"repository": {
"url": params["repository_url"],
"project": params["project"],
"level": params["level"],
},
"push": merge(
{
"revision": params["head_rev"],
# remainder are fake values, but the decision task expects them anyway
"comment": " ",
},
hg_push_context,
),
"now": current_json_time(),
"as_slugid": as_slugid,
},
context,
)
rendered = jsone.render(taskcluster_yml, context)
if len(rendered["tasks"]) != 1:
raise Exception("Expected .taskcluster.yml to only produce one cron task")
task = rendered["tasks"][0]
task_id = task.pop("taskId")
return (task_id, task)
|
