Quelle test_target_tasks.py Sprache: Python

# 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/.

import contextlib
import re
import unittest

import pytest
from mozunit import main
from taskgraph.graph import Graph
from taskgraph.target_tasks import get_method
from taskgraph.task import Task
from taskgraph.taskgraph import TaskGraph

from gecko_taskgraph import target_tasks, try_option_syntax

class FakeTryOptionSyntax:
    def __init__(self, message, task_graph, graph_config):
        self.trigger_tests = 0
        self.talos_trigger_tests = 0
        self.raptor_trigger_tests = 0
        self.notifications = None
        self.env = []
        self.profile = False
        self.tag = None
        self.no_retry = False

    def task_matches(self, task):
        return "at-at" in task.attributes

class TestTargetTasks(unittest.TestCase):
    def default_matches_project(self, run_on_projects, project):
        return self.default_matches(
            attributes={
                "run_on_projects": run_on_projects,
            },
            parameters={
                "project": project,
                "hg_branch": "default",
            },
        )

    def default_matches_hg_branch(self, run_on_hg_branches, hg_branch):
        attributes = {"run_on_projects": ["all"]}
        if run_on_hg_branches is not None:
            attributes["run_on_hg_branches"] = run_on_hg_branches

        return self.default_matches(
            attributes=attributes,
            parameters={
                "project": "mozilla-central",
                "hg_branch": hg_branch,
            },
        )

    def default_matches(self, attributes, parameters):
        method = get_method("default")
        graph = TaskGraph(
            tasks={
                "a": Task(kind="build", label="a", attributes=attributes, task={}),
            },
            graph=Graph(nodes={"a"}, edges=set()),
        )
        return "a" in method(graph, parameters, {})

    def test_default_all(self):
        """run_on_projects=[all] includes release, integration, and other projects"""
        self.assertTrue(self.default_matches_project(["all"], "mozilla-central"))
        self.assertTrue(self.default_matches_project(["all"], "baobab"))

    def test_default_integration(self):
        """run_on_projects=[integration] includes integration projects"""
        self.assertFalse(
            self.default_matches_project(["integration"], "mozilla-central")
        )
        self.assertFalse(self.default_matches_project(["integration"], "baobab"))

    def test_default_release(self):
        """run_on_projects=[release] includes release projects"""
        self.assertTrue(self.default_matches_project(["release"], "mozilla-central"))
        self.assertFalse(self.default_matches_project(["release"], "baobab"))

    def test_default_nothing(self):
        """run_on_projects=[] includes nothing"""
        self.assertFalse(self.default_matches_project([], "mozilla-central"))
        self.assertFalse(self.default_matches_project([], "baobab"))

    def test_default_hg_branch(self):
        self.assertTrue(self.default_matches_hg_branch(None, "default"))
        self.assertTrue(self.default_matches_hg_branch(None, "GECKOVIEW_62_RELBRANCH"))

        self.assertFalse(self.default_matches_hg_branch([], "default"))
        self.assertFalse(self.default_matches_hg_branch([], "GECKOVIEW_62_RELBRANCH"))

        self.assertTrue(self.default_matches_hg_branch(["all"], "default"))
        self.assertTrue(
            self.default_matches_hg_branch(["all"], "GECKOVIEW_62_RELBRANCH")
        )

        self.assertTrue(self.default_matches_hg_branch(["default"], "default"))
        self.assertTrue(self.default_matches_hg_branch([r"default"], "default"))
        self.assertFalse(
            self.default_matches_hg_branch([r"default"], "GECKOVIEW_62_RELBRANCH")
        )

        self.assertTrue(
            self.default_matches_hg_branch(
                ["GECKOVIEW_62_RELBRANCH"], "GECKOVIEW_62_RELBRANCH"
            )
        )
        self.assertTrue(
            self.default_matches_hg_branch(
                [r"GECKOVIEW_\d+_RELBRANCH"], "GECKOVIEW_62_RELBRANCH"
            )
        )
        self.assertTrue(
            self.default_matches_hg_branch(
                [r"GECKOVIEW_\d+_RELBRANCH"], "GECKOVIEW_62_RELBRANCH"
            )
        )
        self.assertFalse(
            self.default_matches_hg_branch([r"GECKOVIEW_\d+_RELBRANCH"], "default")
        )

    def make_task_graph(self):
        tasks = {
            "a": Task(kind=None, label="a", attributes={}, task={}),
            "b": Task(kind=None, label="b", attributes={"at-at": "yep"}, task={}),
            "c": Task(
                kind=None, label="c", attributes={"run_on_projects": ["try"]}, task={}
            ),
            "ddd-1": Task(kind="test", label="ddd-1", attributes={}, task={}),
            "ddd-2": Task(kind="test", label="ddd-2", attributes={}, task={}),
            "ddd-1-cf": Task(kind="test", label="ddd-1-cf", attributes={}, task={}),
            "ddd-2-cf": Task(kind="test", label="ddd-2-cf", attributes={}, task={}),
            "ddd-var-1": Task(kind="test", label="ddd-var-1", attributes={}, task={}),
            "ddd-var-2": Task(kind="test", label="ddd-var-2", attributes={}, task={}),
        }
        graph = Graph(
            nodes=set(
                [
                    "a",
                    "b",
                    "c",
                    "ddd-1",
                    "ddd-2",
                    "ddd-1-cf",
                    "ddd-2-cf",
                    "ddd-var-1",
                    "ddd-var-2",
                ]
            ),
            edges=set(),
        )
        return TaskGraph(tasks, graph)

    @contextlib.contextmanager
    def fake_TryOptionSyntax(self):
        orig_TryOptionSyntax = try_option_syntax.TryOptionSyntax
        try:
            try_option_syntax.TryOptionSyntax = FakeTryOptionSyntax
            yield
        finally:
            try_option_syntax.TryOptionSyntax = orig_TryOptionSyntax

    def test_empty_try(self):
        "try_mode = None runs nothing"
        tg = self.make_task_graph()
        method = get_method("try_tasks")
        params = {
            "try_mode": None,
            "project": "try",
            "message": "",
        }
        # only runs the task with run_on_projects: try
        self.assertEqual(method(tg, params, {}), [])

    def test_try_option_syntax(self):
        "try_mode = try_option_syntax uses TryOptionSyntax"
        tg = self.make_task_graph()
        method = get_method("try_tasks")
        with self.fake_TryOptionSyntax():
            params = {
                "try_mode": "try_option_syntax",
                "message": "try: -p all",
            }
            self.assertEqual(method(tg, params, {}), ["b"])

    def test_try_task_config(self):
        "try_mode = try_task_config uses the try config"
        tg = self.make_task_graph()
        method = get_method("try_tasks")
        params = {
            "try_mode": "try_task_config",
            "try_task_config": {"tasks": ["a"]},
        }
        self.assertEqual(method(tg, params, {}), ["a"])

    def test_try_task_config_regex(self):
        "try_mode = try_task_config uses the try config with regex instead of chunk numbers"
        tg = self.make_task_graph()
        method = get_method("try_tasks")
        params = {
            "try_mode": "try_task_config",
            "try_task_config": {"new-test-config": True, "tasks": ["ddd-*"]},
            "project": "try",
        }
        self.assertEqual(sorted(method(tg, params, {})), ["ddd-1", "ddd-2"])

    def test_try_task_config_regex_with_paths(self):
        "try_mode = try_task_config uses the try config with regex instead of chunk numbers"
        tg = self.make_task_graph()
        method = get_method("try_tasks")
        params = {
            "try_mode": "try_task_config",
            "try_task_config": {
                "new-test-config": True,
                "tasks": ["ddd-*"],
                "env": {"MOZHARNESS_TEST_PATHS": "foo/bar"},
            },
            "project": "try",
        }
        self.assertEqual(sorted(method(tg, params, {})), ["ddd-1"])

    def test_try_task_config_absolute(self):
        "try_mode = try_task_config uses the try config with full task labels"
        tg = self.make_task_graph()
        method = get_method("try_tasks")
        params = {
            "try_mode": "try_task_config",
            "try_task_config": {
                "new-test-config": True,
                "tasks": ["ddd-var-2", "ddd-1"],
            },
            "project": "try",
        }
        self.assertEqual(sorted(method(tg, params, {})), ["ddd-1", "ddd-var-2"])

    def test_try_task_config_regex_var(self):
        "try_mode = try_task_config uses the try config with regex instead of chunk numbers and a test variant"
        tg = self.make_task_graph()
        method = get_method("try_tasks")
        params = {
            "try_mode": "try_task_config",
            "try_task_config": {"new-test-config": True, "tasks": ["ddd-var-*"]},
            "project": "try",
        }
        self.assertEqual(sorted(method(tg, params, {})), ["ddd-var-1", "ddd-var-2"])

