Commit 4833795e authored by Takuto Ikuta's avatar Takuto Ikuta Committed by Chromium LUCI CQ

trigger_scripts: add test for trigger_tasks function

This is to prevent failure causing revert like
https://crrev.com/c/2639349

Bug: 1127205
Change-Id: I6d6fac6a900caa3c64fc80f3b33cf4be5b7713c9
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2642838
Auto-Submit: Takuto Ikuta <tikuta@chromium.org>
Commit-Queue: Brian Sheedy <bsheedy@chromium.org>
Reviewed-by: default avatarBrian Sheedy <bsheedy@chromium.org>
Cr-Commit-Position: refs/heads/master@{#845725}
parent fcb6d1c3
......@@ -6,12 +6,20 @@
"""Tests for base_device_trigger.py."""
import argparse
import json
import unittest
import mock
from pyfakefs import fake_filesystem_unittest
import base_test_triggerer
class UnitTest(unittest.TestCase):
class UnitTest(fake_filesystem_unittest.TestCase):
def setUp(self):
self.setUpPyfakefs()
def test_convert_to_go_swarming_args(self):
args = [
'--swarming', 'x.apphost.com', '--dimension', 'pool', 'ci',
......@@ -71,5 +79,49 @@ class UnitTest(unittest.TestCase):
args, _ = parser.parse_known_args(swarming_args + ['--use-swarming-go'])
self.assertTrue(args.use_swarming_go)
def test_trigger_tasks(self):
parser = base_test_triggerer.BaseTestTriggerer.setup_parser_contract(
argparse.ArgumentParser())
dump_json = 'dump.json'
args, remaining = parser.parse_known_args([
'--multiple-dimension-script-verbose', 'True', '--use-swarming-go',
'trigger', '--shards', '1', '--dump-json', dump_json
])
triggerer = base_test_triggerer.BaseTestTriggerer()
def mock_subprocess_call(args):
# write json file generated by 'swarming trigger' command.
json_path = args[args.index('--dump-json') + 1]
with open(json_path, 'w') as f:
f.write(
json.dumps({
'tasks': [{
'request': {
'task_id': 'f0',
},
}],
}))
# make some not important functions nop.
with mock.patch.object(
triggerer, 'parse_bot_configs'), mock.patch.object(
triggerer, '_bot_configs',
[]), mock.patch.object(
triggerer, 'select_config_indices',
return_value=[(0, 0)]), mock.patch(
'subprocess.call', side_effect=mock_subprocess_call):
triggerer.trigger_tasks(args, remaining)
with open(dump_json) as f:
self.assertEqual(json.load(f), {
u'tasks': {
u'f0:0:1': {
u'shard_index': 0,
u'task_id': u'f0'
}
}
})
if __name__ == '__main__':
unittest.main()
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment