1#!/usr/bin/env vpython3 2# Copyright 2020 The Chromium Authors 3# Use of this source code is governed by a BSD-style license that can be 4# found in the LICENSE file. 5"""Tests for base_device_trigger.py.""" 6 7import argparse 8import json 9import unittest 10import os 11 12import mock 13 14from pyfakefs import fake_filesystem_unittest 15 16import base_test_triggerer 17 18 19class UnitTest(fake_filesystem_unittest.TestCase): 20 def setUp(self): 21 self.setUpPyfakefs() 22 23 @unittest.skipUnless('TRIGGER_SCRIPT_INTEGRATION_TESTS' in os.environ, 24 'this is quick check test using real swarming server') 25 def test_list_bots(self): 26 # This just checks list_bots runs without error. 27 triggerer = base_test_triggerer.BaseTestTriggerer() 28 with fake_filesystem_unittest.Pause(self): 29 bots = triggerer.list_bots({'pool': 'luci.flex.ci'}, True) 30 self.assertGreater(len(bots), 0) 31 bot = bots[0] 32 self.assertIn('bot_id', bot) 33 34 @unittest.skipUnless('TRIGGER_SCRIPT_INTEGRATION_TESTS' in os.environ, 35 'this is quick check test using real swarming server') 36 def test_list_bots(self): 37 # This just checks list_bots runs without error. 38 triggerer = base_test_triggerer.BaseTestTriggerer() 39 with fake_filesystem_unittest.Pause(self): 40 tasks = triggerer.list_tasks(('pool:luci.flex.ci', ), limit=1) 41 self.assertGreater(len(tasks), 0) 42 task = tasks[0] 43 self.assertIn('task_id', task) 44 45 def test_convert_to_go_swarming_args(self): 46 args = [ 47 '--swarming', 'x.apphost.com', '--dimension', 'pool', 'ci', 48 '--dimension', 'os', 'linux', '-env', 'FOO=foo', '--hello', 49 '-cipd-package', 'path:name=123', '--scalar', '42', 50 '--enable-resultdb' 51 ] 52 go_args = base_test_triggerer._convert_to_go_swarming_args(args) 53 expected = [ 54 '--server', 'x.apphost.com', '--dimension', 'pool=ci', 55 '--dimension', 'os=linux', '-env', 'FOO=foo', '--hello', 56 '-cipd-package', 'path:name=123', '--scalar', '42', 57 '--enable-resultdb' 58 ] 59 self.assertEquals(go_args, expected) 60 61 def test_convert_to_go_swarming_args_failed(self): 62 invalid_args = [ 63 # expected format: --dimension key value 64 ([ 65 '--dimension', 66 'key', 67 ], IndexError), 68 ] 69 for args, ex in invalid_args: 70 self.assertRaises(ex, 71 base_test_triggerer._convert_to_go_swarming_args, 72 args) 73 74 def test_trigger_tasks(self): 75 parser = base_test_triggerer.BaseTestTriggerer.setup_parser_contract( 76 argparse.ArgumentParser()) 77 dump_json = 'dump.json' 78 args, remaining = parser.parse_known_args([ 79 '--multiple-dimension-script-verbose', 'True', 'trigger', 80 '--shards', '1', '--dump-json', dump_json 81 ]) 82 triggerer = base_test_triggerer.BaseTestTriggerer() 83 84 def mock_subprocess_call(args): 85 # write json file generated by 'swarming trigger' command. 86 json_path = args[args.index('--dump-json') + 1] 87 with open(json_path, 'w') as f: 88 f.write( 89 json.dumps({ 90 'tasks': [{ 91 'request': { 92 'task_id': 'f0', 93 }, 94 'task_result': { 95 'resultdb_info': { 96 'invocation': 'task-f0', 97 }, 98 }, 99 }], 100 })) 101 102 # make some not important functions nop. 103 with mock.patch.object(triggerer, 104 'parse_bot_configs'), mock.patch.object( 105 triggerer, '_bot_configs', 106 []), mock.patch.object( 107 triggerer, 108 'select_config_indices', 109 return_value=[(0, 0)]), mock.patch( 110 'subprocess.call', 111 side_effect=mock_subprocess_call): 112 triggerer.trigger_tasks(args, remaining) 113 114 with open(dump_json) as f: 115 self.assertEqual( 116 json.load(f), { 117 u'tasks': { 118 u'f0:0:1': { 119 u'shard_index': 0, 120 u'task_id': u'f0', 121 u'invocation': u'task-f0', 122 } 123 } 124 }) 125 126 127if __name__ == '__main__': 128 unittest.main() 129