# tests for specific filters

@pytest.mark.parametrize(
    "name,params,expected",
    (
        pytest.param(
            "filter_tests_without_manifests",
            {
                "task": Task(kind="test", label="a", attributes={}, task={}),
                "parameters": None,
            },
            True,
            id="filter_tests_without_manifests_not_in_attributes",
        ),
        pytest.param(
            "filter_tests_without_manifests",
            {
                "task": Task(
                    kind="test",
                    label="a",
                    attributes={"test_manifests": ["foo"]},
                    task={},
                ),
                "parameters": None,
            },
            True,
            id="filter_tests_without_manifests_has_test_manifests",
        ),
        pytest.param(
            "filter_tests_without_manifests",
            {
                "task": Task(
                    kind="build",
                    label="a",
                    attributes={"test_manifests": None},
                    task={},
                ),
                "parameters": None,
            },
            True,
            id="filter_tests_without_manifests_not_a_test",
        ),
        pytest.param(
            "filter_tests_without_manifests",
            {
                "task": Task(
                    kind="test", label="a", attributes={"test_manifests": None}, task={}
                ),
                "parameters": None,
            },
            False,
            id="filter_tests_without_manifests_has_no_test_manifests",
        ),
        pytest.param(
            "filter_by_regex",
            {
                "task_label": "build-linux64-debug",
                "regexes": [re.compile("build")],
                "mode": "include",
            },
            True,
            id="filter_regex_simple_include",
        ),
        pytest.param(
            "filter_by_regex",
            {
                "task_label": "build-linux64-debug",
                "regexes": [re.compile("linux(.+)debug")],
                "mode": "include",
            },
            True,
            id="filter_regex_re_include",
        ),
        pytest.param(
            "filter_by_regex",
            {
                "task_label": "build-linux64-debug",
                "regexes": [re.compile("nothing"), re.compile("linux(.+)debug")],
                "mode": "include",
            },
            True,
            id="filter_regex_re_include_multiple",
        ),
        pytest.param(
            "filter_by_regex",
            {
                "task_label": "build-linux64-debug",
                "regexes": [re.compile("build")],
                "mode": "exclude",
            },
            False,
            id="filter_regex_simple_exclude",
        ),
        pytest.param(
            "filter_by_regex",
            {
                "task_label": "build-linux64-debug",
                "regexes": [re.compile("linux(.+)debug")],
                "mode": "exclude",
            },
            False,
            id="filter_regex_re_exclude",
        ),
        pytest.param(
            "filter_by_regex",
            {
                "task_label": "build-linux64-debug",
                "regexes": [re.compile("linux(.+)debug"), re.compile("nothing")],
                "mode": "exclude",
            },
            False,
            id="filter_regex_re_exclude_multiple",
        ),
        pytest.param(
            "filter_unsupported_artifact_builds",
            {
                "task": Task(
                    kind="test",
                    label="a",
                    attributes={"supports-artifact-builds": False},
                    task={},
                ),
                "parameters": {
                    "try_task_config": {
                        "use-artifact-builds": False,
                    },
                },
            },
            True,
            id="filter_unsupported_artifact_builds_no_artifact_builds",
        ),
        pytest.param(
            "filter_unsupported_artifact_builds",
            {
                "task": Task(
                    kind="test",
                    label="a",
                    attributes={"supports-artifact-builds": False},
                    task={},
                ),
                "parameters": {
                    "try_task_config": {
                        "use-artifact-builds": True,
                    },
                },
            },
            False,
            id="filter_unsupported_artifact_builds_removed",
        ),
        pytest.param(
            "filter_unsupported_artifact_builds",
            {
                "task": Task(
                    kind="test",
                    label="a",
                    attributes={"supports-artifact-builds": True},
                    task={},
                ),
                "parameters": {
                    "try_task_config": {
                        "use-artifact-builds": True,
                    },
                },
            },
            True,
            id="filter_unsupported_artifact_builds_not_removed",
        ),
        pytest.param(
            "filter_unsupported_artifact_builds",
            {
                "task": Task(kind="test", label="a", attributes={}, task={}),
                "parameters": {
                    "try_task_config": {
                        "use-artifact-builds": True,
                    },
                },
            },
            True,
            id="filter_unsupported_artifact_builds_not_removed",
        ),
    ),
)
def test_filters(name, params, expected):
    func = getattr(target_tasks, name)
    assert func(**params) is expected

if __name__ == "__main__":
    main()

quality95%

¤ Dauer der Verarbeitung: 0.12 Sekunden (vorverarbeitet) ¤

Wurzel

Suchen

Beweissystem der NASA

Beweissystem Isabelle

NIST Cobol Testsuite

Cephes Mathematical Library

Wiener Entwicklungsmethode

Haftungshinweis

Die Informationen auf dieser Webseite wurden nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit, noch Qualität der bereit gestellten Informationen zugesichert.

Bemerkung:

Die farbliche Syntaxdarstellung ist noch experimentell.