standard_isolated_script_merge_test.py 5.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188
  1. #!/usr/bin/env vpython3
  2. # Copyright 2017 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. import json
  6. import os
  7. import shutil
  8. import tempfile
  9. import unittest
  10. import mock
  11. import common_merge_script_tests
  12. THIS_DIR = os.path.dirname(__file__)
  13. import standard_isolated_script_merge
  14. TWO_COMPLETED_SHARDS = {
  15. u'shards': [
  16. {
  17. u'state': u'COMPLETED',
  18. },
  19. {
  20. u'state': u'COMPLETED',
  21. },
  22. ],
  23. }
  24. class StandardIsolatedScriptMergeTest(unittest.TestCase):
  25. def setUp(self):
  26. self.temp_dir = tempfile.mkdtemp()
  27. self.test_files = []
  28. self.summary = None
  29. # pylint: disable=super-with-arguments
  30. def tearDown(self):
  31. shutil.rmtree(self.temp_dir)
  32. super(StandardIsolatedScriptMergeTest, self).tearDown()
  33. # pylint: enable=super-with-arguments
  34. def _write_temp_file(self, path, content):
  35. abs_path = os.path.join(self.temp_dir, path.replace('/', os.sep))
  36. if not os.path.exists(os.path.dirname(abs_path)):
  37. os.makedirs(os.path.dirname(abs_path))
  38. with open(abs_path, 'w') as f:
  39. if isinstance(content, dict):
  40. json.dump(content, f)
  41. else:
  42. assert isinstance(content, str)
  43. f.write(content)
  44. return abs_path
  45. def _stage(self, summary, files):
  46. self.summary = self._write_temp_file('summary.json', summary)
  47. for path, content in files.items():
  48. abs_path = self._write_temp_file(path, content)
  49. self.test_files.append(abs_path)
  50. class OutputTest(StandardIsolatedScriptMergeTest):
  51. def test_success_and_failure(self):
  52. self._stage(TWO_COMPLETED_SHARDS,
  53. {
  54. '0/output.json':
  55. {
  56. 'successes': ['fizz', 'baz'],
  57. },
  58. '1/output.json':
  59. {
  60. 'successes': ['buzz', 'bar'],
  61. 'failures': ['failing_test_one']
  62. }
  63. })
  64. output_json_file = os.path.join(self.temp_dir, 'output.json')
  65. standard_isolated_script_merge.StandardIsolatedScriptMerge(
  66. output_json_file, self.summary, self.test_files)
  67. with open(output_json_file, 'r') as f:
  68. results = json.load(f)
  69. self.assertEquals(results['successes'], ['fizz', 'baz', 'buzz', 'bar'])
  70. self.assertEquals(results['failures'], ['failing_test_one'])
  71. self.assertTrue(results['valid'])
  72. def test_missing_shard(self):
  73. self._stage(TWO_COMPLETED_SHARDS,
  74. {
  75. '0/output.json':
  76. {
  77. 'successes': ['fizz', 'baz'],
  78. },
  79. })
  80. output_json_file = os.path.join(self.temp_dir, 'output.json')
  81. standard_isolated_script_merge.StandardIsolatedScriptMerge(
  82. output_json_file, self.summary, self.test_files)
  83. with open(output_json_file, 'r') as f:
  84. results = json.load(f)
  85. self.assertEquals(results['successes'], ['fizz', 'baz'])
  86. self.assertEquals(results['failures'], [])
  87. self.assertTrue(results['valid'])
  88. self.assertEquals(results['global_tags'], ['UNRELIABLE_RESULTS'])
  89. self.assertEquals(results['missing_shards'], [1])
  90. class InputParsingTest(StandardIsolatedScriptMergeTest):
  91. # pylint: disable=super-with-arguments
  92. def setUp(self):
  93. super(InputParsingTest, self).setUp()
  94. self.merge_test_results_args = []
  95. def mock_merge_test_results(results_list):
  96. self.merge_test_results_args.append(results_list)
  97. return {
  98. 'foo': [
  99. 'bar',
  100. 'baz',
  101. ],
  102. }
  103. m = mock.patch(
  104. 'standard_isolated_script_merge.results_merger.merge_test_results',
  105. side_effect=mock_merge_test_results)
  106. m.start()
  107. self.addCleanup(m.stop)
  108. # pylint: enable=super-with-arguments
  109. def test_simple(self):
  110. self._stage(TWO_COMPLETED_SHARDS,
  111. {
  112. '0/output.json':
  113. {
  114. 'result0': ['bar', 'baz'],
  115. },
  116. '1/output.json':
  117. {
  118. 'result1': {'foo': 'bar'}
  119. }
  120. })
  121. output_json_file = os.path.join(self.temp_dir, 'output.json')
  122. exit_code = standard_isolated_script_merge.StandardIsolatedScriptMerge(
  123. output_json_file, self.summary, self.test_files)
  124. self.assertEquals(0, exit_code)
  125. self.assertEquals(
  126. [
  127. [
  128. {
  129. 'result0': [
  130. 'bar', 'baz',
  131. ],
  132. },
  133. {
  134. 'result1': {
  135. 'foo': 'bar',
  136. },
  137. }
  138. ],
  139. ],
  140. self.merge_test_results_args)
  141. def test_no_jsons(self):
  142. self._stage({
  143. u'shards': [],
  144. }, {})
  145. json_files = []
  146. output_json_file = os.path.join(self.temp_dir, 'output.json')
  147. exit_code = standard_isolated_script_merge.StandardIsolatedScriptMerge(
  148. output_json_file, self.summary, json_files)
  149. self.assertEquals(0, exit_code)
  150. self.assertEquals([[]], self.merge_test_results_args)
  151. class CommandLineTest(common_merge_script_tests.CommandLineTest):
  152. # pylint: disable=super-with-arguments
  153. def __init__(self, methodName='runTest'):
  154. super(CommandLineTest, self).__init__(
  155. methodName, standard_isolated_script_merge)
  156. # pylint: enable=super-with-arguments
  157. if __name__ == '__main__':
  158. unittest.main()