Source code
Revision control
Copy as Markdown
Other Tools
Test Info:
# 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
import json
import os
import shutil
import tempfile
from unittest.mock import patch
import pytest
from mozunit import MockedOpen, main
from taskgraph.util.yaml import load_yaml
from gecko_taskgraph import decision
from gecko_taskgraph.parameters import register_parameters
FAKE_GRAPH_CONFIG = {"product-dir": "browser", "taskgraph": {}}
TTC_FILE = os.path.join(os.getcwd(), "try_task_config.json")
@pytest.fixture(scope="module", autouse=True)
def register():
register_parameters()
@pytest.fixture(scope="module")
def options():
return {
"head_rev": "abcd",
"head_ref": "ef01",
"head_tag": "",
"message": "",
"project": "mozilla-central",
"pushlog_id": "143",
"pushdate": 1503691511,
"owner": "nobody@mozilla.com",
"repository_type": "hg",
"tasks_for": "hg-push",
"level": "3",
}
def test_write_artifact_json():
data = [{"some": "data"}]
tmpdir = tempfile.mkdtemp()
try:
decision.ARTIFACTS_DIR = os.path.join(tmpdir, "artifacts")
decision.write_artifact("artifact.json", data)
with open(os.path.join(decision.ARTIFACTS_DIR, "artifact.json")) as f:
assert json.load(f) == data
finally:
if os.path.exists(tmpdir):
shutil.rmtree(tmpdir)
decision.ARTIFACTS_DIR = "artifacts"
def test_write_artifact_yml():
data = [{"some": "data"}]
tmpdir = tempfile.mkdtemp()
try:
decision.ARTIFACTS_DIR = os.path.join(tmpdir, "artifacts")
decision.write_artifact("artifact.yml", data)
assert load_yaml(decision.ARTIFACTS_DIR, "artifact.yml") == data
finally:
if os.path.exists(tmpdir):
shutil.rmtree(tmpdir)
decision.ARTIFACTS_DIR = "artifacts"
@patch("gecko_taskgraph.decision.get_hg_revision_branch")
@patch("gecko_taskgraph.decision.get_hg_commit_message")
@patch("gecko_taskgraph.decision._determine_more_accurate_base_rev")
@patch("gecko_taskgraph.decision.get_changed_files")
@pytest.mark.parametrize(
"extra_options,commit_msg,ttc,expected",
(
pytest.param(
{},
None,
None,
{
"pushlog_id": "143",
"build_date": 1503691511,
"files_changed": ["bar/baz.md", "foo.txt"],
"hg_branch": "default",
"moz_build_date": "20170825200511",
"try_mode": None,
"try_options": None,
"try_task_config": {},
},
id="simple_options",
),
pytest.param(
{"owner": "ffxbld"},
None,
None,
{
"owner": "ffxbld@noreply.mozilla.org",
},
id="no_email_owner",
),
pytest.param(
{"project": "try"},
"try: -b do -t all --artifact",
None,
{
"try_mode": "try_option_syntax",
"try_options": {
"build_types": "do",
"include_nightly": False,
"interactive": False,
"jobs": None,
"no_retry": False,
"notifications": None,
"platforms": "all",
"raptor": "none",
"raptor_trigger_tests": 1,
"tag": None,
"talos": "all",
"talos_trigger_tests": 1,
"taskcluster_worker": False,
"trigger_tests": 1,
"unittests": "all",
},
"try_task_config": {
"gecko-profile": False,
"use-artifact-builds": True,
"env": {},
},
},
id="try_options",
),
pytest.param(
{
"project": "try",
},
"Fuzzy query=foo",
{"tasks": ["a", "b"]},
{
"try_mode": "try_task_config",
"try_options": None,
"try_task_config": {"tasks": ["a", "b"]},
},
id="try_task_config",
),
),
)
def test_get_decision_parameters(
mock_get_changed_files,
mock_determine_more_accurate_base_rev,
mock_get_hg_commit_message,
mock_get_hg_revision_branch,
options,
extra_options,
commit_msg,
ttc,
expected,
):
mock_get_hg_revision_branch.return_value = "default"
mock_get_hg_commit_message.return_value = commit_msg or "commit message"
mock_determine_more_accurate_base_rev.return_value = "baserev"
mock_get_changed_files.return_value = ["foo.txt", "bar/baz.md"]
options.update(extra_options)
contents = None
if ttc:
contents = json.dumps(ttc)
with MockedOpen({TTC_FILE: contents}):
params = decision.get_decision_parameters(FAKE_GRAPH_CONFIG, options)
for key in expected:
assert params[key] == expected[key], f"key {key} does not match!"
@pytest.mark.parametrize(
"msg, expected",
(
pytest.param("", "", id="empty"),
pytest.param("abc | def", "", id="no_try_syntax"),
pytest.param("try: -f -o -o", "try: -f -o -o", id="initial_try_syntax"),
pytest.param(
"some stuff\ntry: -f -o -o\nabc\ndef",
"try: -f -o -o",
id="embedded_try_syntax_multiline",
),
),
)
def test_try_syntax_from_message(msg, expected):
assert decision.try_syntax_from_message(msg) == expected
if __name__ == "__main__":
main()