Skip to content

Commit

Permalink
100% coverage!
Browse files Browse the repository at this point in the history
  • Loading branch information
@escapewindow
escapewindow committed Jan 10, 2018
1 parent 6ab5cd5 commit b102836
Show file tree
Hide file tree
Showing 2 changed files with 104 additions and 31 deletions.
5 changes: 2 additions & 3 deletions scriptworker/cot/verify.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1152,8 +1152,6 @@ async def verify_parent_task_definition(chain, parent_link):
},
}
if tasks_for == 'action':
# TODO add this in .taskcluster.yml? This is *very* fragile against
# future changes.
actions_path = decision_link.get_artifact_full_path('public/actions.json')
params_path = decision_link.get_artifact_full_path('public/parameters.yml')
action_name = get_action_name(parent_link.task)
Expand All @@ -1169,9 +1167,9 @@ async def verify_parent_task_definition(chain, parent_link):
else:
pushlog_info = await get_pushlog_info(decision_link)
pushlog_id = list(pushlog_info['pushes'].keys())[0]
# Verify comment
decision_comment = get_commit_message(decision_link.task)
if tasks_for == 'hg-push':
# on-push decision task
push_comment = pushlog_info['pushes'][pushlog_id]['changesets'][0]['desc']
if decision_comment not in (' ', push_comment):
raise CoTError(
Expand All @@ -1189,6 +1187,7 @@ async def verify_parent_task_definition(chain, parent_link):
"pushdate": pushlog_info['pushes'][pushlog_id]['date'],
}
else:
# cron decision task
jsone_context['cron'] = load_json_or_yaml(parent_link.task['extra']['cron'])
# I don't love these hardcodes.
jsone_context["push"] = {
Expand Down
130 changes: 102 additions & 28 deletions scriptworker/test/test_cot_verify.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
from copy import deepcopy
from frozendict import frozendict
import json
import jsone
import logging
import mock
import os
Expand Down Expand Up @@ -258,6 +259,30 @@ def get_cot(task_defn, task_id="task_id"):
"workerId": "..."
}

async def cotv2_load_url(context, url, path, **kwargs):
if path.endswith("source_tmpl.yml"):
return load_json_or_yaml(
os.path.join(COTV2_DIR, ".taskcluster.yml"), is_path=True, file_type='yaml'
)
elif path.endswith("projects.yml"):
return load_json_or_yaml(
os.path.join(COTV2_DIR, "projects.yml"), is_path=True, file_type='yaml'
)

def cotv2_load(string, is_path=False, **kwargs):
if is_path:
if string.endswith("parameters.yml"):
return load_json_or_yaml(
os.path.join(COTV2_DIR, "parameters.yml"), is_path=True, file_type='yaml'
)
elif string.endswith("actions.json"):
return load_json_or_yaml(os.path.join(COTV2_DIR, "actions.json"), is_path=True)
else:
return load_json_or_yaml(string)

async def cotv2_pushlog(_):
return load_json_or_yaml(os.path.join(COTV2_DIR, "pushlog.json"), is_path=True)


