# -*- coding: utf-8 -*-
# Copyright (c) 2011 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.

"""Cros unit test library, with utility functions."""

from __future__ import print_function

import collections
import contextlib
import functools
import os
import re
import sys
import time
import unittest

import mock
import six
from six.moves import StringIO

from chromite.lib import cache
from chromite.lib import constants
from chromite.lib import commandline
from chromite.lib import cros_build_lib
from chromite.lib import cros_logging as logging
from chromite.lib import operation
from chromite.lib import osutils
from chromite.lib import partial_mock
from chromite.lib import terminal
from chromite.lib import timeout_util
from chromite.utils import outcap


# Define custom pytestmarks, allowing us to run/skip tests by category.
# Our Pytest marks are documented in chromite/pytest.ini.
# For more about marks, see https://docs.pytest.org/en/latest/mark.html
# Because Pytest is not always present outside the chroot, we must wrap
# our mark definitions in a try/except block.
# TODO(crbug.com/1058422): Once pytest is available in all runtime envs,
# add pytestmarks directly in test files.
try:
  import pytest  # pylint: disable=import-error
  pytest_skip = pytest.skip
  pytestmark_inside_only = pytest.mark.inside_only
  pytestmark_network_test = pytest.mark.network_test
  pytestmark_skip = pytest.mark.skip
  pytestmark_skipif = pytest.mark.skipif
except (ImportError, AttributeError):
  # If Pytest is not present, or too old to allow pytest.mark,
  # define custom pytestmarks as null functions for test files to use.
  null_decorator = lambda obj: obj
  pytest_skip = lambda allow_module_level: True
  pytestmark_inside_only = null_decorator
  pytestmark_network_test = null_decorator
  pytestmark_skip = null_decorator
  pytestmark_skipif = lambda condition, reason=None: None


Directory = collections.namedtuple('Directory', ['name', 'contents'])


class GlobalTestConfig(object):
  """Global configuration for tests."""

  # By default, disable all network tests.
  RUN_NETWORK_TESTS = False
  UPDATE_GENERATED_FILES = False
  NETWORK_TESTS_SKIPPED = 0


def NetworkTest(reason='Skipping network test (re-run w/--network)'):
  """Decorator for unit tests. Skip the test if --network is not specified."""
  def Decorator(test_item):
    @functools.wraps(test_item)
    @pytestmark_network_test
    def NetworkWrapper(*args, **kwargs):
      if not GlobalTestConfig.RUN_NETWORK_TESTS:
        GlobalTestConfig.NETWORK_TESTS_SKIPPED += 1
        raise unittest.SkipTest(reason)
      test_item(*args, **kwargs)

    # We can't check GlobalTestConfig.RUN_NETWORK_TESTS here because
    # __main__ hasn't run yet. Wrap each test so that we check the flag before
    # running it.
    if isinstance(test_item, type) and issubclass(test_item, TestCase):
      test_item.setUp = Decorator(test_item.setUp)
      return test_item
    else:
      return NetworkWrapper

  return Decorator


def _FlattenStructure(base_path, dir_struct):
  """Converts a directory structure to a list of paths."""
  flattened = []
  for obj in dir_struct:
    if isinstance(obj, Directory):
      new_base = os.path.join(base_path, obj.name).rstrip(os.sep)
      flattened.append(new_base + os.sep)
      flattened.extend(_FlattenStructure(new_base, obj.contents))
    else:
      assert isinstance(obj, six.string_types)
      flattened.append(os.path.join(base_path, obj))
  return flattened


def CreateOnDiskHierarchy(base_path, dir_struct):
  """Creates on-disk representation of an in-memory directory structure.

  Args:
    base_path: The absolute root of the directory structure.
    dir_struct: A recursively defined data structure that represents a
      directory tree.  The basic form is a list.  Elements can be file names or
      cros_test_lib.Directory objects.  The 'contents' attribute of Directory
      types is a directory structure representing the contents of the directory.
      Examples:
        - ['file1', 'file2']
        - ['file1', Directory('directory', ['deepfile1', 'deepfile2']), 'file2']
  """
  flattened = _FlattenStructure(base_path, dir_struct)
  for f in flattened:
    f = os.path.join(base_path, f)
    if f.endswith(os.sep):
      osutils.SafeMakedirs(f)
    else:
      osutils.Touch(f, makedirs=True)


def _VerifyDirectoryIterables(existing, expected):
  """Compare two iterables representing contents of a directory.

  Paths in |existing| and |expected| will be compared for exact match.

  Args:
    existing: An iterable containing paths that exist.
    expected: An iterable of paths that are expected.

  Raises:
    AssertionError when there is any divergence between |existing| and
    |expected|.
  """
  def FormatPaths(paths):
    return '\n'.join(sorted(paths))

  existing = set(existing)
  expected = set(expected)

  unexpected = existing - expected
  if unexpected:
    raise AssertionError('Found unexpected paths:\n%s'
                         % FormatPaths(unexpected))
  missing = expected - existing
  if missing:
    raise AssertionError('These files were expected but not found:\n%s'
                         % FormatPaths(missing))


def VerifyOnDiskHierarchy(base_path, dir_struct):
  """Verify that an on-disk directory tree exactly matches a given structure.

  Args:
    base_path: See CreateOnDiskHierarchy()
    dir_struct: See CreateOnDiskHierarchy()

  Raises:
    AssertionError when there is any divergence between the on-disk
    structure and the structure specified by 'dir_struct'.
  """
  expected = _FlattenStructure(base_path, dir_struct)
  _VerifyDirectoryIterables(osutils.DirectoryIterator(base_path), expected)


def VerifyTarball(tarball, dir_struct):
  """Compare the contents of a tarball against a directory structure.

  Args:
    tarball: Path to the tarball.
    dir_struct: See CreateOnDiskHierarchy()

  Raises:
    AssertionError when there is any divergence between the tarball and the
    structure specified by 'dir_struct'.
  """
  result = cros_build_lib.run(['tar', '-tf', tarball], capture_output=True,
                              encoding='utf-8')
  contents = result.stdout.splitlines()
  normalized = set()
  for p in contents:
    norm = os.path.normpath(p)
    if p.endswith('/'):
      norm += '/'
    if norm in normalized:
      raise AssertionError('Duplicate entry %r found in %r!' % (norm, tarball))
    normalized.add(norm)

  expected = _FlattenStructure('', dir_struct)
  _VerifyDirectoryIterables(normalized, expected)


