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