# dependent_task_ids {{{1
def test_dependent_task_ids(chain):
Expand Down Expand Up @@ -1110,40 +1135,90 @@ async def test_verify_parent_task_definition(chain, name, task_id, path,
decision_link = cotverify.LinkOfTrust(chain.context, 'decision', decision_task_id)
decision_link.task = load_json_or_yaml(decision_path, is_path=True)

async def fake_load_url(context, url, path, **kwargs):
if path.endswith("source_tmpl.yml"):
return load_json_or_yaml(
os.path.join(COTV2_DIR, ".taskcluster.yml"), is_path=True, file_type='yaml'
)
elif path.endswith("projects.yml"):
return load_json_or_yaml(
os.path.join(COTV2_DIR, "projects.yml"), is_path=True, file_type='yaml'
)

def fake_load(string, is_path=False, **kwargs):
if is_path:
if string.endswith("parameters.yml"):
return load_json_or_yaml(
os.path.join(COTV2_DIR, "parameters.yml"), is_path=True, file_type='yaml'
)
elif string.endswith("actions.json"):
return load_json_or_yaml(os.path.join(COTV2_DIR, "actions.json"), is_path=True)
else:
return load_json_or_yaml(string)

async def fake_pushlog(_):
return load_json_or_yaml(os.path.join(COTV2_DIR, "pushlog.json"), is_path=True)

mocker.patch.object(cotverify, 'load_json_or_yaml_from_url', new=fake_load_url)
mocker.patch.object(cotverify, 'load_json_or_yaml', new=fake_load)
mocker.patch.object(cotverify, 'get_pushlog_info', new=fake_pushlog)
mocker.patch.object(cotverify, 'load_json_or_yaml_from_url', new=cotv2_load_url)
mocker.patch.object(cotverify, 'load_json_or_yaml', new=cotv2_load)
mocker.patch.object(cotverify, 'get_pushlog_info', new=cotv2_pushlog)

chain.links = list(set([decision_link, link]))
await cotverify.verify_parent_task_definition(
chain, link
)


@pytest.mark.asyncio
async def test_verify_parent_task_definition_bad_project(chain, mocker):
link = cotverify.LinkOfTrust(chain.context, 'decision', "VQU9QMO4Teq7zr91FhBusg")
link.task = load_json_or_yaml(os.path.join(COTV2_DIR, "decision_hg-push.json"), is_path=True)

def fake_url(*args):
return "https://fake_server"

mocker.patch.object(cotverify, 'load_json_or_yaml_from_url', new=cotv2_load_url)
mocker.patch.object(cotverify, 'load_json_or_yaml', new=cotv2_load)
mocker.patch.object(cotverify, 'get_pushlog_info', new=cotv2_pushlog)
mocker.patch.object(cotverify, 'get_source_url', new=fake_url)

chain.links = [link]
with pytest.raises(CoTError):
await cotverify.verify_parent_task_definition(
chain, link
)


@pytest.mark.asyncio
async def test_verify_parent_task_definition_bad_comment(chain, mocker):
link = cotverify.LinkOfTrust(chain.context, 'decision', "VQU9QMO4Teq7zr91FhBusg")
link.task = load_json_or_yaml(os.path.join(COTV2_DIR, "decision_hg-push.json"), is_path=True)
link.task['payload']['env']['GECKO_COMMIT_MSG'] = "invalid comment"

mocker.patch.object(cotverify, 'load_json_or_yaml_from_url', new=cotv2_load_url)
mocker.patch.object(cotverify, 'load_json_or_yaml', new=cotv2_load)
mocker.patch.object(cotverify, 'get_pushlog_info', new=cotv2_pushlog)

chain.links = [link]
with pytest.raises(CoTError):
await cotverify.verify_parent_task_definition(
chain, link
)


@pytest.mark.asyncio
async def test_verify_parent_task_definition_failed_jsone(chain, mocker):
link = cotverify.LinkOfTrust(chain.context, 'decision', "VQU9QMO4Teq7zr91FhBusg")
link.task = load_json_or_yaml(os.path.join(COTV2_DIR, "decision_hg-push.json"), is_path=True)

def die(*args):
raise jsone.JSONTemplateError("foo")

mocker.patch.object(cotverify, 'load_json_or_yaml_from_url', new=cotv2_load_url)
mocker.patch.object(cotverify, 'load_json_or_yaml', new=cotv2_load)
mocker.patch.object(cotverify, 'get_pushlog_info', new=cotv2_pushlog)
mocker.patch.object(jsone, 'render', new=die)

chain.links = [link]
with pytest.raises(CoTError):
await cotverify.verify_parent_task_definition(
chain, link
)


@pytest.mark.asyncio
async def test_verify_parent_task_definition_failed_diff(chain, mocker):
link = cotverify.LinkOfTrust(chain.context, 'decision', "VQU9QMO4Teq7zr91FhBusg")
link.task = load_json_or_yaml(os.path.join(COTV2_DIR, "decision_hg-push.json"), is_path=True)
link.task['illegal'] = 'boom'

mocker.patch.object(cotverify, 'load_json_or_yaml_from_url', new=cotv2_load_url)
mocker.patch.object(cotverify, 'load_json_or_yaml', new=cotv2_load)
mocker.patch.object(cotverify, 'get_pushlog_info', new=cotv2_pushlog)

chain.links = [link]
with pytest.raises(CoTError):
await cotverify.verify_parent_task_definition(
chain, link
)


# verify_parent_task {{{1
@pytest.mark.asyncio
@pytest.mark.parametrize("defn_fn,min_cot_version,raises", ((
Expand Down Expand Up @@ -1216,7 +1291,6 @@ def task_graph(*args, **kwargs):
decision_link.task['workerType'] = 'bad-worker-type'
mocker.patch.object(cotverify, 'load_json_or_yaml', new=task_graph)
mocker.patch.object(cotverify, 'verify_decision_command', new=noop_sync)
# TODO fix
mocker.patch.object(cotverify, 'verify_parent_task_definition', new=noop_async)
with pytest.raises(CoTError):
await cotverify.verify_parent_task(chain, decision_link)
Expand Down

0 comments on commit b102836

Please sign in to comment.