buildbot/cbuildbot_metadata_unittest.py

#!/usr/bin/python
# Copyright (c) 2014 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.

"""Test the cbuildbot_archive module."""

import logging
import multiprocessing
import os
import sys

sys.path.insert(0, os.path.abspath('%s/../..' % os.path.dirname(__file__)))
from chromite.buildbot import cbuildbot_metadata
from chromite.buildbot import cbuildbot_results as results_lib
from chromite.buildbot import constants
from chromite.lib import cros_test_lib
from chromite.lib import parallel


class LocalBuilderStatusTest(cros_test_lib.TestCase):
  """Test the correctness of the various LocalBuilderStatus methods."""

  Results = results_lib.Results
  SUCCESS = results_lib.Results.SUCCESS

  def setUp(self):
    self.Results.Clear()

  def tearDown(self):
    self.Results.Clear()

  def testEmptyFail(self):
    """Sometimes there is no board-specific information."""
    self.Results.Record('Sync', self.SUCCESS)
    builder_status = cbuildbot_metadata.LocalBuilderStatus.Get()
    self.assertEqual({}, builder_status.board_status_map)
    self.assertEqual(constants.FINAL_STATUS_FAILED,
                     builder_status.GetBuilderStatus('lumpy-paladin'))
    self.assertEqual(constants.FINAL_STATUS_PASSED,
                     builder_status.GetBuilderStatus('master-paladin'))

  def testAllFail(self):
    """Failures in a shared stage cause specific stages to fail."""
    self.Results.Record('Sync', self.SUCCESS)
    self.Results.Record('BuildPackages', self.SUCCESS, board='lumpy')
    self.Results.Record('Completion', 'FAIL')
    builder_status = cbuildbot_metadata.LocalBuilderStatus.Get()
    self.assertEqual({'lumpy': constants.FINAL_STATUS_PASSED},
                     builder_status.board_status_map)
    self.assertEqual(constants.FINAL_STATUS_FAILED,
                     builder_status.GetBuilderStatus('lumpy-paladin'))
    self.assertEqual(constants.FINAL_STATUS_FAILED,
                     builder_status.GetBuilderStatus('master-paladin'))

  def testMixedFail(self):
    """Specific stages can pass/fail separately."""
    self.Results.Record('Sync', self.SUCCESS)
    self.Results.Record('BuildPackages [lumpy]', 'FAIL', board='lumpy')
    self.Results.Record('BuildPackages [stumpy]', self.SUCCESS, board='stumpy')
    builder_status = cbuildbot_metadata.LocalBuilderStatus.Get()
    self.assertEqual({'lumpy': constants.FINAL_STATUS_FAILED,
                      'stumpy': constants.FINAL_STATUS_PASSED},
                     builder_status.board_status_map)
    self.assertEqual(constants.FINAL_STATUS_FAILED,
                     builder_status.GetBuilderStatus('lumpy-paladin'))
    self.assertEqual(constants.FINAL_STATUS_PASSED,
                     builder_status.GetBuilderStatus('stumpy-paladin'))
    self.assertEqual(constants.FINAL_STATUS_FAILED,
                     builder_status.GetBuilderStatus('winky-paladin'))
    self.assertEqual(constants.FINAL_STATUS_PASSED,
                     builder_status.GetBuilderStatus('master-paladin'))


@cros_test_lib.NetworkTest()
class MetadataFetchTest(cros_test_lib.TestCase):
  """Test functions for fetching metadata from GS."""

  def testPaladinBuilder(self):
    bot, version = ('x86-mario-paladin', '5611.0.0')
    full_version = cbuildbot_metadata.FindLatestFullVersion(bot, version)
    self.assertEqual(full_version, 'R35-5611.0.0-rc2')
    metadata = cbuildbot_metadata.GetBuildMetadata(bot, full_version)
    metadata_dict = metadata._metadata_dict # pylint: disable=W0212
    self.assertEqual(metadata_dict['status']['status'], 'passed')


class MetadataTest(cros_test_lib.TestCase):
  """Tests the correctness of various metadata methods."""

  def testGetDict(self):
    starting_dict = {'key1': 1,
                     'key2': '2',
                     'cl_actions': [('a', 1), ('b', 2)]}
    metadata = cbuildbot_metadata.CBuildbotMetadata(starting_dict)
    ending_dict = metadata.GetDict()
    self.assertEqual(starting_dict, ending_dict)

  def testMultiprocessSafety(self):
    m = multiprocessing.Manager()
    metadata = cbuildbot_metadata.CBuildbotMetadata(multiprocess_manager=m)
    key_dict = {'key1': 1, 'key2': 2}
    starting_dict = {'key1': 1,
                     'key2': '2',
                     'key3': key_dict,
                     'cl_actions': [('a', 1), ('b', 2)]}

    # Test that UpdateWithDict is process-safe
    parallel.RunParallelSteps([lambda: metadata.UpdateWithDict(starting_dict)])
    ending_dict = metadata.GetDict()
    self.assertEqual(starting_dict, ending_dict)

    # Test that RecordCLAction is process-safe
    fake_change = cbuildbot_metadata.GerritPatchTuple(12345, 1, False)
    fake_action = ('asdf,')
    parallel.RunParallelSteps([lambda: metadata.RecordCLAction(fake_change,
                                                               fake_action)])
    ending_dict = metadata.GetDict()
    # Assert that an action was recorded.
    self.assertEqual(len(starting_dict['cl_actions']) + 1,
                     len(ending_dict['cl_actions']))


if __name__ == '__main__':
  cros_test_lib.main(level=logging.DEBUG)