base_test_triggerer_unittest.py 4.8 KB

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