DXR is a code search and navigation tool aimed at making sense of large projects. It supports full-text and regex searches as well as structural queries.

Mercurial (1aeaa33a64f9)

VCS Links

Line Code
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
# 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, 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_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>'}, {})
        )


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, 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_invalid(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}, {})
            self.assertEqual(resolved, inv)


if __name__ == '__main__':
    main()