verify_link_in_task_graph 65%

scriptworker/cot/verify.py

@@ -96,11 +96,13 @@ class LinkOfTrust(object):
         is_try (bool): whether the task is a try task
         name (str): the name of the task (e.g., signing.decision)
         task_id (str): the taskId of the task
+        task_graph (dict): the task graph of the task, if this is a decision task
         task_type (str): the task type of the task (e.g., decision, build)
         worker_impl (str): the taskcluster worker class (e.g., docker-worker) of the task
     """
     _task = None
     _cot = None
+    _task_graph = None
     status = None
 
     def __init__(self, context, name, task_id):
@@ -124,11 +126,10 @@ def _set(self, prop_name, value):
 
     @property
     def task(self):
-        """frozendict: the task definition.
+        """dict: the task definition.
 
-        When set, the task dict is converted to a frozendict, and we also set
-        `self.decision_task_id`, `self.worker_impl`, and `self.is_try` based
-        on the task definition.
+        When set, we also set `self.decision_task_id`, `self.worker_impl`,
+        and `self.is_try` based on the task definition.
         """
         return self._task
 
@@ -141,16 +142,24 @@ def task(self, task):
 
     @property
     def cot(self):
-        """frozendict: the chain of trust json body.
-
-        When set, the chain of trust dict is converted to a frozendict.
+        """dict: the chain of trust json body.
         """
         return self._cot
 
     @cot.setter
     def cot(self, cot):
         self._set('_cot', cot)
 
+    @property
+    def task_graph(self):
+        """dict: the decision task graph, if this is a decision task.
+        """
+        return self._task_graph
+
+    @task_graph.setter
+    def task_graph(self, task_graph):
+        self._set('_task_graph', task_graph)
+
 
 # raise_on_errors {{{1
 def raise_on_errors(errors):

scriptworker/test/test_cot_verify.py

@@ -2,6 +2,7 @@
 # coding=utf-8
 """Test scriptworker.cot.verify
 """
+from copy import deepcopy
 from frozendict import frozendict
 import json
 import logging
@@ -71,8 +72,18 @@ def build_link(chain):
         'provisionerId': 'provisioner',
         'workerType': 'workerType',
         'scopes': [],
+        'dependencies': [],
         'metadata': {},
         'payload': {
+            'artifacts': {
+                'foo': {
+                    'sha256': "foo_sha",
+                    'expires': "blah",
+                },
+                'bar': {
+                    'sha256': "bar_sha",
+                },
+            },
             'image': {
                 'taskId': 'docker_image_task_id',
                 'path': 'path/image',
@@ -568,3 +579,29 @@ def fake_body(*args, **kwargs):
     assert os.path.exists(path)
     with open(path, "r") as fh:
         assert json.load(fh) == {}
+
+
+# verify_link_in_task_graph {{{1
+def test_verify_link_in_task_graph(chain, decision_link, build_link):
+    chain.links = [decision_link, build_link]
+    decision_link.task_graph = {
+        build_link.task_id: {
+            'task': deepcopy(build_link.task)
+        },
+    }
+    cotverify.verify_link_in_task_graph(chain, decision_link, build_link)
+
+
+def test_verify_link_in_task_graph_exception(chain, decision_link, build_link):
+    chain.links = [decision_link, build_link]
+    bad_task = deepcopy(build_link.task)
+    bad_task['dependencies'].append("foo")
+    bad_task['x'] = 'y'
+    build_link.task['x'] = 'z'
+    decision_link.task_graph = {
+        build_link.task_id: {
+            'task': bad_task
+        },
+    }
+    with pytest.raises(CoTError):
+        cotverify.verify_link_in_task_graph(chain, decision_link, build_link)