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