| # -*- coding: utf-8 -*- |
| # Copyright 2018 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Unit tests for buildbucket_v2.""" |
| |
| from __future__ import print_function |
| |
| from datetime import datetime, date |
| import sys |
| |
| from google.protobuf import field_mask_pb2 |
| from google.protobuf.struct_pb2 import Struct, Value |
| from google.protobuf.timestamp_pb2 import Timestamp |
| |
| from chromite.lib import buildbucket_v2 |
| from chromite.lib import constants |
| from chromite.lib import cros_test_lib |
| from chromite.lib import cros_logging as logging |
| from chromite.lib import metadata_lib |
| from chromite.lib.luci.prpc.client import Client, ProtocolError |
| |
| from infra_libs.buildbucket.proto import build_pb2, rpc_pb2, common_pb2 |
| from infra_libs.buildbucket.proto import step_pb2 |
| |
| |
| assert sys.version_info >= (3, 6), 'This module requires Python 3.6+' |
| |
| |
| class BuildbucketV2Test(cros_test_lib.MockTestCase): |
| """Tests for buildbucket_v2.""" |
| # pylint: disable=attribute-defined-outside-init |
| |
| def testCreatesClient(self): |
| ret = buildbucket_v2.BuildbucketV2(test_env=True) |
| self.assertIsInstance(ret.client, Client) |
| |
| def testGetBuildWithProperties(self): |
| fake_field_mask = field_mask_pb2.FieldMask(paths=['properties']) |
| fake_get_build_request = object() |
| bbv2 = buildbucket_v2.BuildbucketV2() |
| client = bbv2.client |
| self.get_build_request_fn = self.PatchObject( |
| rpc_pb2, 'GetBuildRequest', return_value=fake_get_build_request) |
| self.get_build_function = self.PatchObject(client, 'GetBuild') |
| bbv2.GetBuild('some-id', 'properties') |
| self.get_build_request_fn.assert_called_with(id='some-id', |
| fields=fake_field_mask) |
| self.get_build_function.assert_called_with(fake_get_build_request) |
| |
| def testGetBuildWithoutProperties(self): |
| fake_get_build_request = object() |
| bbv2 = buildbucket_v2.BuildbucketV2() |
| client = bbv2.client |
| self.get_build_request_fn = self.PatchObject( |
| rpc_pb2, 'GetBuildRequest', return_value=fake_get_build_request) |
| self.get_build_function = self.PatchObject(client, 'GetBuild') |
| bbv2.GetBuild('some-id') |
| self.get_build_request_fn.assert_called_with(id='some-id') |
| self.get_build_function.assert_called_with(fake_get_build_request) |
| |
| def testGetBuildStages(self): |
| """Test the GetBuildStages functionality.""" |
| bbv2 = buildbucket_v2.BuildbucketV2() |
| start_time = Timestamp() |
| start_time.GetCurrentTime() |
| step = step_pb2.Step(name='stage_name', start_time=start_time, |
| status=2) |
| build_with_steps = build_pb2.Build(steps=[step]) |
| get_build_fn = self.PatchObject(bbv2, 'GetBuild', |
| return_value=build_with_steps) |
| get_build_status_fn = self.PatchObject( |
| bbv2, 'GetBuildStatus', |
| return_value={'build_config': 'something-paladin'}) |
| expected_result = [{ |
| 'name': 'stage_name', |
| 'start_time': datetime.fromtimestamp(start_time.seconds), |
| 'finish_time': None, |
| 'buildbucket_id': 1234, |
| 'status': constants.BUILDER_STATUS_INFLIGHT, |
| 'build_config': 'something-paladin'}] |
| self.assertEqual(bbv2.GetBuildStages(1234), expected_result) |
| get_build_fn.assert_called_once_with(1234, properties='steps') |
| get_build_status_fn.assert_called_once_with(1234) |
| |
| def testGetKilledChildBuildsWithValidId(self): |
| """Test a valid query flow.""" |
| bbv2 = buildbucket_v2.BuildbucketV2() |
| buildbucket_id = 1234 |
| expected_child_builds = [8921795536486453568, 8921795536486453567] |
| fake_properties = Struct(fields={ |
| 'killed_child_builds': Value(string_value=str(expected_child_builds)) |
| }) |
| fake_output = build_pb2.Build.Output(properties=fake_properties) |
| fake_build = build_pb2.Build(id=1234, output=fake_output) |
| self.PatchObject(buildbucket_v2.BuildbucketV2, 'GetBuild', |
| return_value=fake_build) |
| builds = bbv2.GetKilledChildBuilds(buildbucket_id) |
| self.assertEqual(builds, expected_child_builds) |
| |
| def testGetKilledChildBuildsWithInvalidId(self): |
| """Test an unsuccessful query.""" |
| bbv2 = buildbucket_v2.BuildbucketV2() |
| buildbucket_id = 1234 |
| self.PatchObject(buildbucket_v2.BuildbucketV2, 'GetBuild', |
| side_effect=ProtocolError) |
| builds = bbv2.GetKilledChildBuilds(buildbucket_id) |
| self.assertIsNone(builds) |
| |
| def testGetBuildStatusWithValidId(self): |
| """Tests for GetBuildStatus with a valid ID.""" |
| properties_dict = { |
| 'cidb_id': '1234', |
| 'bot_id': 'swarm-cros-34', |
| 'cbb_branch': 'master', |
| 'cbb_config': 'sludge-paladin-tryjob', |
| 'cbb_master_build_id': '4321', |
| 'platform_version':'11721.0.0', |
| 'milestone_version': '74', |
| 'full_version': 'R74-11721.0.0-b3457724', |
| 'critical': '1', |
| 'build_type': 'Try', |
| } |
| start_time = Timestamp() |
| start_time.GetCurrentTime() |
| fake_properties = Struct(fields={ |
| key: Value(string_value=value) for key, value in properties_dict.items() |
| }) |
| fake_output = build_pb2.Build.Output(properties=fake_properties) |
| fake_build = build_pb2.Build( |
| id=1234, start_time=start_time, status=2, output=fake_output) |
| self.PatchObject(buildbucket_v2.BuildbucketV2, 'GetBuild', |
| return_value=fake_build) |
| expected_valid_status = { |
| 'build_config': 'sludge-paladin-tryjob', |
| 'start_time': datetime.fromtimestamp(start_time.seconds), |
| 'finish_time': None, |
| 'id': 1234, |
| 'status': constants.BUILDER_STATUS_INFLIGHT, |
| 'chrome_version': None, |
| 'platform_version':'11721.0.0', |
| 'milestone_version': '74', |
| 'full_version': 'R74-11721.0.0-b3457724', |
| 'important': 1, |
| 'buildbucket_id': 1234, |
| 'summary': None, |
| 'master_build_id': 4321, |
| 'bot_hostname': 'swarm-cros-34', |
| 'builder_name': None, |
| 'build_number': None, |
| 'buildbot_generation': None, |
| 'waterfall': None, |
| 'deadline': None, |
| 'build_type': 'Try', |
| 'metadata_url': None, |
| 'toolchain_url': None, |
| 'branch': 'master' |
| } |
| bbv2 = buildbucket_v2.BuildbucketV2() |
| status = bbv2.GetBuildStatus(1234) |
| self.assertEqual(status, expected_valid_status) |
| |
| def testGetBuildStatusWithInvalidId(self): |
| """Test the function for an ID that doesn't exist in Buildbucket.""" |
| expected_invalid_status = { |
| 'build_config': None, |
| 'start_time': None, |
| 'finish_time': None, |
| 'status': None, |
| 'id': None, |
| 'chrome_version': None, |
| 'platform_version': None, |
| 'milestone_version': None, |
| 'full_version': None, |
| 'important': None, |
| 'buildbucket_id': 0, |
| 'summary': None, |
| 'master_build_id': None, |
| 'bot_hostname': None, |
| 'builder_name': None, |
| 'build_number': None, |
| 'buildbot_generation': None, |
| 'waterfall': None, |
| 'deadline': None, |
| 'build_type': None, |
| 'metadata_url': None, |
| 'toolchain_url': None, |
| 'branch': None |
| } |
| self.PatchObject(buildbucket_v2.BuildbucketV2, 'GetBuild', |
| side_effect=ProtocolError) |
| bbv2 = buildbucket_v2.BuildbucketV2() |
| status = bbv2.GetBuildStatus(0) |
| self.assertEqual(status, expected_invalid_status) |
| |
| def testSearchBuildExceptionCases(self): |
| """Test scenarios where SearchBuild raises an Exception.""" |
| bbv2 = buildbucket_v2.BuildbucketV2() |
| builder = build_pb2.BuilderID(project='chromeos', bucket='general') |
| tag = common_pb2.StringPair(key='cbb_master_buildbucket_id', |
| value=str(1234)) |
| build_predicate = rpc_pb2.BuildPredicate( |
| builder=builder, tags=[tag]) |
| with self.assertRaises(AssertionError): |
| bbv2.SearchBuild(None, None, 100) |
| with self.assertRaises(AssertionError): |
| bbv2.SearchBuild(build_predicate, None, None) |
| with self.assertRaises(AssertionError): |
| bbv2.SearchBuild(build_predicate, 'str_fields', 100) |
| |
| def testSearchBuild(self): |
| """Test redirection to the underlying RPC call.""" |
| bbv2 = buildbucket_v2.BuildbucketV2() |
| builder = build_pb2.BuilderID(project='chromeos', bucket='general') |
| tag = common_pb2.StringPair(key='cbb_master_buildbucket_id', |
| value=str(1234)) |
| build_predicate = rpc_pb2.BuildPredicate( |
| builder=builder, tags=[tag]) |
| fields = field_mask_pb2.FieldMask() |
| search_builds_fn = self.PatchObject(bbv2.client, 'SearchBuilds') |
| bbv2.SearchBuild(build_predicate, fields=fields, page_size=123) |
| search_builds_fn.assert_called_once_with(rpc_pb2.SearchBuildsRequest( |
| predicate=build_predicate, fields=fields, page_size=123)) |
| |
| def testGetBuildHistoryIgnoreIdFoundId(self): |
| """Test GetBuildHistory ignore_build_id logic.""" |
| # pylint: disable=unused-variable |
| bbv2 = buildbucket_v2.BuildbucketV2() |
| builder = build_pb2.BuilderID(project='chromeos', bucket='general') |
| tags = [common_pb2.StringPair(key='cbb_config', |
| value='something-paladin')] |
| build_list = [build_pb2.Build(id=1234), |
| build_pb2.Build(id=2341)] |
| search_fn = self.PatchObject( |
| bbv2, 'SearchBuild', |
| return_value=rpc_pb2.SearchBuildsResponse(builds=build_list)) |
| status_fn = self.PatchObject(bbv2, 'GetBuildStatus') |
| bbv2.GetBuildHistory('something-paladin', 1, ignore_build_id=1234) |
| fake_predicate = rpc_pb2.BuildPredicate( |
| builder=builder, tags=tags, create_time=None, build=None) |
| search_fn.assert_called_once_with(fake_predicate, page_size=2) |
| status_fn.assert_called_once_with(2341) |
| |
| def testGetBuildHistoryIgnoreIdWithoutId(self): |
| """Test GetBuildHistory ignore_build_id logic when ID is absent.""" |
| bbv2 = buildbucket_v2.BuildbucketV2() |
| builder = build_pb2.BuilderID(project='chromeos', bucket='general') |
| tags = [common_pb2.StringPair(key='cbb_config', |
| value='something-paladin')] |
| build_list = [build_pb2.Build(id=1234), |
| build_pb2.Build(id=2341)] |
| search_fn = self.PatchObject( |
| bbv2, 'SearchBuild', |
| return_value=rpc_pb2.SearchBuildsResponse(builds=build_list)) |
| status_fn = self.PatchObject(bbv2, 'GetBuildStatus') |
| bbv2.GetBuildHistory('something-paladin', 1, ignore_build_id=1000) |
| fake_predicate = rpc_pb2.BuildPredicate( |
| builder=builder, tags=tags, create_time=None, build=None) |
| search_fn.assert_called_with(fake_predicate, page_size=2) |
| status_fn.assert_called_with(1234) |
| |
| def testGetBuildHistoryOtherArgs(self): |
| """Test GetBuildHistory's processing of (args - ignore_build_id).""" |
| builder = build_pb2.BuilderID(project='chromeos', bucket='general') |
| tags = [common_pb2.StringPair(key='cbb_config', |
| value='something-paladin'), |
| common_pb2.StringPair(key='cbb_branch', |
| value='master')] |
| start_date = date(2019, 4, 16) |
| create_time = buildbucket_v2.DateToTimeRange(start_date) |
| fake_build = rpc_pb2.BuildRange(start_build_id=1234) |
| fake_predicate = rpc_pb2.BuildPredicate( |
| builder=builder, tags=tags, create_time=create_time, build=fake_build) |
| bbv2 = buildbucket_v2.BuildbucketV2() |
| search_fn = self.PatchObject(bbv2, 'SearchBuild') |
| self.PatchObject(bbv2, 'GetBuildStatus') |
| bbv2.GetBuildHistory('something-paladin', 10, start_date=start_date, |
| branch='master', start_build_id=1234) |
| search_fn.assert_called_once_with(fake_predicate, page_size=10) |
| |
| def testGetChildStatusesSuccess(self): |
| """Test GetChildStatuses when RPC succeeds.""" |
| bbv2 = buildbucket_v2.BuildbucketV2() |
| fake_search_build_response = rpc_pb2.SearchBuildsResponse() |
| self.PatchObject(bbv2, 'SearchBuild', |
| return_value=fake_search_build_response) |
| self.assertEqual(bbv2.GetChildStatuses(1234), []) |
| fake_build = build_pb2.Build(id=2341) |
| fake_search_build_response = rpc_pb2.SearchBuildsResponse( |
| builds=[fake_build]) |
| self.PatchObject(bbv2, 'SearchBuild', |
| return_value=fake_search_build_response) |
| get_build_status = self.PatchObject(bbv2, 'GetBuildStatus') |
| bbv2.GetChildStatuses(1234) |
| get_build_status.assert_called_once_with(2341) |
| |
| def testGetStageFailures(self): |
| """Test GetStageFailures logic.""" |
| bbv2 = buildbucket_v2.BuildbucketV2() |
| fake_build_status = {'buildbucket_id': 1234, |
| 'status': 'Passed!!', |
| 'build_config': 'something-paladin', |
| 'important': True} |
| fake_stages = [ |
| {'name': 'stage_1', 'status':constants.BUILDER_STATUS_PASSED}, |
| {'name': 'stage_2', 'status':constants.BUILDER_STATUS_FAILED}, |
| {'name': 'stage_3', 'status':constants.BUILDER_STATUS_FORGIVEN}, |
| ] |
| self.PatchObject(bbv2, 'GetBuildStatus', return_value=fake_build_status) |
| self.PatchObject(bbv2, 'GetBuildStages', return_value=fake_stages) |
| expected_answer = [ |
| {'stage_name': 'stage_2', |
| 'stage_status': constants.BUILDER_STATUS_FAILED, |
| 'buildbucket_id': 1234, |
| 'build_config': 'something-paladin', |
| 'build_status': 'Passed!!', |
| 'important': True} |
| ] |
| self.assertEqual(bbv2.GetStageFailures(1234), expected_answer) |
| |
| |
| class StaticFunctionsTest(cros_test_lib.MockTestCase): |
| """Test static functions in lib/buildbucket_v2.py.""" |
| # pylint: disable=attribute-defined-outside-init |
| |
| def testUpdateSelfBuildPropertiesNonBlocking(self): |
| self.logging_function = self.PatchObject( |
| logging, 'PrintKitchenSetBuildProperty') |
| buildbucket_v2.UpdateSelfBuildPropertiesNonBlocking('key', 'value') |
| self.logging_function.assert_called_with( |
| 'key', 'value') |
| |
| def testUpdateSelfCommonBuildProperties(self): |
| self.underlying_function = self.PatchObject( |
| buildbucket_v2, 'UpdateSelfBuildPropertiesNonBlocking') |
| fake_value = 123 |
| fake_id = 8921795536486453568 |
| buildbucket_v2.UpdateSelfCommonBuildProperties(critical=True) |
| self.underlying_function.assert_called_with('critical', True) |
| buildbucket_v2.UpdateSelfCommonBuildProperties(cidb_id=fake_value) |
| self.underlying_function.assert_called_with('cidb_id', fake_value) |
| buildbucket_v2.UpdateSelfCommonBuildProperties( |
| chrome_version=fake_value) |
| self.underlying_function.assert_called_with('chrome_version', fake_value) |
| buildbucket_v2.UpdateSelfCommonBuildProperties( |
| milestone_version=fake_value) |
| self.underlying_function.assert_called_with( |
| 'milestone_version', fake_value) |
| buildbucket_v2.UpdateSelfCommonBuildProperties( |
| platform_version=fake_value) |
| self.underlying_function.assert_called_with( |
| 'platform_version', fake_value) |
| buildbucket_v2.UpdateSelfCommonBuildProperties(full_version=fake_value) |
| self.underlying_function.assert_called_with('full_version', fake_value) |
| buildbucket_v2.UpdateSelfCommonBuildProperties(toolchain_url=fake_value) |
| self.underlying_function.assert_called_with('toolchain_url', fake_value) |
| buildbucket_v2.UpdateSelfCommonBuildProperties(build_type=fake_value) |
| self.underlying_function.assert_called_with('build_type', fake_value) |
| buildbucket_v2.UpdateSelfCommonBuildProperties(unibuild=True) |
| self.underlying_function.assert_called_with('unibuild', True) |
| buildbucket_v2.UpdateSelfCommonBuildProperties(suite_scheduling=True) |
| self.underlying_function.assert_called_with('suite_scheduling', True) |
| buildbucket_v2.UpdateSelfCommonBuildProperties( |
| killed_child_builds=[fake_id, fake_value]) |
| self.underlying_function.assert_called_with('killed_child_builds', |
| str([fake_id, fake_value])) |
| buildbucket_v2.UpdateSelfCommonBuildProperties(board='grunt') |
| self.underlying_function.assert_called_with('board', 'grunt') |
| buildbucket_v2.UpdateSelfCommonBuildProperties( |
| main_firmware_version='Google_Grunt.11031.62.0') |
| self.underlying_function.assert_called_with( |
| 'main_firmware_version', 'Google_Grunt.11031.62.0') |
| buildbucket_v2.UpdateSelfCommonBuildProperties( |
| ec_firmware_version='aleena_v2.1.108-9ca28c388') |
| self.underlying_function.assert_called_with( |
| 'ec_firmware_version', 'aleena_v2.1.108-9ca28c388') |
| |
| def testUpdateBuildMetadata(self): |
| fake_dict = {'version': {'chrome': 'chrome_version', |
| 'milestone': 'milestone_version', |
| 'platform': 'platform_version', |
| 'full': 'full_version'}, |
| 'toolchain-url': 'toolchain_url', |
| 'build_type': 'canary', |
| 'important': True, |
| 'unibuild': True} |
| self.PatchObject(metadata_lib.CBuildbotMetadata, 'GetDict', |
| return_value=fake_dict) |
| self.PatchObject(buildbucket_v2, 'UpdateSelfCommonBuildProperties') |
| fake_metadata = metadata_lib.CBuildbotMetadata() |
| buildbucket_v2.UpdateBuildMetadata(fake_metadata) |
| buildbucket_v2.UpdateSelfCommonBuildProperties.assert_called_with( |
| critical=True, chrome_version='chrome_version', |
| milestone_version='milestone_version', |
| platform_version='platform_version', |
| full_version='full_version', |
| toolchain_url='toolchain_url', |
| build_type='canary', |
| unibuild=True, |
| suite_scheduling=False, |
| channels=None) |
| |
| def testBuildStepToDict(self): |
| """Test the working of BuildStepToDict.""" |
| build_values = { |
| 'buildbucket_id': 1234, |
| 'build_config': 'something-paladin'} |
| start_time = Timestamp() |
| start_time.GetCurrentTime() |
| step = step_pb2.Step(name='stage_name', start_time=start_time, |
| status=2) |
| expected_result = { |
| 'name': 'stage_name', |
| 'start_time': datetime.fromtimestamp(start_time.seconds), |
| 'finish_time': None, |
| 'buildbucket_id': 1234, |
| 'status': constants.BUILDER_STATUS_INFLIGHT, |
| 'build_config': 'something-paladin'} |
| |
| self.assertEqual(buildbucket_v2.BuildStepToDict(step, build_values), |
| expected_result) |
| |
| def testDateToTimeRangeNoneInput(self): |
| self.assertIsNone(buildbucket_v2.DateToTimeRange(None)) |
| |
| def testDateToTimeRangeStartDate(self): |
| date_example = date(2019, 4, 15) |
| result = buildbucket_v2.DateToTimeRange(start_date=date_example) |
| self.assertEqual(result.start_time.seconds, 1555286400) |
| self.assertEqual(result.end_time.seconds, 0) |
| |
| def testDateToTimeRangeEndDate(self): |
| date_example = date(2019, 4, 15) |
| result = buildbucket_v2.DateToTimeRange(end_date=date_example) |
| self.assertEqual(result.end_time.seconds, 1555372740) |
| self.assertEqual(result.start_time.seconds, 0) |