class StackedSetup(type):
  """Metaclass to simplify unit testing and make it more robust.

  A metaclass alters the way that classes are initialized, enabling us to
  modify the class dictionary prior to the class being created. We use this
  feature here to modify the way that unit tests work a bit.

  This class does three things:
    1) When a test case is set up or torn down, we now run all setUp and
       tearDown methods in the inheritance tree.
    2) If a setUp or tearDown method fails, we still run tearDown methods
       for any test classes that were partially or completely set up.
    3) All test cases time out after TEST_CASE_TIMEOUT seconds.

  Use by adding this line before a class:
    @six.add_metaclass(StackedSetup)

  Since cros_test_lib.TestCase uses this metaclass, all derivatives of TestCase
  also inherit the above behavior (unless they override the metaclass attribute
  manually).
  """

  TEST_CASE_TIMEOUT = 10 * 60

  def __new__(cls, clsname, bases, scope):
    """Generate the new class with pointers to original funcs & our helpers"""
    if 'setUp' in scope:
      scope['__raw_setUp__'] = scope.pop('setUp')
    scope['setUp'] = cls._stacked_setUp

    if 'tearDown' in scope:
      scope['__raw_tearDown__'] = scope.pop('tearDown')
    scope['tearDown'] = cls._stacked_tearDown

    # Modify all test* methods to time out after TEST_CASE_TIMEOUT seconds.
    timeout = scope.get('TEST_CASE_TIMEOUT', StackedSetup.TEST_CASE_TIMEOUT)
    if timeout is not None:
      for name, func in scope.items():
        if name.startswith('test') and hasattr(func, '__call__'):
          wrapper = timeout_util.TimeoutDecorator(timeout)
          scope[name] = wrapper(func)

    return type.__new__(cls, clsname, bases, scope)

  @staticmethod
  def _walk_mro_stacking(obj, attr, reverse=False):
    """Walk the stacked classes (python method resolution order)"""
    iterator = iter if reverse else reversed
    methods = (getattr(x, attr, None) for x in iterator(obj.__class__.__mro__))
    seen = set()
    for method in (x for x in methods if x):
      method = getattr(method, 'im_func', method)
      if method not in seen:
        seen.add(method)
        yield method

  @staticmethod
  def _stacked_setUp(obj):
    """Run all the setUp funcs; if any fail, run all the tearDown funcs"""
    obj.__test_was_run__ = False
    try:
      for target in StackedSetup._walk_mro_stacking(obj, '__raw_setUp__'):
        target(obj)
    except:
      # TestCase doesn't trigger tearDowns if setUp failed; thus
      # manually force it ourselves to ensure cleanup occurs.
      StackedSetup._stacked_tearDown(obj)
      raise

    # Now mark the object as fully setUp; this is done so that
    # any last minute assertions in tearDown can know if they should
    # run or not.
    obj.__test_was_run__ = True

  @staticmethod
  def _stacked_tearDown(obj):
    """Run all the tearDown funcs; if any fail, we move on to the next one"""
    exc_info = None
    for target in StackedSetup._walk_mro_stacking(obj, '__raw_tearDown__',
                                                  True):
      # pylint: disable=bare-except
      try:
        target(obj)
      except:
        # Preserve the exception, throw it after running
        # all tearDowns; we throw just the first also.  We suppress
        # pylint's warning here since it can't understand that we're
        # actually raising the exception, just in a nonstandard way.
        if exc_info is None:
          exc_info = sys.exc_info()

    if exc_info:
      # Chuck the saved exception, w/ the same TB from
      # when it occurred.
      six.reraise(exc_info[0], exc_info[1], exc_info[2])


