blob: 2df23579a9b0e5a058d9c6436ac022aee3a85cec [file] [log] [blame]
# Copyright 2012 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Unittests for request_build.py."""
from chromite.third_party.google.protobuf.struct_pb2 import Struct
from chromite.third_party.infra_libs.buildbucket.proto import (
build_pb2,
builder_common_pb2,
common_pb2,
)
from chromite.lib import buildbucket_v2
from chromite.lib import config_lib
from chromite.lib import cros_test_lib
from chromite.lib import request_build
# Tests need internal access.
# pylint: disable=protected-access
class RequestBuildHelperTestsBase(cros_test_lib.MockTestCase):
"""Tests for RequestBuild."""
BRANCH = "test-branch"
PATCHES = ("5555", "6666")
BUILD_CONFIG_MIN = "amd64-generic-paladin-tryjob"
BUILD_CONFIG_MAX = "amd64-generic-paladin"
UNKNOWN_CONFIG = "unknown-config"
LUCI_BUILDER = "luci_build"
DISPLAY_LABEL = "display"
PASS_THROUGH_ARGS = ["funky", "cold", "medina"]
TEST_EMAIL = "explicit_email"
TEST_TEMPLATE = "explicit_template"
MASTER_CIDB_ID = "master_cidb_id"
MASTER_BUILDBUCKET_ID = "master_bb_id"
TEST_BUCKET = "luci.chromeos.general" # Use Prod bucket for network test.
EXTRA_PROPERTIES = {"full_version": "R84-13099.77.0"}
def _CreateJobMin(self):
return request_build.RequestBuild(build_config=self.BUILD_CONFIG_MIN)
def _CreateJobMax(self):
return request_build.RequestBuild(
build_config=self.BUILD_CONFIG_MAX,
luci_builder=self.LUCI_BUILDER,
display_label=self.DISPLAY_LABEL,
branch=self.BRANCH,
extra_args=self.PASS_THROUGH_ARGS,
extra_properties=self.EXTRA_PROPERTIES,
user_email=self.TEST_EMAIL,
email_template=self.TEST_TEMPLATE,
master_cidb_id=self.MASTER_CIDB_ID,
master_buildbucket_id=self.MASTER_BUILDBUCKET_ID,
bucket=self.TEST_BUCKET,
requested_bot="botname",
)
def _CreateJobUnknown(self):
return request_build.RequestBuild(
build_config=self.UNKNOWN_CONFIG,
display_label=self.DISPLAY_LABEL,
branch="master",
extra_args=[],
user_email="default_email",
master_buildbucket_id=None,
)
class RequestBuildHelperTestsMock(RequestBuildHelperTestsBase):
"""Perform real buildbucket requests against a fake instance."""
def setUp(self) -> None:
# This mocks out the class, then creates a return_value for a function
# on instances of it. We do this instead of just mocking out the
# function to ensure not real network requests are made in other parts
# of the class.
client_mock = self.PatchObject(buildbucket_v2, "BuildbucketV2")
client_mock().ScheduleBuilder.return_value = build_pb2.Build(
id=12345,
)
def testMinCreateRequestBody(self) -> None:
"""Verify our request body with min options."""
job = self._CreateJobMin()
self.assertEqual(
job.bucket, request_build.INTERNAL_SWARMING_BUILDBUCKET_BUCKET
)
self.assertEqual(job.luci_builder, config_lib.LUCI_BUILDER_TRY)
self.assertEqual(job.display_label, config_lib.DISPLAY_LABEL_TRYJOB)
body = job.CreateBuildRequest()
self.assertEqual(
builder_common_pb2.BuilderID(
project="chromeos",
bucket=request_build.INTERNAL_SWARMING_BUILDBUCKET_BUCKET,
builder=config_lib.LUCI_BUILDER_TRY,
),
body["builder"],
)
self.assertEqual(
[
common_pb2.StringPair(key="cbb_branch", value="master"),
common_pb2.StringPair(
key="cbb_config", value="amd64-generic-paladin-tryjob"
),
common_pb2.StringPair(key="cbb_display_label", value="tryjob"),
],
body["tags"],
)
def testMaxRequestBody(self) -> None:
"""Verify our request body with max options."""
job = self._CreateJobMax()
self.assertEqual(job.bucket, self.TEST_BUCKET)
self.assertEqual(job.luci_builder, self.LUCI_BUILDER)
self.assertEqual(job.display_label, "display")
body = job.CreateBuildRequest()
self.assertEqual(
builder_common_pb2.BuilderID(
project="chromeos",
bucket=self.TEST_BUCKET,
builder=self.LUCI_BUILDER,
),
body["builder"],
)
self.assertEqual(
[
common_pb2.StringPair(
key="buildset",
value="cros/parent_buildbucket_id/master_bb_id",
),
common_pb2.StringPair(key="cbb_branch", value="test-branch"),
common_pb2.StringPair(
key="cbb_config", value="amd64-generic-paladin"
),
common_pb2.StringPair(key="cbb_display_label", value="display"),
common_pb2.StringPair(key="cbb_email", value="explicit_email"),
common_pb2.StringPair(
key="cbb_master_build_id", value="master_cidb_id"
),
common_pb2.StringPair(
key="cbb_master_buildbucket_id", value="master_bb_id"
),
common_pb2.StringPair(
key="full_version", value="R84-13099.77.0"
),
common_pb2.StringPair(key="master", value="False"),
],
body["tags"],
)
props = {
"buildset": "cros/parent_buildbucket_id/master_bb_id",
"cbb_branch": "test-branch",
"cbb_config": "amd64-generic-paladin",
"cbb_display_label": "display",
"cbb_email": "explicit_email",
"cbb_master_build_id": "master_cidb_id",
"cbb_master_buildbucket_id": "master_bb_id",
"full_version": "R84-13099.77.0",
"master": "False",
}
test_properties = Struct()
test_properties.update({k: str(v) for k, v in props.items() if v})
test_properties.update({"cbb_extra_args": ["funky", "cold", "medina"]})
test_properties.update(
{
"email_notify": [
{
"email": "explicit_email",
"template": "explicit_template",
}
]
}
)
self.assertEqual(
test_properties,
body["properties"],
)
def testUnknownRequestBody(self) -> None:
"""Verify our request body with max options."""
job = self._CreateJobUnknown()
body = job.CreateBuildRequest()
self.assertEqual(
builder_common_pb2.BuilderID(
project="chromeos",
bucket=request_build.INTERNAL_SWARMING_BUILDBUCKET_BUCKET,
builder=config_lib.LUCI_BUILDER_TRY,
),
body["builder"],
)
self.assertEqual(
[
common_pb2.StringPair(key="cbb_branch", value="master"),
common_pb2.StringPair(key="cbb_config", value="unknown-config"),
common_pb2.StringPair(key="cbb_display_label", value="display"),
common_pb2.StringPair(key="cbb_email", value="default_email"),
],
body["tags"],
)
props = {
"cbb_display_label": "display",
"cbb_branch": "master",
"cbb_config": "unknown-config",
"cbb_email": "default_email",
}
test_properties = Struct()
test_properties.update({k: str(v) for k, v in props.items() if v})
test_properties.update({"cbb_extra_args": job.extra_args})
test_properties.update(
{
"email_notify": [
{
"email": "default_email",
"template": "default",
}
]
}
)
self.assertEqual(body["properties"], test_properties)
def testLogGeneration(self) -> None:
"""Validate an import log message."""
sb = request_build.ScheduledBuild(
"bucket", "buildbucket_id", "build_config", "url", "created_ts"
)
msg = (
request_build.RequestBuild.BUILDBUCKET_PUT_RESP_FORMAT
% sb._asdict()
)
expected = (
"Successfully sent PUT request to [buildbucket_bucket:bucket] "
"with [config:build_config] [buildbucket_id:buildbucket_id]."
)
# This test both validates that we can generate the log message, and
# that it's format hasn't changed. Since there are scripts that parse it
# via regex, the latter is important.
self.assertEqual(msg, expected)
class RequestBuildHelperTestsNetork(RequestBuildHelperTestsBase):
"""Perform real buildbucket requests against a test instance."""
def verifyBuildbucketRequest(
self,
buildbucket_id,
expected_bucket,
expected_tags,
expected_properties,
) -> None:
"""Verify the contents of a push to the TEST buildbucket instance.
Args:
buildbucket_id: Id to verify.
expected_bucket: Bucket the push was supposed to go to as a string.
expected_tags: List of buildbucket tags.
expected_properties: List of buildbucket properties.
"""
client = buildbucket_v2.BuildbucketV2(test_env=True)
request = client.GetBuild(buildbucket_id)
self.assertEqual(request.id, buildbucket_id)
self.assertEqual(request.builder.bucket, expected_bucket)
self.assertCountEqual(request.tags, expected_tags)
self.assertCountEqual(request.properties, expected_properties)
@cros_test_lib.pytestmark_network_test
def testMinTestBucket(self) -> None:
"""Talk to a test buildbucket instance with min job settings."""
job = self._CreateJobMin()
request = job.CreateBuildRequest()
client = buildbucket_v2.BuildbucketV2(test_env=True)
result = client.ScheduleBuild(
request_id=request.request_id,
builder=request.builder,
properties=request.properties,
tags=request.tags,
dimensions=request.dimensions,
)
props = {
"builder_name": "Try",
"properties": {
"cbb_branch": "master",
"cbb_config": "amd64-generic-paladin-tryjob",
"cbb_display_label": "tryjob",
"cbb_extra_args": [],
},
}
expected_properties = Struct()
expected_properties.update({k: str(v) for k, v in props.items() if v})
self.verifyBuildbucketRequest(
result.id,
"luci.chromeos.general",
[
common_pb2.StringPair(key="cbb_branch", value="master"),
common_pb2.StringPair(
key="cbb_config", value="amd64-generic-paladin-tryjob"
),
common_pb2.StringPair(key="cbb_display_label", value="tryjob"),
],
expected_properties,
)
@cros_test_lib.pytestmark_network_test
def testMaxTestBucket(self) -> None:
"""Talk to a test buildbucket instance with max job settings."""
job = self._CreateJobMax()
request = job.CreateBuildRequest()
client = buildbucket_v2.BuildbucketV2(test_env=True)
result = client.ScheduleBuild(
request_id=request.request_id,
builder=request.builder,
properties=request.properties,
tags=request.tags,
dimensions=request.dimensions,
)
props = {
"buildset": "cros/parent_buildbucket_id/master_bb_id",
"cbb_branch": "test-branch",
"cbb_config": "amd64-generic-paladin",
"cbb_display_label": "display",
"cbb_email": "explicit_email",
"cbb_extra_args": ["funky", "cold", "medina"],
"cbb_master_build_id": "master_cidb_id",
"cbb_master_buildbucket_id": "master_bb_id",
"master": "False",
}
expected_properties = Struct()
expected_properties.update({k: str(v) for k, v in props.items() if v})
self.verifyBuildbucketRequest(
result.id,
self.TEST_BUCKET,
[
common_pb2.StringPair(
key="buildset",
value="cros/parent_buildbucket_id/master_bb_id",
),
common_pb2.StringPair(key="cbb_branch", value="test-branch"),
common_pb2.StringPair(
key="cbb_config", value="amd64-generic-paladin"
),
common_pb2.StringPair(key="cbb_display_label", value="display"),
common_pb2.StringPair(key="cbb_email", value="explicit_email"),
common_pb2.StringPair(
key="cbb_master_build_id", value="master_cidb_id"
),
common_pb2.StringPair(
key="cbb_master_buildbucket_id", value="master_bb_id"
),
common_pb2.StringPair(
key="full_version", value="R84-13099.77.0"
),
common_pb2.StringPair(key="master", value="False"),
],
expected_properties,
)