cbuildbot/buildbucket_lib_unittest.py

# Copyright 2016 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_lib"""

from __future__ import print_function

import json
import mock

from chromite.cbuildbot import buildbucket_lib
from chromite.lib import cros_test_lib

class BuildbucketLibTest(cros_test_lib.TestCase):
  """Tests for buildbucket_lib."""

  def setUp(self):
    self.http = mock.MagicMock()
    self.success_response = {'status': 200}

  def testPutBuild(self):
    """Test PutBuild."""
    buildbucket_id = 'test_buildbucket_id'
    content = json.dumps({
        'build': {
            'id': buildbucket_id
        }
    })
    self.http.request.return_value = (self.success_response, content)
    body = json.dumps({
        'bucket': 'test-bucket',
    })

    result_content = buildbucket_lib.PutBuildBucket(
        body, self.http, False, False)
    self.assertEqual(buildbucket_lib.GetBuildId(result_content), buildbucket_id)

    # Test dryrun
    result_content_2 = buildbucket_lib.PutBuildBucket(
        body, self.http, False, True)
    self.assertIsNone(result_content_2)

  def testGetBuild(self):
    """Test GetBuild."""
    buildbucket_id = 'test_buildbucket_id'
    complete_status = 'COMPLETED'
    content = json.dumps({
        'build': {
            'id': buildbucket_id,
            'status': complete_status
        }
    })
    self.http.request.return_value = (self.success_response, content)
    result_content = buildbucket_lib.GetBuildBucket(
        buildbucket_id, self.http, False, False)
    self.assertEqual(buildbucket_lib.GetBuildStatus(result_content),
                     complete_status)

    # Test dryrun
    result_content_2 = buildbucket_lib.GetBuildBucket(
        buildbucket_id, self.http, False, True)
    self.assertIsNone(result_content_2)

  def testCancelBuild(self):
    """Test CancelBuild."""
    buildbucket_id = 'test_buildbucket_id'
    result = 'CANCELED'
    content = json.dumps({
        'build':{
            'id': buildbucket_id,
            'result': result
        }
    })
    self.http.request.return_value = (self.success_response, content)
    result_content = buildbucket_lib.CancelBuildBucket(
        buildbucket_id, self.http, False, False)
    self.assertEqual(buildbucket_lib.GetBuildResult(result_content), result)

    # Test dryrun
    result_content_2 = buildbucket_lib.CancelBuildBucket(
        buildbucket_id, self.http, False, True)
    self.assertIsNone(result_content_2)

  def testCancelBatchBuilds(self):
    """Test CancelBatchBuilds."""
    buildbucket_id_1 = 'test_buildbucket_id_1'
    buildbucket_id_2 = 'test_buildbucket_id_2'
    status = 'COMPLETED'
    result = 'CANCELED'
    error_reason = 'BUILD_IS_COMPLETED'
    content = json.dumps({
        'results':[{
            'build_id': buildbucket_id_1,
            'build': {
                'status': status,
                'result': result,
            }
        }, {
            'build_id': buildbucket_id_2,
            'error': {
                'message': "Cannot cancel a completed build",
                'reason': error_reason,
            }
        }]
    })
    self.http.request.return_value = (self.success_response, content)
    result_content = buildbucket_lib.CancelBatchBuildBucket(
        [buildbucket_id_1, buildbucket_id_2], self.http, False, False)

    result_map = buildbucket_lib.GetResultMap(result_content)
    self.assertTrue(buildbucket_lib.HasBuild(result_map[buildbucket_id_1]))
    self.assertTrue(buildbucket_lib.HasError(result_map[buildbucket_id_2]))
    self.assertEqual(buildbucket_lib.GetBuildStatus(
        result_map[buildbucket_id_1]), status)
    self.assertEqual(buildbucket_lib.GetErrorReason(
        result_map[buildbucket_id_2]), error_reason)

    # Test dryrun
    result_content_2 = buildbucket_lib.CancelBatchBuildBucket(
        [buildbucket_id_1, buildbucket_id_2], self.http, False, True)
    self.assertIsNone(result_content_2)