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.lib import cros_test_lib
from chromite.lib import parallel

@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)
    starting_dict = {'key1' : 1,
                     'key2' : '2',
                     '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)