# 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 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 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, "repository_type": "hg", "hg_branch": "default", "level": "3", }, ) 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", "repository_type": "hg", "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) 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_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", ), pytest.param( "filter_for_repo_type", { "task": Task(kind="test", label="a", attributes={}, task={}), "parameters": { "repository_type": "hg", }, }, True, id="filter_for_repo_type_default_hg_not_removed", ), pytest.param( "filter_for_repo_type", { "task": Task(kind="test", label="a", attributes={}, task={}), "parameters": { "repository_type": "git", }, }, True, id="filter_for_repo_type_default_git_not_removed", ), pytest.param( "filter_for_repo_type", { "task": Task( kind="test", label="a", attributes={"run_on_repo_type": ["hg"]}, task={}, ), "parameters": { "repository_type": "git", }, }, False, id="filter_for_repo_type_no_match_removed", ), pytest.param( "filter_for_repo_type", { "task": Task( kind="test", label="a", attributes={"run_on_repo_type": ["git"]}, task={}, ), "parameters": { "repository_type": "git", }, }, True, id="filter_for_repo_type_match_not_removed", ), pytest.param( "filter_for_repo_type", { "task": Task( kind="test", label="a", attributes={"run_on_repo_type": ["all"]}, task={}, ), "parameters": { "repository_type": "git", }, }, True, id="filter_for_repo_type_all_not_removed", ), ), ) def test_filters(name, params, expected): func = getattr(target_tasks, name) assert func(**params) is expected if __name__ == "__main__": main()