class TruthTable(object):
  """Class to represent a boolean truth table, useful in unit tests.

  If you find yourself testing the behavior of some function that should
  basically follow the behavior of a particular truth table, then this class
  can allow you to fully test that function without being overly verbose
  in the unit test code.

  The following usage is supported on a constructed TruthTable:
  1) Iterate over input lines of the truth table, expressed as tuples of
  bools.
  2) Access a particular input line by index, expressed as a tuple of bools.
  3) Access the expected output for a set of inputs.

  For example, say function "Foo" in module "mod" should consists of the
  following code:

  def Foo(A, B, C):
    return A and B and not C

  In the unittest for Foo, do this:

  def testFoo(self):
    truth_table = cros_test_lib.TruthTable(inputs=[(True, True, True)])
    for inputs in truth_table:
      a, b, c = inputs
      result = mod.Foo(a, b, c)
      self.assertEqual(result, truth_table.GetOutput(inputs))
  """

  class TruthTableInputIterator(object):
    """Class to support iteration over inputs of a TruthTable."""

    def __init__(self, truth_table):
      self.truth_table = truth_table
      self.next_line = 0

    def __iter__(self):
      return self

    def __next__(self):
      if self.next_line < self.truth_table.num_lines:
        self.next_line += 1
        return self.truth_table.GetInputs(self.next_line - 1)
      else:
        raise StopIteration()

    # Python 2 glue.
    next = __next__

  def __init__(self, inputs, input_result=True):
    """Construct a TruthTable from given inputs.

    Args:
      inputs: Iterable of input lines, each expressed as a tuple of bools.
        Each tuple must have the same length.
      input_result: The output intended for each specified input.  For
        truth tables that mostly output True it is more concise to specify
        the false inputs and then set input_result to False.
    """
    # At least one input required.
    if not inputs:
      raise ValueError('Inputs required to construct TruthTable.')

    # Save each input tuple in a set.  Also confirm that the length
    # of each input tuple is the same.
    self.dimension = len(inputs[0])
    self.num_lines = pow(2, self.dimension)
    self.expected_inputs = set()
    self.expected_inputs_result = input_result

    for input_vals in inputs:
      if len(input_vals) != self.dimension:
        raise ValueError('All TruthTable inputs must have same dimension.')

      self.expected_inputs.add(input_vals)

    # Start generator index at 0.
    self.next_line = 0

  def __len__(self):
    return self.num_lines

  def __iter__(self):
    return self.TruthTableInputIterator(self)

  def GetInputs(self, inputs_index):
    """Get the input line at the given input index.

    Args:
      inputs_index: Following must hold: 0 <= inputs_index < self.num_lines.

    Returns:
      Tuple of bools representing one line of inputs.
    """
    if inputs_index >= 0 and inputs_index < self.num_lines:
      line_values = []

      # Iterate through each column in truth table.  Any order will
      # produce a valid truth table, but going backward through
      # columns will produce the traditional truth table ordering.
      # For 2-dimensional example: F,F then F,T then T,F then T,T.
      for col in range(self.dimension - 1, -1, -1):
        line_values.append(bool(inputs_index // pow(2, col) % 2))

      return tuple(line_values)

    raise ValueError('This truth table has no line at index %r.' % inputs_index)

  def GetOutput(self, inputs):
    """Get the boolean output for the given inputs.

    Args:
      inputs: Tuple of bools, length must be equal to self.dimension.

    Returns:
      bool value representing truth table output for given inputs.
    """
    if not isinstance(inputs, tuple):
      raise TypeError('Truth table inputs must be specified as a tuple.')

    if not len(inputs) == self.dimension:
      raise ValueError('Truth table inputs must match table dimension.')

    return self.expected_inputs_result == (inputs in self.expected_inputs)


class EasyAttr(dict):
  """Convenient class for simulating objects with attributes in tests.

  An EasyAttr object can be created with any attributes initialized very
  easily.  Examples:

  1) An object with .id=45 and .name="Joe":
  testobj = EasyAttr(id=45, name="Joe")
  2) An object with .title.text="Big" and .owner.text="Joe":
  testobj = EasyAttr(title=EasyAttr(text="Big"), owner=EasyAttr(text="Joe"))
  """

  __slots__ = ()

  def __getattr__(self, attr):
    try:
      return self[attr]
    except KeyError:
      raise AttributeError(attr)

  def __delattr__(self, attr):
    try:
      self.pop(attr)
    except KeyError:
      raise AttributeError(attr)

  def __setattr__(self, attr, value):
    self[attr] = value

  def __dir__(self):
    return list(self.keys())


class LogFilter(logging.Filter):
  """A simple log filter that intercepts log messages and stores them."""

  def __init__(self):
    logging.Filter.__init__(self)
    self.messages = StringIO()

  def filter(self, record):
    self.messages.write(record.getMessage() + '\n')
    # Return False to prevent the message from being displayed.
    return False


class LoggingCapturer(object):
  """Captures all messages emitted by the logging module."""

  def __init__(self, logger_name='', log_level=logging.DEBUG):
    self._log_filter = LogFilter()
    self._old_level = None
    self._log_level = log_level
    self.logger_name = logger_name

  def __enter__(self):
    self.StartCapturing()
    return self

  def __exit__(self, exc_type, exc_val, exc_tb):
    self.StopCapturing()

  def StartCapturing(self):
    """Begin capturing logging messages."""
    logger = logging.getLogger(self.logger_name)
    self._old_level = logger.getEffectiveLevel()
    logger.setLevel(self._log_level)
    logger.addFilter(self._log_filter)

  def StopCapturing(self):
    """Stop capturing logging messages."""
    logger = logging.getLogger(self.logger_name)
    logger.setLevel(self._old_level)
    logger.removeFilter(self._log_filter)

  @property
  def messages(self):
    return self._log_filter.messages.getvalue()

  def LogsMatch(self, regex):
    """Checks whether the logs match a given regex."""
    match = re.search(regex, self.messages, re.MULTILINE)
    return match is not None

  def LogsContain(self, msg):
    """Checks whether the logs contain a given string."""
    return self.LogsMatch(re.escape(msg))


@six.add_metaclass(StackedSetup)
class TestCase(unittest.TestCase):
  """Basic chromite test case.

  Provides sane setUp/tearDown logic so that tearDown is correctly cleaned up.

  Takes care of saving/restoring process-wide settings like the environment so
  that sub-tests don't have to worry about gettings this right.

  Also includes additional assert helpers beyond python stdlib.
  """

  # List of vars chromite is globally sensitive to and that should
  # be suppressed for tests.
  ENVIRON_VARIABLE_SUPPRESSIONS = ('CROS_CACHEDIR',)

  # The default diff is limited to 8 rows (of 80 cols).  Make this unlimited
  # so we always see the output.  If it's too much, people can use loggers or
  # pagers to scroll.
  maxDiff = None

  def __init__(self, *args, **kwargs):
    unittest.TestCase.__init__(self, *args, **kwargs)
    # This is set to keep pylint from complaining.
    self.__test_was_run__ = False

  @staticmethod
  def _CheckTestEnv(msg):
    """Sanity check the environment.  https://crbug.com/1015450"""
    # Note: We use print+sys.exit here instead of logging/Die because it might
    # cause errors in tests that expect their own setUp to run before their own
    # tearDown executes.  By failing in the core funcs, we violate that.
    st = os.stat('/')
    if st.st_mode & 0o7777 != 0o755:
      print('%s %s\nError: The root directory has broken permissions: %o\n'
            'Fix with: sudo chmod 755 /' % (sys.argv[0], msg, st.st_mode),
            file=sys.stderr)
      sys.exit(1)
    if st.st_uid or st.st_gid:
      print('%s %s\nError: The root directory has broken ownership: %i:%i'
            ' (should be 0:0)\nFix with: sudo chown 0:0 /' %
            (sys.argv[0], msg, st.st_uid, st.st_gid), file=sys.stderr)
      sys.exit(1)

  def setUp(self):
    self._CheckTestEnv('%s.setUp' % (self.id(),))

    self.__saved_env__ = os.environ.copy()
    self.__saved_cwd__ = os.getcwd()
    self.__saved_umask__ = os.umask(0o22)
    for x in self.ENVIRON_VARIABLE_SUPPRESSIONS:
      os.environ.pop(x, None)
    # Force all log lines in tests to include ANSI color prefixes, since it can
    # be configured per-user.
    os.environ['NOCOLOR'] = 'no'

  def tearDown(self):
    self._CheckTestEnv('%s.tearDown' % (self.id(),))

    osutils.SetEnvironment(self.__saved_env__)
    os.chdir(self.__saved_cwd__)
    os.umask(self.__saved_umask__)

  def id(self):
    """Return a name that can be passed in via the command line."""
    return '%s.%s' % (self.__class__.__name__, self._testMethodName)

  def __str__(self):
    """Return a pretty name that can be passed in via the command line."""
    return '[%s] %s' % (self.__module__, self.id())

  def assertRaises2(self, exception, functor, *args, **kwargs):
    """Like assertRaises, just with checking of the exception.

    Args:
      exception: The expected exception type to intecept.
      functor: The function to invoke.
      args: Positional args to pass to the function.
      kwargs: Optional args to pass to the function.  Note we pull
        exact_kls, msg, and check_attrs from these kwargs.
      exact_kls: If given, the exception raise must be *exactly* that class
        type; derivatives are a failure.
      check_attrs: If given, a mapping of attribute -> value to assert on
        the resultant exception.  Thus if you wanted to catch a ENOENT, you
        would do:
          assertRaises2(EnvironmentError, func, args,
                        check_attrs={'errno': errno.ENOENT})
      ex_msg: A substring that should be in the stringified exception.
      msg: The error message to be displayed if the exception isn't raised.
        If not given, a suitable one is defaulted to.
      returns: The exception object.
    """
    exact_kls = kwargs.pop('exact_kls', None)
    check_attrs = kwargs.pop('check_attrs', {})
    ex_msg = kwargs.pop('ex_msg', None)
    msg = kwargs.pop('msg', None)
    if msg is None:
      msg = ("%s(*%r, **%r) didn't throw an exception"
             % (functor.__name__, args, kwargs))
    try:
      functor(*args, **kwargs)
      raise AssertionError(msg)
    except exception as e:
      if ex_msg:
        self.assertIn(ex_msg, str(e))
      if exact_kls:
        self.assertEqual(e.__class__, exception)
      bad = []
      for attr, required in check_attrs.items():
        self.assertTrue(hasattr(e, attr),
                        msg='%s lacks attr %s' % (e, attr))
        value = getattr(e, attr)
        if value != required:
          bad.append('%s attr is %s, needed to be %s'
                     % (attr, value, required))
      if bad:
        raise AssertionError('\n'.join(bad))
      return e

  def assertExists(self, path, msg=None):
    """Make sure |path| exists"""
    if os.path.exists(path):
      return

    if msg is None:
      msg = ['path is missing: %s' % path]
      while path != '/':
        path = os.path.dirname(path)
        if not path:
          # If we're given something like "foo", abort once we get to "".
          break
        result = os.path.exists(path)
        msg.append('\tos.path.exists(%s): %s' % (path, result))
        if result:
          msg.append('\tcontents: %r' % os.listdir(path))
          break
      msg = '\n'.join(msg)

    raise self.failureException(msg)

  def assertNotExists(self, path, msg=None):
    """Make sure |path| does not exist"""
    if not os.path.exists(path):
      return

    if msg is None:
      msg = 'path exists when it should not: %s' % (path,)

    raise self.failureException(msg)

  def assertStartsWith(self, s, prefix, msg=None):
    """Asserts that |s| starts with |prefix|.

    This function should be preferred over assertTrue(s.startswith(prefix)) for
    it produces better error failure message than the other.
    """
    if s.startswith(prefix):
      return

    if msg is None:
      msg = '%s does not starts with %s' % (s, prefix)

    raise self.failureException(msg)

  def assertEndsWith(self, s, suffix, msg=None):
    """Asserts that |s| ends with |suffix|.

    This function should be preferred over assertTrue(s.endswith(suffix)) for
    it produces better error failure message than the other.
    """
    if s.endswith(suffix):
      return

    if msg is None:
      msg = '%s does not starts with %s' % (s, suffix)

    raise self.failureException(msg)

  def GetSequenceDiff(self, seq1, seq2):
    """Get a string describing the difference between two sequences.

    Args:
      seq1: First sequence to compare.
      seq2: Second sequence to compare.

    Returns:
      A string that describes how the two sequences differ.
    """
    try:
      self.assertSequenceEqual(seq1, seq2)
    except AssertionError as ex:
      return str(ex)
    else:
      return 'no differences'

  # Upstream deprecated these in Python 3, but left them in Python 2.
  # Deprecate them ourselves to help with migration.  We can delete these
  # once upstream drops them.
  def _disable(deprecated, replacement):  # pylint: disable=no-self-argument
    def disable_func(*_args, **_kwargs):
      raise RuntimeError('%s() is removed in Python 3; use %s() instead' %
                         (deprecated, replacement))
    return disable_func

  assertEquals = _disable('assertEquals', 'assertEqual')
  assertNotEquals = _disable('assertNotEquals', 'assertNotEqual')
  assertAlmostEquals = _disable('assertAlmostEquals', 'assertAlmostEqual')
  assertNotAlmostEquals = _disable('assertNotAlmostEquals',
                                   'assertNotAlmostEqual')
  assert_ = _disable('assert_', 'assertTrue')
  failUnlessEqual = _disable('failUnlessEqual', 'assertEqual')
  failIfEqual = _disable('failIfEqual', 'assertNotEqual')
  failUnlessAlmostEqual = _disable('failUnlessAlmostEqual', 'assertAlmostEqual')
  failIfAlmostEqual = _disable('failIfAlmostEqual', 'assertNotAlmostEqual')
  failUnless = _disable('failUnless', 'assertTrue')
  failUnlessRaises = _disable('failUnlessRaises', 'assertRaises')
  failIf = _disable('failIf', 'assertFalse')

  # Python 3 renamed these.
  if sys.version_info.major < 3:
    assertCountEqual = unittest.TestCase.assertItemsEqual
    assertRaisesRegex = unittest.TestCase.assertRaisesRegexp
    assertRegex = unittest.TestCase.assertRegexpMatches

  assertItemsEqual = _disable('assertItemsEqual', 'assertCountEqual')
  assertRaisesRegexp = _disable('assertRaisesRegexp', 'assertRaisesRegex')
  assertRegexpMatches = _disable('assertRegexpMatches', 'assertRegex')


class LoggingTestCase(TestCase):
  """Base class for logging capturer test cases."""

  def AssertLogsMatch(self, log_capturer, regex, inverted=False):
    """Verifies a regex matches the logs."""
    assert_msg = '%r not found in %r' % (regex, log_capturer.messages)
    assert_fn = self.assertTrue
    if inverted:
      assert_msg = '%r found in %r' % (regex, log_capturer.messages)
      assert_fn = self.assertFalse

    assert_fn(log_capturer.LogsMatch(regex), msg=assert_msg)

  def AssertLogsContain(self, log_capturer, msg, inverted=False):
    """Verifies a message is contained in the logs."""
    return self.AssertLogsMatch(log_capturer, re.escape(msg), inverted=inverted)


class OutputTestCase(TestCase):
  """Base class for cros unit tests with utility methods."""

  # These work with error output from operation module.
  ERROR_MSG_RE = re.compile(r'^\033\[1;%dm(.+?)(?:\033\[0m)+$' %
                            (30 + terminal.Color.RED,), re.DOTALL)
  WARNING_MSG_RE = re.compile(r'^\033\[1;%dm(.+?)(?:\033\[0m)+$' %
                              (30 + terminal.Color.YELLOW,), re.DOTALL)

  def __init__(self, *args, **kwargs):
    """Base class __init__ takes a second argument."""
    TestCase.__init__(self, *args, **kwargs)
    self._output_capturer = None

  def OutputCapturer(self, *args, **kwargs):
    """Create and return OutputCapturer object."""
    self._output_capturer = outcap.OutputCapturer(*args, **kwargs)
    return self._output_capturer

  def _GetOutputCapt(self):
    """Internal access to existing OutputCapturer.

    Raises RuntimeError if output capturing was never on.
    """
    if self._output_capturer:
      return self._output_capturer

    raise RuntimeError('Output capturing was never turned on for this test.')

  def _GenCheckMsgFunc(self, prefix_re, line_re):
    """Return boolean func to check a line given |prefix_re| and |line_re|."""
    def _method(line):
      if prefix_re:
        # Prefix regexp will strip off prefix (and suffix) from line.
        match = prefix_re.search(line)

        if match:
          line = match.group(1)
        else:
          return False

      return line_re.search(line) if line_re else True

    if isinstance(prefix_re, str):
      prefix_re = re.compile(prefix_re)
    if isinstance(line_re, str):
      line_re = re.compile(line_re)

    # Provide a description of what this function looks for in a line.  Error
    # messages can make use of this.
    _method.description = None
    if prefix_re and line_re:
      _method.description = ('line matching prefix regexp %r then regexp %r' %
                             (prefix_re.pattern, line_re.pattern))
    elif prefix_re:
      _method.description = 'line matching prefix regexp %r' % prefix_re.pattern
    elif line_re:
      _method.description = 'line matching regexp %r' % line_re.pattern
    else:
      raise RuntimeError('Nonsensical usage of _GenCheckMsgFunc: '
                         'no prefix_re or line_re')

    return _method

  def _ContainsMsgLine(self, lines, msg_check_func):
    return any(msg_check_func(ln) for ln in lines)

  def _GenOutputDescription(self, check_stdout, check_stderr):
    # Some extra logic to make an error message useful.
    if check_stdout and check_stderr:
      return 'stdout or stderr'
    elif check_stdout:
      return 'stdout'
    elif check_stderr:
      return 'stderr'

  def _AssertOutputContainsMsg(self, check_msg_func, invert,
                               check_stdout, check_stderr):
    assert check_stdout or check_stderr

    lines = []
    if check_stdout:
      lines.extend(self._GetOutputCapt().GetStdoutLines())
    if check_stderr:
      lines.extend(self._GetOutputCapt().GetStderrLines())

    result = self._ContainsMsgLine(lines, check_msg_func)

    # Some extra logic to make an error message useful.
    output_desc = self._GenOutputDescription(check_stdout, check_stderr)

    if invert:
      msg = ('expected %s to not contain %s,\nbut found it in:\n%s' %
             (output_desc, check_msg_func.description, lines))
      self.assertFalse(result, msg=msg)
    else:
      msg = ('expected %s to contain %s,\nbut did not find it in:\n%s' %
             (output_desc, check_msg_func.description, lines))
      self.assertTrue(result, msg=msg)

  def AssertOutputContainsError(self, regexp=None, invert=False,
                                check_stdout=True, check_stderr=False):
    """Assert requested output contains at least one error line.

    If |regexp| is non-null, then the error line must also match it.
    If |invert| is true, then assert the line is NOT found.

    Raises RuntimeError if output capturing was never on for this test.
    """
    check_msg_func = self._GenCheckMsgFunc(self.ERROR_MSG_RE, regexp)
    return self._AssertOutputContainsMsg(check_msg_func, invert,
                                         check_stdout, check_stderr)

  def AssertOutputContainsWarning(self, regexp=None, invert=False,
                                  check_stdout=True, check_stderr=False):
    """Assert requested output contains at least one warning line.

    If |regexp| is non-null, then the warning line must also match it.
    If |invert| is true, then assert the line is NOT found.

    Raises RuntimeError if output capturing was never on for this test.
    """
    check_msg_func = self._GenCheckMsgFunc(self.WARNING_MSG_RE, regexp)
    return self._AssertOutputContainsMsg(check_msg_func, invert,
                                         check_stdout, check_stderr)

  def AssertOutputContainsLine(self, regexp, invert=False,
                               check_stdout=True, check_stderr=False):
    """Assert requested output contains line matching |regexp|.

    If |invert| is true, then assert the line is NOT found.

    Raises RuntimeError if output capturing was never on for this test.
    """
    check_msg_func = self._GenCheckMsgFunc(None, regexp)
    return self._AssertOutputContainsMsg(check_msg_func, invert,
                                         check_stdout, check_stderr)

  def _AssertOutputEndsInMsg(self, check_msg_func,
                             check_stdout, check_stderr):
    """Pass if requested output(s) ends(end) with an error message."""
    assert check_stdout or check_stderr

    lines = []
    if check_stdout:
      stdout_lines = self._GetOutputCapt().GetStdoutLines(include_empties=False)
      if stdout_lines:
        lines.append(stdout_lines[-1])
    if check_stderr:
      stderr_lines = self._GetOutputCapt().GetStderrLines(include_empties=False)
      if stderr_lines:
        lines.append(stderr_lines[-1])

    result = self._ContainsMsgLine(lines, check_msg_func)

    # Some extra logic to make an error message useful.
    output_desc = self._GenOutputDescription(check_stdout, check_stderr)

    msg = ('expected %s to end with %s,\nbut did not find it in:\n%s' %
           (output_desc, check_msg_func.description, lines))
    self.assertTrue(result, msg=msg)

  def AssertOutputEndsInError(self, regexp=None,
                              check_stdout=True, check_stderr=False):
    """Assert requested output ends in error line.

    If |regexp| is non-null, then the error line must also match it.

    Raises RuntimeError if output capturing was never on for this test.
    """
    check_msg_func = self._GenCheckMsgFunc(self.ERROR_MSG_RE, regexp)
    return self._AssertOutputEndsInMsg(check_msg_func,
                                       check_stdout, check_stderr)

  def AssertOutputEndsInWarning(self, regexp=None,
                                check_stdout=True, check_stderr=False):
    """Assert requested output ends in warning line.

    If |regexp| is non-null, then the warning line must also match it.

    Raises RuntimeError if output capturing was never on for this test.
    """
    check_msg_func = self._GenCheckMsgFunc(self.WARNING_MSG_RE, regexp)
    return self._AssertOutputEndsInMsg(check_msg_func,
                                       check_stdout, check_stderr)

  def AssertOutputEndsInLine(self, regexp,
                             check_stdout=True, check_stderr=False):
    """Assert requested output ends in line matching |regexp|.

    Raises RuntimeError if output capturing was never on for this test.
    """
    check_msg_func = self._GenCheckMsgFunc(None, regexp)
    return self._AssertOutputEndsInMsg(check_msg_func,
                                       check_stdout, check_stderr)

  def FuncCatchSystemExit(self, func, *args, **kwargs):
    """Run |func| with |args| and |kwargs| and catch SystemExit.

    Return tuple (return value or None, SystemExit number code or None).
    """
    try:
      returnval = func(*args, **kwargs)

      return returnval, None
    except SystemExit as ex:
      exit_code = ex.args[0]
      return None, exit_code

  def AssertFuncSystemExitZero(self, func, *args, **kwargs):
    """Run |func| with |args| and |kwargs| catching SystemExit.

    If the func does not raise a SystemExit with exit code 0 then assert.
    """
    exit_code = self.FuncCatchSystemExit(func, *args, **kwargs)[1]
    self.assertIsNot(exit_code, None,
                     msg='Expected system exit code 0, but caught none')
    self.assertEqual(exit_code, 0,
                     msg=('Expected system exit code 0, but caught %d' %
                          exit_code))

  def AssertFuncSystemExitNonZero(self, func, *args, **kwargs):
    """Run |func| with |args| and |kwargs| catching SystemExit.

    If the func does not raise a non-zero SystemExit code then assert.
    """
    exit_code = self.FuncCatchSystemExit(func, *args, **kwargs)[1]
    self.assertIsNot(exit_code, None,
                     msg='Expected non-zero system exit code, but caught none')
    self.assertNotEqual(exit_code, 0,
                        msg=('Expected non-zero system exit code, but caught %d'
                             % exit_code))

  def AssertRaisesAndReturn(self, error, func, *args, **kwargs):
    """Like assertRaises, but return exception raised."""
    try:
      func(*args, **kwargs)
      self.fail(msg='Expected %s but got none' % error)
    except error as ex:
      return ex


class TempDirTestCase(TestCase):
  """Mixin used to give each test a tempdir that is cleansed upon finish"""

  # Whether to delete tempdir used by this test. cf: SkipCleanup.
  DELETE = True
  _NO_DELETE_TEMPDIR_OBJ = None

  def __init__(self, *args, **kwargs):
    TestCase.__init__(self, *args, **kwargs)
    self.tempdir = None
    self._tempdir_obj = None

  @classmethod
  def SkipCleanup(cls):
    """Leave behind tempdirs created by instances of this class.

    Calling this function ensures that all future instances will leak their
    temporary directories. Additionally, all future temporary directories will
    be created inside one top level temporary directory, so that you can easily
    blow them away when you're done.
    Currently, this function is pretty stupid. You should call it *before*
    creating any instances.

    Returns:
      Path to a temporary directory that contains all future temporary
      directories created by instances of this class.
    """
    cls.DELETE = False
    cls._NO_DELETE_TEMPDIR_OBJ = osutils.TempDir(
        prefix='chromite.test_no_cleanup',
        set_global=True,
        delete=cls.DELETE)
    logging.info('%s requested to SkipCleanup. Will leak %s',
                 cls.__name__, cls._NO_DELETE_TEMPDIR_OBJ.tempdir)
    return cls._NO_DELETE_TEMPDIR_OBJ.tempdir

  def setUp(self):
    self._tempdir_obj = osutils.TempDir(prefix='chromite.test', set_global=True,
                                        delete=self.DELETE)
    self.tempdir = self._tempdir_obj.tempdir
    # We must use addCleanup here so that inheriting TestCase classes can use
    # addCleanup with the guarantee that the tempdir will be cleand up _after_
    # their addCleanup has run. TearDown runs before cleanup functions.
    self.addCleanup(self._CleanTempDir)

  def _CleanTempDir(self):
    if self._tempdir_obj is not None:
      self._tempdir_obj.Cleanup()
      self._tempdir_obj = None
      self.tempdir = None

  def ExpectRootOwnedFiles(self):
    """Tells us that we may need to clean up root owned files."""
    if self._tempdir_obj is not None:
      self._tempdir_obj.SetSudoRm()

  def assertFileContents(self, file_path, content):
    """Assert that the file contains the given content."""
    self.assertExists(file_path)
    read_content = osutils.ReadFile(file_path)
    self.assertEqual(read_content, content)

  def assertTempFileContents(self, file_path, content):
    """Assert that a file in the temp directory contains the given content."""
    self.assertFileContents(os.path.join(self.tempdir, file_path), content)

  def ReadTempFile(self, path):
    """Read a given file from the temp directory.

    Args:
      path: The path relative to the temp directory to read.
    """
    return osutils.ReadFile(os.path.join(self.tempdir, path))

  def WriteTempFile(self, path, content, **kwargs):
    """Write the given content to the temp directory

    Args:
      path: The path relative to the temp directory to write to.
      content: Content to write. May be either an iterable, or a string.
      kwargs: Additional args to pass to osutils.WriteFile.
    """
    osutils.WriteFile(os.path.join(self.tempdir, path), content, **kwargs)


class FakeSDKCache(object):
  """Creates a fake SDK Cache."""

  def __init__(self, cache_dir, sdk_version='12225.0.0'):
    """Creates a fake SDK Cache.

    Args:
      cache_dir: The top level cache directory to use.
      sdk_version: The SDK Version.
    """
    self.cache_dir = cache_dir
    # Sets the SDK Version.
    self.sdk_version = sdk_version
    os.environ['%SDK_VERSION'] = sdk_version
    # Defines the path for the fake SDK Symlink Cache. (No backing tarball cache
    # is needed.)
    self.symlink_cache_path = os.path.join(self.cache_dir, 'chrome-sdk',
                                           'symlinks')
    # Creates an SDK SymlinkCache instance.
    self.symlink_cache = cache.DiskCache(self.symlink_cache_path)

  def CreateCacheReference(self, board, key):
    """Creates the Cache Reference.

    Args:
      board: The board to use.
      key: The key of the item in the tarball cache.

    Returns:
      Path to the cache directory.
    """
    # Adds the cache path at the key.
    return self.symlink_cache.Lookup((board, self.sdk_version, key)).path


class MockTestCase(TestCase):
  """Python-mock based test case; compatible with StackedSetup"""

  def setUp(self):
    self._patchers = []

  def tearDown(self):
    # We can't just run stopall() by itself, and need to stop our patchers
    # manually since stopall() doesn't handle repatching.
    cros_build_lib.SafeRun([p.stop for p in reversed(self._patchers)] +
                           [mock.patch.stopall])

  def StartPatcher(self, patcher):
    """Call start() on the patcher, and stop() in tearDown."""
    m = patcher.start()
    self._patchers.append(patcher)
    return m

  def PatchObject(self, *args, **kwargs):
    """Create and start a mock.patch.object().

    stop() will be called automatically during tearDown.
    """
    return self.StartPatcher(mock.patch.object(*args, **kwargs))

  def PatchDict(self, *args, **kwargs):
    """Create and start a mock.patch.dict().

    stop() will be called automatically during tearDown.
    """
    return self.StartPatcher(mock.patch.dict(*args, **kwargs))


# MockTestCase must be before TempDirTestCase in this inheritance order,
# because MockTestCase.StartPatcher() calls may be for PartialMocks, which
# create their own temporary directory.  The teardown for those directories
# occurs during MockTestCase.tearDown(), which needs to be run before
# TempDirTestCase.tearDown().
class MockTempDirTestCase(MockTestCase, TempDirTestCase):
  """Convenience class mixing TempDir and Mock."""


class MockOutputTestCase(MockTestCase, OutputTestCase):
  """Convenience class mixing Output and Mock."""


class ProgressBarTestCase(MockOutputTestCase):
  """Test class to test the progress bar."""

  # pylint: disable=protected-access

  def setUp(self):
    self._terminal_size = self.PatchObject(
        operation.ProgressBarOperation, '_GetTerminalSize',
        return_value=operation._TerminalSize(100, 20))
    self.PatchObject(os, 'isatty', return_value=True)

  def SetMockTerminalSize(self, width, height):
    """Set mock terminal's size."""
    self._terminal_size.return_value = operation._TerminalSize(width, height)

  def AssertProgressBarAllEvents(self, num_events):
    """Check that the progress bar generates expected events."""
    skipped = 0
    for i in range(num_events):
      try:
        self.AssertOutputContainsLine('%d%%' % (i * 100 // num_events))
      except AssertionError:
        skipped += 1

    # crbug.com/560953 It's normal to skip a few events under heavy CPU load.
    self.assertLessEqual(skipped, num_events // 2,
                         'Skipped %s of %s progress updates' %
                         (skipped, num_events))

    self.AssertOutputContainsLine('100%')

class MockLoggingTestCase(MockTestCase, LoggingTestCase):
  """Convenience class mixing Logging and Mock."""


@contextlib.contextmanager
def SetTimeZone(tz):
  """Temporarily set the timezone to the specified value.

  This is needed because cros_test_lib.TestCase doesn't call time.tzset()
  after resetting the environment.
  """
  old_environ = os.environ.copy()
  try:
    os.environ['TZ'] = tz
    time.tzset()
    yield
  finally:
    osutils.SetEnvironment(old_environ)
    time.tzset()


class ListTestSuite(unittest.BaseTestSuite):
  """Stub test suite to list all possible tests"""

  # We hack in |top| for local recursive usage.
  # pylint: disable=arguments-differ
  def run(self, result, _debug=False, top=True):
    """List all the tests this suite would have run."""
    # Recursively build a list of all the tests and the descriptions.
    # We do this so we can align the output when printing.
    tests = []
    # Walk all the tests that this suite itself holds.
    for test in self:
      if isinstance(test, type(self)):
        tests += test(result, top=False)
      else:
        desc = test.shortDescription()
        if desc is None:
          desc = ''
        tests.append((test.id(), desc))

    if top:
      if tests:
        # Now that we have all the tests, print them in lined up columns.
        maxlen = max(len(x[0]) for x in tests)
        for test, desc in tests:
          print('%-*s  %s' % (maxlen, test, desc))
      return result
    else:
      return tests


class ListTestLoader(unittest.TestLoader):
  """Stub test loader to list all possible tests"""

  suiteClass = ListTestSuite


class ListTestRunner(object):
  """Stub test runner to list all possible tests"""

  def run(self, test):
    result = unittest.TestResult()
    test(result)
    return result


class TraceTestRunner(unittest.TextTestRunner):
  """Test runner that traces the test code as it runs

  We insert tracing at the test runner level rather than test suite or test
  case because both of those can execute code we've written (e.g. setUpClass
  and setUp), and we want to trace that code too.
  """

  TRACE_KWARGS = {}

  def run(self, test):
    import trace
    tracer = trace.Trace(**self.TRACE_KWARGS)
    return tracer.runfunc(unittest.TextTestRunner.run, self, test)


class ProfileTestRunner(unittest.TextTestRunner):
  """Test runner that profiles the test code as it runs

  We insert profiling at the test runner level rather than test suite or test
  case because both of those can execute code we've written (e.g. setUpClass
  and setUp), and we want to profile that code too.  It might be unexpectedly
  heavy by invoking expensive setup logic.
  """

  PROFILE_KWARGS = {}
  SORT_STATS_KEYS = ()

  def run(self, test):
    import cProfile
    profiler = cProfile.Profile(**self.PROFILE_KWARGS)
    ret = profiler.runcall(unittest.TextTestRunner.run, self, test)

    import pstats
    stats = pstats.Stats(profiler, stream=sys.stderr)
    stats.strip_dirs().sort_stats(*self.SORT_STATS_KEYS).print_stats()
    return ret


class TestProgram(unittest.TestProgram):
  """Helper wrapper around unittest.TestProgram

  Any passed in kwargs are passed directly down to unittest.main; via this, you
  can inject custom argv for example (to limit what tests run).
  """

  def __init__(self, **kwargs):
    self.default_log_level = kwargs.pop('level', 'critical')
    self._leaked_tempdir = None

    try:
      super(TestProgram, self).__init__(**kwargs)
    finally:
      if GlobalTestConfig.NETWORK_TESTS_SKIPPED:
        print('Note: %i network test(s) skipped; use --network to run them.' %
              GlobalTestConfig.NETWORK_TESTS_SKIPPED)

  def parseArgs(self, argv):
    """Parse the command line for the test"""
    description = """Examples:
  %(prog)s                            - run default set of tests
  %(prog)s MyTestSuite                - run suite MyTestSuite
  %(prog)s MyTestCase.testSomething   - run MyTestCase.testSomething
  %(prog)s MyTestCase                 - run all MyTestCase.test* methods
"""
    parser = commandline.ArgumentParser(
        description=description, default_log_level=self.default_log_level)

    # These are options the standard unittest.TestProgram supports.
    parser.add_argument('-q', '--quiet', default=False, action='store_true',
                        help='Minimal output')
    parser.add_argument('-f', '--failfast', default=False, action='store_true',
                        help='Stop on first failure')
    parser.add_argument('tests', nargs='*',
                        help='specific test classes or methods to run')
    parser.add_argument('-c', '--catch', default=False, action='store_true',
                        help='Catch control-C and display results')
    parser.add_argument('-b', '--buffer', default=False, action='store_true',
                        help='Buffer stdout and stderr during test runs')

    # These are custom options we added.
    parser.add_argument('-l', '--list', default=False, action='store_true',
                        help='List all the available tests')
    parser.add_argument('--network', default=False, action='store_true',
                        help='Run tests that depend on good network '
                             'connectivity')
    parser.add_argument('--no-wipe', default=True, action='store_false',
                        dest='wipe',
                        help='Do not wipe the temporary working directory '
                             '(default is to always wipe)')
    parser.add_argument('-u', '--update', default=False, action='store_true',
                        help='Update generated test files as needed.')

    # Note: The tracer module includes coverage options ...
    group = parser.add_argument_group('Tracing options')
    group.add_argument('--trace', default=False, action='store_true',
                       help='Trace test execution')
    group.add_argument('--ignore-module', default='',
                       help='Ignore the specified modules (comma delimited)')
    group.add_argument('--ignore-dir', default='',
                       help='Ignore modules/packages in the specified dirs '
                            '(comma delimited)')
    group.add_argument('--no-ignore-system', default=True, action='store_false',
                       dest='ignore_system',
                       help='Do not ignore sys paths automatically')

    group = parser.add_argument_group('Profiling options')
    group.add_argument('--profile', default=False, action='store_true',
                       help='Profile test execution')
    group.add_argument('--profile-sort-keys', default='time',
                       help='Keys to sort stats by (comma delimited)')
    group.add_argument('--no-profile-builtins', default=True,
                       action='store_false', dest='profile_builtins',
                       help='Do not profile builtin functions')

    opts = parser.parse_args(argv[1:])
    opts.Freeze()

    # Process the common options first.
    if opts.verbose:
      self.verbosity = 2

    if opts.quiet:
      self.verbosity = 0

    if opts.failfast:
      self.failfast = True

    if opts.catch:
      self.catchbreak = True

    if opts.buffer:
      self.buffer = True

    # Then handle the chromite extensions.
    if opts.network:
      GlobalTestConfig.RUN_NETWORK_TESTS = True

    if opts.update:
      GlobalTestConfig.UPDATE_GENERATED_FILES = True

    # We allow --list because it's nice to be able to throw --list onto an
    # existing command line to quickly get the output.  It's clear to users
    # that it does nothing else.
    if sum((opts.trace, opts.profile)) > 1:
      parser.error('--trace/--profile are exclusive')

    if opts.list:
      self.testRunner = ListTestRunner
      self.testLoader = ListTestLoader()
    elif opts.trace:
      self.testRunner = TraceTestRunner

      # Create the automatic ignore list based on sys.path.  We need to filter
      # out chromite paths though as we might have automatic local paths in it.
      auto_ignore = set()
      if opts.ignore_system:
        auto_ignore.add(os.path.join(constants.CHROMITE_DIR, 'third_party'))
        for path in sys.path:
          path = os.path.realpath(path)
          if path.startswith(constants.CHROMITE_DIR):
            continue
          auto_ignore.add(path)

      TraceTestRunner.TRACE_KWARGS = {
          # Disable counting as it only applies to coverage collection.
          'count': False,
          # Enable tracing support since that's what we want w/--trace.
          'trace': True,
          # Enable relative timestamps before each traced line.
          'timing': True,
          'ignoremods': opts.ignore_module.split(','),
          'ignoredirs': set(opts.ignore_dir.split(',')) | auto_ignore,
      }
    elif opts.profile:
      self.testRunner = ProfileTestRunner

      ProfileTestRunner.PROFILE_KWARGS = {
          'subcalls': True,
          'builtins': opts.profile_builtins,
      }

      ProfileTestRunner.SORT_STATS_KEYS = opts.profile_sort_keys.split(',')

    # Figure out which tests the user/unittest wants to run.
    if not opts.tests and self.defaultTest is None:
      self.testNames = None
    elif opts.tests:
      self.testNames = opts.tests
    else:
      self.testNames = (self.defaultTest,)

    if not opts.wipe:
      # Instruct the TempDirTestCase to skip cleanup before actually creating
      # any tempdirs.
      self._leaked_tempdir = TempDirTestCase.SkipCleanup()

    self.createTests()

  def runTests(self):
    # If cidb has been imported, stub it out.  We do this dynamically so we
    # don't have to import cidb in every single test module.
    if 'chromite.lib.cidb' in sys.modules:
      # Unit tests should never connect to the live prod or debug instances
      # of the cidb. This call ensures that they will not accidentally
      # do so through the normal cidb SetUp / GetConnectionForBuilder factory.
      sys.modules['chromite.lib.cidb'].CIDBConnectionFactory.SetupMockCidb()

    try:
      super(TestProgram, self).runTests()
    finally:
      if self._leaked_tempdir is not None:
        logging.info('Working directory %s left behind. Please cleanup later.',
                     self._leaked_tempdir)


class PopenMock(partial_mock.PartialCmdMock):
  """Provides a context where all _Popen instances are low-level mocked."""

  TARGET = 'chromite.lib.cros_build_lib._Popen'
  ATTRS = ('__init__',)
  DEFAULT_ATTR = '__init__'

  def __init__(self):
    partial_mock.PartialCmdMock.__init__(self, create_tempdir=True)

  def _target__init__(self, inst, cmd, *args, **kwargs):
    result = self._results['__init__'].LookupResult(
        (cmd,), hook_args=(inst, cmd,) + args, hook_kwargs=kwargs)

    script = os.path.join(self.tempdir, 'mock_cmd.sh')
    stdout = os.path.join(self.tempdir, 'output')
    stderr = os.path.join(self.tempdir, 'error')
    # This encoding handling might appear a bit wonky, but it's OK, I promise.
    # The purpose of this mock is to stuff data into files so that we can run a
    # fake script in place of the real command.  So any cros_build_lib.run()
    # settings will still be fully checked including encoding.  This code just
    # takes care of writing the data from AddCmdResult objects.  Those might be
    # specified in strings or in bytes, but there's no value in forcing all code
    # to use the same encoding with the mocks.
    def _MaybeEncode(src):
      return src.encode('utf-8') if isinstance(src, six.text_type) else src
    osutils.WriteFile(stdout, _MaybeEncode(result.output), mode='wb')
    osutils.WriteFile(stderr, _MaybeEncode(result.error), mode='wb')
    osutils.WriteFile(
        script,
        ['#!/bin/bash\n', 'cat %s\n' % stdout, 'cat %s >&2\n' % stderr,
         'exit %s' % result.returncode])
    os.chmod(script, 0o700)
    kwargs['cwd'] = self.tempdir
    self.backup['__init__'](inst, [script, '--'] + cmd, *args, **kwargs)


class RunCommandMock(partial_mock.PartialCmdMock):
  """Provides a context where all run invocations low-level mocked."""

  TARGET = 'chromite.lib.cros_build_lib'
  ATTRS = ('run',)
  DEFAULT_ATTR = 'run'

  def run(self, cmd, *args, **kwargs):
    result = self._results['run'].LookupResult(
        (cmd,), kwargs=kwargs, hook_args=(cmd,) + args, hook_kwargs=kwargs)

    popen_mock = PopenMock()
    popen_mock.AddCmdResult(partial_mock.Ignore(), result.returncode,
                            result.output, result.error)
    with popen_mock:
      return self.backup['run'](cmd, *args, **kwargs)

  # Backwards compat API.
  RunCommand = run


class RunCommandTestCase(MockTestCase):
  """MockTestCase that mocks out run by default."""

  def setUp(self):
    self.rc = self.StartPatcher(RunCommandMock())
    self.rc.SetDefaultCmdResult()
    self.assertCommandCalled = self.rc.assertCommandCalled
    self.assertCommandContains = self.rc.assertCommandContains

    # These ENV variables affect run behavior, hide them.
    self._old_envs = {e: os.environ.pop(e) for e in constants.ENV_PASSTHRU
                      if e in os.environ}

  def tearDown(self):
    # Restore hidden ENVs.
    if hasattr(self, '_old_envs'):
      os.environ.update(self._old_envs)


class RunCommandTempDirTestCase(RunCommandTestCase, TempDirTestCase):
  """Convenience class mixing TempDirTestCase and RunCommandTestCase"""


class main(TestProgram):
  """Chromite's version of unittest.main.  Invoke this, not unittest.main."""