blob: f0b106d0c7fb68ac11568a2e6a16860cf8781e92 [file] [log] [blame]
#!/usr/bin/env vpython3
# Copyright 2020 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
from __future__ import print_function
import json
import os
import sys
import unittest
if sys.version_info[0] == 2:
import mock
else:
import unittest.mock as mock
from pyfakefs import fake_filesystem_unittest
from unexpected_passes_common import builders
from unexpected_passes_common import multiprocessing_utils
from unexpected_passes_common import unittest_utils
class GetCiBuildersUnittest(fake_filesystem_unittest.TestCase):
def setUp(self):
self._builders_instance = unittest_utils.GenericBuilders()
self._isolate_patcher = mock.patch.object(
self._builders_instance,
'GetIsolateNames',
return_value={'telemetry_gpu_integration_test'})
self._isolate_mock = self._isolate_patcher.start()
self.addCleanup(self._isolate_patcher.stop)
def CreateFile(self, *args, **kwargs):
# TODO(crbug.com/1156806): Remove this and just use fs.create_file() when
# Catapult is updated to a newer version of pyfakefs that is compatible with
# Chromium's version.
if hasattr(self.fs, 'create_file'):
self.fs.create_file(*args, **kwargs)
else:
self.fs.CreateFile(*args, **kwargs)
def testJsonContentLoaded(self):
"""Tests that the correct JSON data is loaded in."""
self.setUpPyfakefs()
gpu_json = {
'AAAAA1 AUTOGENERATED FILE DO NOT EDIT': {},
'Android Release (Nexus 5X)': {
'isolated_scripts': [{
'args': [
'webgl_conformance',
],
'isolate_name':
'telemetry_gpu_integration_test',
}],
},
'GPU Linux Builder': {},
}
gpu_fyi_json = {
'AAAAA1 AUTOGENERATED FILE DO NOT EDIT': {},
'ANGLE GPU Android Release (Nexus 5X)': {
'isolated_scripts': [{
'args': [
'webgl_conformance',
],
'isolate_name':
'telemetry_gpu_integration_test',
}],
},
'GPU FYI Linux Builder': {},
}
self.CreateFile(os.path.join(builders.TESTING_BUILDBOT_DIR,
'chromium.gpu.json'),
contents=json.dumps(gpu_json))
self.CreateFile(os.path.join(builders.TESTING_BUILDBOT_DIR,
'chromium.gpu.fyi.json'),
contents=json.dumps(gpu_fyi_json))
gpu_builders = self._builders_instance.GetCiBuilders('webgl_conformance')
self.assertEqual(
gpu_builders,
set([
'Android Release (Nexus 5X)', 'ANGLE GPU Android Release (Nexus 5X)'
]))
def testFilterBySuite(self):
"""Tests that only builders that run the given suite are returned."""
def SideEffect(tm, s):
tests = tm.get('isolated_scripts', [])
for t in tests:
if t.get('isolate_name') == 'foo_integration_test':
if s in t.get('args', []):
return True
return False
self.setUpPyfakefs()
gpu_json = {
'AAAAA1 AUTOGENERATED FILE DO NOT EDIT': {},
'Android Tester': {
'isolated_scripts': [
{
'args': [
'bar_conformance',
],
'isolate_name': 'not_telemetry',
},
],
},
'Linux Tester': {
'isolated_scripts': [
{
'args': [
'not_a_suite',
],
'isolate_name': 'foo_integration_test',
},
],
},
'Windows Tester': {
'isolated_scripts': [
{
'args': [
'bar_conformance',
],
'isolate_name': 'foo_integration_test',
},
],
},
}
self.CreateFile(os.path.join(builders.TESTING_BUILDBOT_DIR,
'chromium.json'),
contents=json.dumps(gpu_json))
with mock.patch.object(self._builders_instance,
'_BuilderRunsTestOfInterest',
side_effect=SideEffect):
gpu_builders = self._builders_instance.GetCiBuilders('bar_conformance')
self.assertEqual(gpu_builders, set(['Windows Tester']))
def testRealContentCanBeLoaded(self):
"""Tests that *something* from the real JSON files can be loaded."""
# This directory is not available on swarming, so if it doesn't exist, just
# skip the test.
if not os.path.exists(builders.TESTING_BUILDBOT_DIR):
return
self.assertNotEqual(
len(self._builders_instance.GetCiBuilders('webgl_conformance')), 0)
class GetMirroredBuildersForCiBuilderUnittest(unittest.TestCase):
def setUp(self):
self._builders_instance = builders.Builders()
self._bb_patcher = mock.patch.object(self._builders_instance,
'_GetBuildbucketOutputForCiBuilder')
self._bb_mock = self._bb_patcher.start()
self.addCleanup(self._bb_patcher.stop)
self._fake_ci_patcher = mock.patch.object(self._builders_instance,
'GetFakeCiBuilders',
return_value={})
self._fake_ci_mock = self._fake_ci_patcher.start()
self.addCleanup(self._fake_ci_patcher.stop)
self._non_chromium_patcher = mock.patch.object(
self._builders_instance,
'GetNonChromiumBuilders',
return_value={'foo_non_chromium'})
self._non_chromium_mock = self._non_chromium_patcher.start()
self.addCleanup(self._non_chromium_patcher.stop)
def testFakeCiBuilder(self):
"""Tests that a fake CI builder gets properly mapped."""
self._fake_ci_mock.return_value = {'foo_ci': {'foo_try'}}
try_builder, found_mirror = (
self._builders_instance._GetMirroredBuildersForCiBuilder('foo_ci'))
self.assertTrue(found_mirror)
self.assertEqual(try_builder, set(['foo_try']))
self._bb_mock.assert_not_called()
def testNoBuildbucketOutput(self):
"""Tests that a failure to get Buildbucket output is surfaced."""
self._bb_mock.return_value = ''
try_builder, found_mirror = (
self._builders_instance._GetMirroredBuildersForCiBuilder('nonexistent'))
self.assertFalse(found_mirror)
self.assertEqual(try_builder, set(['nonexistent']))
def testBuildbucketOutput(self):
"""Tests that Buildbucket output is parsed correctly."""
self._bb_mock.return_value = json.dumps({
'output': {
'properties': {
'mirrored_builders': [
'try:foo_try',
'try:bar_try',
]
}
}
})
try_builders, found_mirror = (
self._builders_instance._GetMirroredBuildersForCiBuilder('foo_ci'))
self.assertTrue(found_mirror)
self.assertEqual(try_builders, set(['foo_try', 'bar_try']))
class GetTryBuildersUnittest(unittest.TestCase):
def setUp(self):
self._builders_instance = builders.Builders()
self._get_patcher = mock.patch.object(self._builders_instance,
'_GetMirroredBuildersForCiBuilder')
self._get_mock = self._get_patcher.start()
self.addCleanup(self._get_patcher.stop)
self._pool_patcher = mock.patch.object(multiprocessing_utils,
'GetProcessPool')
self._pool_mock = self._pool_patcher.start()
self._pool_mock.return_value = unittest_utils.FakePool()
self.addCleanup(self._pool_patcher.stop)
def testNoOutputCausesFailure(self):
"""Tests that a failure to get Buildbot output raises an exception."""
self._get_mock.return_value = (set(['foo_ci']), False)
with self.assertRaises(RuntimeError):
self._builders_instance.GetTryBuilders(['foo_ci'])
def testOutputReturned(self):
"""Tests that parsed builders get returned on success."""
def SideEffect(ci_builder):
b = [
ci_builder.replace('ci', 'try'),
ci_builder.replace('ci', 'try2'),
]
return set(b), True
self._get_mock.side_effect = SideEffect
mirrored_builders = self._builders_instance.GetTryBuilders(
['foo_ci', 'bar_ci'])
self.assertEqual(mirrored_builders,
set(['foo_try', 'foo_try2', 'bar_try', 'bar_try2']))
if __name__ == '__main__':
unittest.main(verbosity=2)