summaryrefslogtreecommitdiffstats
path: root/taskcluster/taskgraph/test/test_taskgraph.py
blob: 67966ce61fa92571d44a417151ec4db71cd583f8 (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
# 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

from taskgraph.graph import Graph
from taskgraph.task import Task
from taskgraph.taskgraph import TaskGraph
from mozunit import main


class TestTaskGraph(unittest.TestCase):

    maxDiff = None

    def test_taskgraph_to_json(self):
        tasks = {
            "a": Task(
                kind="test",
                label="a",
                description="Task A",
                attributes={"attr": "a-task"},
                task={"taskdef": True},
            ),
            "b": Task(
                kind="test",
                label="b",
                attributes={},
                task={"task": "def"},
                optimization={"skip-unless-has-relevant-tests": None},
                # note that this dep is ignored, superseded by that
                # from the taskgraph's edges
                dependencies={"first": "a"},
            ),
        }
        graph = Graph(nodes=set("ab"), edges={("a", "b", "edgelabel")})
        taskgraph = TaskGraph(tasks, graph)

        res = taskgraph.to_json()

        self.assertEqual(
            res,
            {
                "a": {
                    "kind": "test",
                    "label": "a",
                    "description": "Task A",
                    "attributes": {"attr": "a-task", "kind": "test"},
                    "task": {"taskdef": True},
                    "dependencies": {"edgelabel": "b"},
                    "soft_dependencies": [],
                    "if_dependencies": [],
                    "optimization": None,
                },
                "b": {
                    "kind": "test",
                    "label": "b",
                    "description": "",
                    "attributes": {"kind": "test"},
                    "task": {"task": "def"},
                    "dependencies": {},
                    "soft_dependencies": [],
                    "if_dependencies": [],
                    "optimization": {"skip-unless-has-relevant-tests": None},
                },
            },
        )

    def test_round_trip(self):
        graph = TaskGraph(
            tasks={
                "a": Task(
                    kind="fancy",
                    label="a",
                    description="Task A",
                    attributes={},
                    dependencies={"prereq": "b"},  # must match edges, below
                    optimization={"skip-unless-has-relevant-tests": None},
                    task={"task": "def"},
                ),
                "b": Task(
                    kind="pre",
                    label="b",
                    attributes={},
                    dependencies={},
                    optimization={"skip-unless-has-relevant-tests": None},
                    task={"task": "def2"},
                ),
            },
            graph=Graph(nodes={"a", "b"}, edges={("a", "b", "prereq")}),
        )

        tasks, new_graph = TaskGraph.from_json(graph.to_json())
        self.assertEqual(graph, new_graph)

    simple_graph = TaskGraph(
        tasks={
            "a": Task(
                kind="fancy",
                label="a",
                attributes={},
                dependencies={"prereq": "b"},  # must match edges, below
                optimization={"skip-unless-has-relevant-tests": None},
                task={"task": "def"},
            ),
            "b": Task(
                kind="pre",
                label="b",
                attributes={},
                dependencies={},
                optimization={"skip-unless-has-relevant-tests": None},
                task={"task": "def2"},
            ),
        },
        graph=Graph(nodes={"a", "b"}, edges={("a", "b", "prereq")}),
    )

    def test_contains(self):
        assert "a" in self.simple_graph
        assert "c" not in self.simple_graph


if __name__ == "__main__":
    main()