# Copyright (c) 2012 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.

"""Unittests for the osutils.py module (imagine that!)."""

import collections
import filecmp
import glob
import grp
import os
import pwd
import re
import stat
import sys
import unittest
from unittest import mock

from chromite.lib import cros_build_lib
from chromite.lib import cros_test_lib
from chromite.lib import osutils
from chromite.lib import partial_mock


if sys.version_info.major >= 3:
  from pathlib import Path


class TestOsutils(cros_test_lib.TempDirTestCase):
  """General unittests for the osutils module."""

  def testIsSubPath(self):
    self.assertTrue(osutils.IsSubPath('/a', '/a'))
    self.assertTrue(osutils.IsSubPath('/a', '/a/'))

    self.assertTrue(osutils.IsSubPath('/a/b', '/a'))
    self.assertTrue(osutils.IsSubPath('/a/b', '/a/'))

    self.assertTrue(osutils.IsSubPath('/a/b/e', '/a/b/c/../../b'))

    self.assertFalse(osutils.IsSubPath('/ab', '/a/b'))
    self.assertFalse(osutils.IsSubPath('/a/bcde', '/a/b'))

  def testAllocateFile(self):
    """Verify we can allocate a file of a certain length."""
    filename = os.path.join(self.tempdir, 'foo')
    size = 1234
    osutils.AllocateFile(filename, size)

    self.assertExists(filename)
    self.assertEqual(size, os.path.getsize(filename))

  def testReadWriteFile(self):
    """Verify we can write data to a file, and then read it back."""
    filename = os.path.join(self.tempdir, 'foo')
    data = 'alsdkfjasldkfjaskdlfjasdf'
    self.assertEqual(osutils.WriteFile(filename, data), None)
    self.assertEqual(osutils.ReadFile(filename), data)

  def testReadWritePath(self):
    """Verify we can write data to a Path, and then read it back."""
    filename = Path(self.tempdir) / 'foo'
    data = 'alsdkfjasldkfjaskdlfjasdf'
    self.assertEqual(osutils.WriteFile(filename, data), None)
    self.assertEqual(osutils.ReadFile(filename), data)

  def testReadBinary(self):
    """Verify we can read data as binary."""
    filename = os.path.join(self.tempdir, 'foo')
    data = b'alsdkfjasldkfjaskdlfjasdf'
    self.assertEqual(osutils.WriteFile(filename, data, mode='wb'), None)
    self.assertEqual(osutils.ReadFile(filename, mode='rb'), data)

  def testWriteFileStringIter(self):
    """Verify that we can write an iterable of strings."""
    filename = os.path.join(self.tempdir, 'foo')
    data = ['a', 'cd', 'ef']
    self.assertEqual(osutils.WriteFile(filename, data), None)
    self.assertEqual(osutils.ReadFile(filename), ''.join(data))

  def testWriteFileBytesIter(self):
    """Verify that we can write an iterable of bytes."""
    filename = os.path.join(self.tempdir, 'foo')
    data = [b'ab', b'cd', b'ef']
    self.assertEqual(osutils.WriteFile(filename, data, mode='wb'), None)
    self.assertEqual(osutils.ReadFile(filename, mode='rb'), b''.join(data))

  def testSudoWrite(self):
    """Verify that we can write a file as sudo."""
    with osutils.TempDir(sudo_rm=True) as tempdir:
      root_owned_dir = os.path.join(tempdir, 'foo')
      self.assertTrue(osutils.SafeMakedirs(root_owned_dir, sudo=True))
      for atomic in (True, False):
        filename = os.path.join(root_owned_dir,
                                'bar.atomic' if atomic else 'bar')
        self.assertRaises(IOError, osutils.WriteFile, filename, 'data')

        osutils.WriteFile(filename, 'test', atomic=atomic, sudo=True)
        self.assertEqual('test', osutils.ReadFile(filename))
        self.assertEqual(0, os.stat(filename).st_uid)

  def testSudoWriteAppendNew(self):
    """Verify that we can write a new file as sudo when appending."""
    with osutils.TempDir(sudo_rm=True) as tempdir:
      path = os.path.join(tempdir, 'foo')
      osutils.WriteFile(path, 'two', mode='a', sudo=True)
      self.assertEqual('two', osutils.ReadFile(path))

  def testSudoWriteAppendExisting(self):
    """Verify that we can write a file as sudo when appending."""
    with osutils.TempDir(sudo_rm=True) as tempdir:
      path = os.path.join(tempdir, 'foo')
      osutils.WriteFile(path, 'one', sudo=True)
      self.assertRaises(IOError, osutils.WriteFile, path, 'data')
      osutils.WriteFile(path, 'two', mode='a', sudo=True)
      self.assertEqual('onetwo', osutils.ReadFile(path))

  def testSudoReadNoTrunc(self):
    """Verify that we can write a new file as sudo when r+."""
    def testit(path, sudo):
      osutils.WriteFile(path, 'two')
      self.assertEqual('two', osutils.ReadFile(path))
      osutils.WriteFile(path, 'X', mode='r+', sudo=sudo)
      self.assertEqual('Xwo', osutils.ReadFile(path))

    # Make sure that r+ works as we think it does.
    path = os.path.join(self.tempdir, 'foo')
    testit(path, False)

    # Now run it again with sudo.
    with osutils.TempDir(sudo_rm=True) as tempdir:
      path = os.path.join(tempdir, 'foo')
      testit(path, True)

  def testReadFileNonExistent(self):
    """Verify what happens if you ReadFile a file that isn't there."""
    filename = os.path.join(self.tempdir, 'bogus')
    with self.assertRaises(IOError):
      osutils.ReadFile(filename)

  def testWriteChmod(self):
    """Verify writing files with perms works."""
    def getmode(path):
      return os.stat(path).st_mode & 0o7777
    def assertMode(path, mode):
      self.assertEqual(getmode(path), mode)

    path = os.path.join(self.tempdir, 'file')
    osutils.WriteFile(path, 'asdf')
    assertMode(path, 0o644)

    osutils.WriteFile(path, 'asdf', chmod=0o666)
    assertMode(path, 0o666)

    osutils.WriteFile(path, 'asdf', atomic=True, chmod=0o664)
    assertMode(path, 0o664)

    osutils.WriteFile(path, 'asdf', sudo=True, chmod=0o755)
    assertMode(path, 0o755)

    osutils.WriteFile(path, 'asdf', sudo=True, atomic=True, chmod=0o775)
    assertMode(path, 0o775)

  def testSafeSymlink(self):
    """Test that we can create symlinks."""
    with osutils.TempDir(sudo_rm=True) as tempdir:
      file_a = os.path.join(tempdir, 'a')
      osutils.WriteFile(file_a, 'a')

      file_b = os.path.join(tempdir, 'b')
      osutils.WriteFile(file_b, 'b')

      user_dir = os.path.join(tempdir, 'bar')
      user_link = os.path.join(user_dir, 'link')
      osutils.SafeMakedirs(user_dir)

      root_dir = os.path.join(tempdir, 'foo')
      root_link = os.path.join(root_dir, 'link')
      osutils.SafeMakedirs(root_dir, sudo=True)

      # We can create and override links owned by a non-root user.
      osutils.SafeSymlink(file_a, user_link)
      self.assertEqual('a', osutils.ReadFile(user_link))

      osutils.SafeSymlink(file_b, user_link)
      self.assertEqual('b', osutils.ReadFile(user_link))

      # Handle Path objects.
      osutils.SafeSymlink(Path(file_a), Path(user_link))
      self.assertEqual('a', osutils.ReadFile(user_link))

      osutils.SafeSymlink(Path(file_b), Path(user_link))
      self.assertEqual('b', osutils.ReadFile(user_link))

      # We can create and override links owned by root.
      osutils.SafeSymlink(file_a, root_link, sudo=True)
      self.assertEqual('a', osutils.ReadFile(root_link))

      osutils.SafeSymlink(file_b, root_link, sudo=True)
      self.assertEqual('b', osutils.ReadFile(root_link))

      # Handle Path objects.
      osutils.SafeSymlink(Path(file_a), Path(root_link), sudo=True)
      self.assertEqual('a', osutils.ReadFile(root_link))

      osutils.SafeSymlink(Path(file_b), Path(root_link), sudo=True)
      self.assertEqual('b', osutils.ReadFile(root_link))

  def testSafeUnlink(self):
    """Test unlinking files work (existing or not)."""
    def f(sudo=False, as_path=False):
      with osutils.TempDir(sudo_rm=sudo) as dirname:
        path = os.path.join(dirname, 'foon')
        if as_path:
          path = Path(path)
        osutils.Touch(path, makedirs=True)
        self.assertExists(path)
        if sudo:
          osutils.Chown(dirname, user='root', group='root', recursive=False)
          self.assertRaises(EnvironmentError, os.unlink, path)
        self.assertTrue(osutils.SafeUnlink(path, sudo=sudo))
        self.assertNotExists(path)
        self.assertFalse(osutils.SafeUnlink(path))
        self.assertNotExists(path)

    f(False)
    f(True)
    f(False, True)
    f(True, True)

  def testSafeUnlinkSudoInaccessible(self):
    """Test unlinking files work in a dir only root can read."""
    with osutils.TempDir(sudo_rm=True) as dirname:
      path = os.path.join(dirname, 'exists')
      osutils.Touch(path, mode=0o000)
      os.chmod(dirname, 0o000)
      self.assertRaises(EnvironmentError, os.unlink, path)
      self.assertTrue(osutils.SafeUnlink(path, sudo=True))
      self.assertFalse(osutils.SafeUnlink(path, sudo=True))

      os.chmod(dirname, 0o700)
      self.assertNotExists(path)

  def testSafeMakedirs(self):
    """Test creating directory trees work (existing or not)."""
    path = os.path.join(self.tempdir, 'a', 'b', 'c', 'd', 'e')
    self.assertTrue(osutils.SafeMakedirs(path))
    self.assertExists(path)
    self.assertFalse(osutils.SafeMakedirs(path))
    self.assertExists(path)

  @unittest.skipIf(sys.version_info.major < 3, 'Requires pathlib from py3')
  def testSafeMakedirsWithPathObject(self):
    """Test creating directory trees work (existing or not) on |Path|s."""
    path = Path(self.tempdir) / 'a' / 'b' / 'c' / 'd' / 'e'
    self.assertTrue(osutils.SafeMakedirs(path))
    self.assertExists(path)
    self.assertFalse(osutils.SafeMakedirs(path))
    self.assertExists(path)

  def testSafeMakedirsMode(self):
    """Test that mode is honored."""
    path = os.path.join(self.tempdir, 'a', 'b', 'c', 'd', 'e')
    self.assertTrue(osutils.SafeMakedirs(path, mode=0o775))
    self.assertEqual(0o775, stat.S_IMODE(os.stat(path).st_mode))
    self.assertFalse(osutils.SafeMakedirs(path, mode=0o777))
    self.assertEqual(0o777, stat.S_IMODE(os.stat(path).st_mode))
    cros_build_lib.sudo_run(['chown', 'root:root', path], print_cmd=False)
    # Tries, but fails to change the mode.
    self.assertFalse(osutils.SafeMakedirs(path, 0o755))
    self.assertEqual(0o777, stat.S_IMODE(os.stat(path).st_mode))

  def testSafeMakedirs_error(self):
    """Check error paths."""
    with self.assertRaises(OSError):
      osutils.SafeMakedirs('/foo/bar/cow/moo/wee')
      ret = cros_build_lib.run(['ls', '-Rla', '/foo'], check=False,
                               capture_output=True)
      print('ls output of /foo:\n{{{%s}}}' % (ret.stdout,), file=sys.stderr)
    self.assertRaises(OSError, osutils.SafeMakedirs, '')

  def testSafeMakedirsSudo(self):
    """Test creating directory trees work as root (existing or not)."""
    self.ExpectRootOwnedFiles()
    path = os.path.join(self.tempdir, 'a', 'b', 'c', 'd', 'e')
    self.assertTrue(osutils.SafeMakedirs(path, sudo=True))
    self.assertExists(path)
    self.assertFalse(osutils.SafeMakedirs(path, sudo=True))
    self.assertExists(path)
    self.assertEqual(os.stat(path).st_uid, 0)

  def testSafeMakedirsNoSudoRootOwnedDirs(self):
    """Test that we can recover some root owned directories."""
    self.ExpectRootOwnedFiles()
    root_owned_prefix = os.path.join(self.tempdir, 'root_owned_prefix')
    root_owned_dir = os.path.join(root_owned_prefix, 'root_owned_dir')
    non_root_dir = os.path.join(root_owned_prefix, 'non_root_dir')
    self.assertTrue(osutils.SafeMakedirs(root_owned_dir, sudo=True))
    self.assertExists(root_owned_prefix)
    self.assertEqual(os.stat(root_owned_prefix).st_uid, 0)
    self.assertExists(root_owned_dir)
    self.assertEqual(os.stat(root_owned_dir).st_uid, 0)

    # Test that we can reclaim a root-owned dir.
    # Note, return value is False because the directory already exists.
    self.assertFalse(osutils.SafeMakedirsNonRoot(root_owned_dir))
    self.assertNotEqual(os.stat(root_owned_dir).st_uid, 0)

    # Test that we can create a non-root directory in a root-path.
    self.assertTrue(osutils.SafeMakedirsNonRoot(non_root_dir))
    self.assertNotEqual(os.stat(non_root_dir).st_uid, 0)

  def testRmDir(self):
    """Test that removing dirs work."""
    main_path = os.path.join(self.tempdir, 'a', 'b', 'c', 'd', 'e')
    paths_to_test = [main_path]
    if sys.version_info.major >= 3:
      paths_to_test.append(Path(main_path))

    for path in paths_to_test:
      self.assertRaises(EnvironmentError, osutils.RmDir, path)
      osutils.SafeMakedirs(path)
      osutils.RmDir(path)
      osutils.RmDir(path, ignore_missing=True)
      self.assertRaises(EnvironmentError, osutils.RmDir, path)

      osutils.SafeMakedirs(path)
      osutils.RmDir(path)
      self.assertNotExists(path)

  def testRmDirSudo(self):
    """Test that removing dirs via sudo works."""
    subpath = os.path.join(self.tempdir, 'a')
    main_path = os.path.join(subpath, 'b', 'c', 'd', 'e')
    paths_to_test = [main_path]
    if sys.version_info.major >= 3:
      paths_to_test.append(Path(main_path))

    for path in paths_to_test:
      self.assertTrue(osutils.SafeMakedirs(path, sudo=True))
      self.assertRaises(OSError, osutils.RmDir, path)
      osutils.RmDir(subpath, sudo=True)
      self.assertRaises(
          cros_build_lib.RunCommandError, osutils.RmDir, subpath, sudo=True)

  def testTouchFile(self):
    """Test that we can touch files."""
    path = os.path.join(self.tempdir, 'touchit')
    self.assertNotExists(path)
    osutils.Touch(path)
    self.assertExists(path)
    self.assertEqual(os.path.getsize(path), 0)

  def testTouchFileSubDir(self):
    """Test that we can touch files in non-existent subdirs."""
    path = os.path.join(self.tempdir, 'a', 'b', 'c', 'touchit')
    self.assertNotExists(os.path.dirname(path))
    osutils.Touch(path, makedirs=True)
    self.assertExists(path)
    self.assertEqual(os.path.getsize(path), 0)

  def testChmod(self):
    """Test Chmod."""
    def getmode(path):
      return os.stat(path).st_mode & 0o7777

    path = os.path.join(self.tempdir, 'chmodtests')
    osutils.Touch(path)

    osutils.Chmod(path, 0o700)
    self.assertEqual(getmode(path), 0o700)

    osutils.Chmod(Path(path), 0o644)
    self.assertEqual(getmode(path), 0o644)

    osutils.Chown(path, user='root', group='root')
    osutils.Chmod(path, 0o660, sudo=True)
    self.assertEqual(getmode(path), 0o660)
    osutils.Chmod(Path(path), 0o661, sudo=True)
    self.assertEqual(getmode(path), 0o661)

    self.assertRaises(OSError, osutils.Chmod, path, 0o600)

  def testChown(self):
    """Test chown."""
    # Helpers to get the user and group name of the given path's owner.
    def User(path):
      return pwd.getpwuid(os.stat(path).st_uid).pw_name
    def Group(path):
      return grp.getgrgid(os.stat(path).st_gid).gr_name

    filename = os.path.join(self.tempdir, 'chowntests')
    osutils.Touch(filename)

    user = User(filename)
    group = Group(filename)

    new_user = new_group = 'root'

    # Change only the user.
    osutils.Chown(filename, user=new_user)
    self.assertEqual(new_user, User(filename))
    self.assertEqual(new_group, Group(filename))

    # Change both user and group.
    osutils.Chown(filename, user=user, group=group)
    self.assertEqual(user, User(filename))
    self.assertEqual(group, Group(filename))

    # Change only the group.
    osutils.Chown(filename, group=new_group)
    self.assertEqual(user, User(filename))
    self.assertEqual(new_group, Group(filename))

    # Chown with no arguments.
    osutils.Chown(filename, user=new_user, group=new_group)
    self.assertEqual(new_user, User(filename))
    self.assertEqual(new_group, Group(filename))
    osutils.Chown(filename)
    self.assertEqual(user, User(filename))
    self.assertEqual(group, Group(filename))

    # With Path object.
    osutils.Chown(Path(filename), user=new_user, group=new_group)
    self.assertEqual(new_user, User(filename))
    self.assertEqual(new_group, Group(filename))
    osutils.Chown(Path(filename))
    self.assertEqual(user, User(filename))
    self.assertEqual(group, Group(filename))

    # User and group ids as the arguments.
    osutils.Chown(filename, user=0, group=0)
    self.assertEqual(new_user, User(filename))
    self.assertEqual(new_group, Group(filename))

    # Recursive.
    dirname = os.path.join(self.tempdir, 'chowntestsdir')
    osutils.SafeMakedirs(dirname)
    filename = os.path.join(dirname, 'chowntestsfile')
    osutils.Touch(filename)
    # Chown without recursive.
    osutils.Chown(dirname, user=new_user, group=new_group)
    self.assertEqual(new_user, User(dirname))
    self.assertEqual(new_group, Group(dirname))
    self.assertEqual(user, User(filename))
    self.assertEqual(group, Group(filename))
    # Chown with recursive.
    osutils.Chown(dirname, user=new_user, group=new_group, recursive=True)
    self.assertEqual(new_user, User(filename))
    self.assertEqual(new_group, Group(filename))
    osutils.Chown(dirname, user=user, group=group, recursive=True)
    self.assertEqual(user, User(dirname))
    self.assertEqual(group, Group(dirname))
    self.assertEqual(user, User(filename))
    self.assertEqual(group, Group(filename))


class TestEmptyDir(cros_test_lib.TempDirTestCase):
  """Test osutils.EmptyDir."""

  def setUp(self):
    self.subdir = os.path.join(self.tempdir, 'a')
    self.nestedfile = os.path.join(self.subdir, 'b', 'c', 'd', 'e')
    self.topfile = os.path.join(self.tempdir, 'file')

  def testEmptyDir(self):
    """Empty an empty directory."""
    osutils.EmptyDir(self.tempdir)
    osutils.EmptyDir(self.tempdir, ignore_missing=True, sudo=True)

  def testNonExistentDir(self):
    """Non-existent directory."""
    # Ignore_missing=False
    with self.assertRaises(osutils.EmptyDirNonExistentException):
      osutils.EmptyDir(self.subdir)

    # Ignore missing=True
    osutils.EmptyDir(self.subdir, ignore_missing=True)

  def testEmptyWithContentsMinFlags(self):
    """Test ability to empty actual directory contents."""
    osutils.Touch(self.nestedfile, makedirs=True)
    osutils.Touch(self.topfile, makedirs=True)

    osutils.EmptyDir(self.tempdir)

    self.assertExists(self.tempdir)
    self.assertNotExists(self.subdir)
    self.assertNotExists(self.topfile)

  def testEmptyWithContentsMaxFlags(self):
    """Test ability to empty actual directory contents."""
    osutils.Touch(self.nestedfile, makedirs=True)
    osutils.Touch(self.topfile, makedirs=True)

    osutils.EmptyDir(self.tempdir, ignore_missing=True, sudo=True)

    self.assertExists(self.tempdir)
    self.assertNotExists(self.subdir)
    self.assertNotExists(self.topfile)

  def testEmptyWithRootOwnedContents(self):
    """Test handling of root owned sub directories."""
    # Root owned contents.
    osutils.SafeMakedirs(self.nestedfile, sudo=True)

    # Fails without sudo=True
    with self.assertRaises(OSError):
      osutils.EmptyDir(self.tempdir)
    self.assertExists(self.nestedfile)

    # Works with sudo=True
    osutils.EmptyDir(self.tempdir, sudo=True)
    self.assertExists(self.tempdir)
    self.assertNotExists(self.subdir)


  def testExclude(self):
    """Test ability to empty actual directory contents.

    Also ensure that the excludes argument can really be just an iterable.
    """
    files = {
        'keep': True,
        'keepdir/foo': True,
        'keepdir/bar': True,
        'remove': False,
        'removedir/foo': False,
        'removedir/bar': False,
    }

    excludes = ['keep', 'keepdir', 'bogus']

    # Perform exclusion of non-existent files.
    osutils.EmptyDir(self.tempdir, exclude=iter(excludes))

    # Create files.
    for f in files.keys():
      osutils.Touch(os.path.join(self.tempdir, f), makedirs=True)

    # Empty with excludes.
    osutils.EmptyDir(self.tempdir, exclude=iter(excludes))

    # Verify that the results are what we expect.
    for f, expected in files.items():
      f = os.path.join(self.tempdir, f)
      self.assertEqual(os.path.exists(f), expected, 'Unexpected: %s' % f)
    self.assertExists(os.path.join(self.tempdir, 'keepdir'))
    self.assertNotExists(os.path.join(self.tempdir, 'removedir'))


class TestProcess(cros_test_lib.RunCommandTestCase):
  """Tests for osutils.IsChildProcess."""

  def testIsChildProcess(self):
    """Test IsChildProcess with no name."""
    mock_pstree_output = 'a(1)-+-b(2)\n\t|-c(3)\n\t|-foo(4)-bar(5)'
    self.rc.AddCmdResult(partial_mock.Ignore(), output=mock_pstree_output)
    self.assertTrue(osutils.IsChildProcess(4))
    self.assertTrue(osutils.IsChildProcess(4, name='foo'))
    self.assertFalse(osutils.IsChildProcess(5, name='foo'))


class TempDirTests(cros_test_lib.TestCase):
  """Unittests of osutils.TempDir.

  Unlike other test classes in this file, TempDirTestCase isn't used as a base
  class, because that is the functionality under test.
  """
  PREFIX = 'chromite.test.osutils.TempDirTests'

  class HelperException(Exception):
    """Exception for tests to raise to test exception handling."""

  class HelperExceptionInner(Exception):
    """Exception for tests to raise to test exception handling."""

  def testBasicSuccessEmpty(self):
    """Test we create and cleanup an empty tempdir."""
    with osutils.TempDir(prefix=self.PREFIX) as td:
      tempdir = td
      # Show the temp directory exists and is empty.
      self.assertTrue(os.path.isdir(tempdir))
      self.assertEqual(os.listdir(tempdir), [])

    # Show the temp directory no longer exists.
    self.assertNotExists(tempdir)

  def testBasicSuccessNotEmpty(self):
    """Test we cleanup tempdir with stuff in it."""
    with osutils.TempDir(prefix=self.PREFIX) as td:
      tempdir = td
      # Show the temp directory exists and is empty.
      self.assertTrue(os.path.isdir(tempdir))
      self.assertEqual(os.listdir(tempdir), [])

      # Create an empty file.
      osutils.Touch(os.path.join(tempdir, 'foo.txt'))

      # Create nested sub directories.
      subdir = os.path.join(tempdir, 'foo', 'bar', 'taco')
      os.makedirs(subdir)
      osutils.Touch(os.path.join(subdir, 'sauce.txt'))

    # Show the temp directory no longer exists.
    self.assertNotExists(tempdir)

  def testErrorCleanup(self):
    """Test we cleanup, even if an exception is raised."""
    try:
      with osutils.TempDir(prefix=self.PREFIX) as td:
        tempdir = td
        raise TempDirTests.HelperException()
    except TempDirTests.HelperException:
      pass

    # Show the temp directory no longer exists.
    self.assertNotExists(tempdir)

  def testCleanupExceptionContextException(self):
    """Test an exception during cleanup if the context DID raise."""
    was_raised = False
    tempdir_obj = osutils.TempDir(prefix=self.PREFIX)

    with mock.patch.object(osutils, '_TempDirTearDown',
                           side_effect=TempDirTests.HelperException):
      try:
        with tempdir_obj as td:
          tempdir = td
          raise TempDirTests.HelperExceptionInner()
      except TempDirTests.HelperExceptionInner:
        was_raised = True

    # Show that the exception exited the context.
    self.assertTrue(was_raised)

    # Verify the tempdir object no longer contains a reference to the tempdir.
    self.assertIsNone(tempdir_obj.tempdir)

    # Cleanup the dir leaked by our mock exception.
    os.rmdir(tempdir)

  def testCleanupExceptionNoContextException(self):
    """Test an exception during cleanup if the context did NOT raise."""
    was_raised = False
    tempdir_obj = osutils.TempDir(prefix=self.PREFIX)

    with mock.patch.object(osutils, '_TempDirTearDown',
                           side_effect=TempDirTests.HelperException):
      try:
        with tempdir_obj as td:
          tempdir = td
      except TempDirTests.HelperException:
        was_raised = True

    # Show that the exception exited the context.
    self.assertTrue(was_raised)

    # Verify the tempdir object no longer contains a reference to the tempdir.
    self.assertIsNone(tempdir_obj.tempdir)

    # Cleanup the dir leaked by our mock exception.
    os.rmdir(tempdir)

  def testSkipCleanup(self):
    """Test that we leave behind tempdirs when requested."""
    tempdir_obj = osutils.TempDir(prefix=self.PREFIX, delete=False)
    tempdir = tempdir_obj.tempdir
    tempdir_obj.Cleanup()
    # Ensure we cleaned up ...
    self.assertIsNone(tempdir_obj.tempdir)
    # ... but leaked the directory.
    self.assertExists(tempdir)
    # Now really cleanup the directory leaked by the test.
    os.rmdir(tempdir)

  def testSkipCleanupGlobal(self):
    """Test that we reset global tempdir as expected even with skip."""
    with osutils.TempDir(prefix=self.PREFIX, set_global=True) as tempdir:
      tempdir_before = osutils.GetGlobalTempDir()
      tempdir_obj = osutils.TempDir(prefix=self.PREFIX, set_global=True,
                                    delete=False)
      tempdir_inside = osutils.GetGlobalTempDir()
      tempdir_obj.Cleanup()
      tempdir_after = osutils.GetGlobalTempDir()

    # We shouldn't leak the outer directory.
    self.assertNotExists(tempdir)
    self.assertEqual(tempdir_before, tempdir_after)
    # This is a strict substring check.
    self.assertLess(tempdir_before, tempdir_inside)


class MountTests(cros_test_lib.TestCase):
  """Unittests for osutils mounting and umounting helpers."""

  def testMountTmpfsDir(self):
    """Verify mounting a tmpfs works"""
    cleaned = False
    with osutils.TempDir(prefix='chromite.test.osutils') as tempdir:
      st_before = os.stat(tempdir)
      try:
        # Mount the dir and verify it worked.
        osutils.MountTmpfsDir(tempdir)
        st_after = os.stat(tempdir)
        self.assertNotEqual(st_before.st_dev, st_after.st_dev)

        # Unmount the dir and verify it worked.
        osutils.UmountDir(tempdir)
        cleaned = True

        # Finally make sure it's cleaned up.
        self.assertNotExists(tempdir)
      finally:
        if not cleaned:
          cros_build_lib.sudo_run(['umount', '-lf', tempdir],
                                  check=False)

  def testUnmountTree(self):
    with osutils.TempDir(prefix='chromite.test.osutils') as tempdir:
      # Mount the dir and verify it worked.
      st_before = os.stat(tempdir)
      osutils.MountTmpfsDir(tempdir)
      st_after = os.stat(tempdir)
      self.assertNotEqual(st_before.st_dev, st_after.st_dev)

      # Mount an inner dir the same way.
      tempdir2 = os.path.join(tempdir, 'inner')
      osutils.SafeMakedirsNonRoot(tempdir2)
      st_before2 = os.stat(tempdir2)
      osutils.MountTmpfsDir(tempdir2)
      st_after2 = os.stat(tempdir2)
      self.assertNotEqual(st_before2.st_dev, st_after2.st_dev)

      # Unmount the whole tree and verify it worked.
      osutils.UmountTree(tempdir)
      st_umount = os.stat(tempdir)
      self.assertNotExists(tempdir2)
      self.assertEqual(st_before.st_dev, st_umount.st_dev)


class IteratePathsTest(cros_test_lib.TestCase):
  """Test iterating through all segments of a path."""

  def testType(self):
    """Check that return value is an iterator."""
    self.assertIsInstance(osutils.IteratePaths('/'), collections.abc.Iterator)

  def testRoot(self):
    """Test iterating from root directory."""
    inp = '/'
    exp = ['/']
    self.assertEqual(list(osutils.IteratePaths(inp)), exp)

  def testOneDir(self):
    """Test iterating from a directory in a root directory."""
    inp = '/abc'
    exp = ['/', '/abc']
    self.assertEqual(list(osutils.IteratePaths(inp)), exp)

  def testTwoDirs(self):
    """Test iterating two dirs down."""
    inp = '/abc/def'
    exp = ['/', '/abc', '/abc/def']
    self.assertEqual(list(osutils.IteratePaths(inp)), exp)

  def testNormalize(self):
    """Test argument being normalized."""
    cases = [
        ('//', ['/']),
        ('///', ['/']),
        ('/abc/', ['/', '/abc']),
        ('/abc//def', ['/', '/abc', '/abc/def']),
    ]
    for inp, exp in cases:
      self.assertEqual(list(osutils.IteratePaths(inp)), exp)


class IteratePathParentsTest(cros_test_lib.TestCase):
  """Test parent directory iteration functionality."""

  def _RunForPath(self, path, expected):
    result_components = []
    for p in osutils.IteratePathParents(path):
      result_components.append(os.path.basename(p))

    result_components.reverse()
    if expected is not None:
      self.assertEqual(expected, result_components)

  def testIt(self):
    """Run the test vectors."""
    vectors = {
        '/': [''],
        '/path/to/nowhere': ['', 'path', 'to', 'nowhere'],
        '/path/./to': ['', 'path', 'to'],
        '//path/to': ['', 'path', 'to'],
        'path/to': None,
        '': None,
    }
    for p, e in vectors.items():
      self._RunForPath(p, e)


class FindInPathParentsTest(cros_test_lib.TempDirTestCase):
  """Test FindInPathParents functionality."""

  D = cros_test_lib.Directory

  DIR_STRUCT = [
      D('a', [
          D('.repo', []),
          D('b', [
              D('c', [])
          ])
      ])
  ]

  START_PATH = os.path.join('a', 'b', 'c')

  def setUp(self):
    cros_test_lib.CreateOnDiskHierarchy(self.tempdir, self.DIR_STRUCT)

  def testFound(self):
    """Target is found."""
    found = osutils.FindInPathParents(
        '.repo', os.path.join(self.tempdir, self.START_PATH))
    self.assertEqual(found, os.path.join(self.tempdir, 'a', '.repo'))

  def testNotFound(self):
    """Target is not found."""
    found = osutils.FindInPathParents(
        'does.not/exist', os.path.join(self.tempdir, self.START_PATH))
    self.assertEqual(found, None)


class SourceEnvironmentTest(cros_test_lib.TempDirTestCase):
  """Test osutil's environmental variable related methods."""

  ENV_WHITELIST = {
      'ENV1': 'monkeys like bananas',
      'ENV3': 'merci',
      'ENV6': '',
  }

  ENV_OTHER = {
      'ENV2': 'bananas are yellow',
      'ENV4': 'de rien',
  }

  ENV = """
declare -x ENV1="monkeys like bananas"
declare -x ENV2="bananas are yellow"
declare -x ENV3="merci"
declare -x ENV4="de rien"
declare -x ENV6=''
declare -x ENVA=('a b c' 'd' 'e 1234 %')
"""

  ENV_MULTILINE = """
declare -x ENVM="gentil
mechant"
"""

  def setUp(self):
    self.env_file = os.path.join(self.tempdir, 'environment')
    self.env_file_multiline = os.path.join(self.tempdir, 'multiline')
    osutils.WriteFile(self.env_file, self.ENV)
    osutils.WriteFile(self.env_file_multiline, self.ENV_MULTILINE)

  def testWhiteList(self):
    env_dict = osutils.SourceEnvironment(
        self.env_file, ('ENV1', 'ENV3', 'ENV5', 'ENV6'))
    self.assertEqual(env_dict, self.ENV_WHITELIST)

  def testArrays(self):
    env_dict = osutils.SourceEnvironment(self.env_file, ('ENVA',))
    self.assertEqual(env_dict, {'ENVA': 'a b c,d,e 1234 %'})

    env_dict = osutils.SourceEnvironment(self.env_file, ('ENVA',), ifs=' ')
    self.assertEqual(env_dict, {'ENVA': 'a b c d e 1234 %'})

    env_dict = osutils.SourceEnvironment(self.env_file_multiline, ('ENVM',),
                                         multiline=True)
    self.assertEqual(env_dict, {'ENVM': 'gentil\nmechant'})


class DeviceInfoTests(cros_test_lib.RunCommandTestCase):
  """Tests methods retrieving information about devices."""

  FULL_OUTPUT = """
NAME="sda" RM="0" TYPE="disk" SIZE="128G"
NAME="sda1" RM="1" TYPE="part" SIZE="100G"
NAME="sda2" RM="1" TYPE="part" SIZE="28G"
NAME="sdc" RM="1" TYPE="disk" SIZE="7.4G"
NAME="sdc1" RM="1" TYPE="part" SIZE="1G"
NAME="sdc2" RM="1" TYPE="part" SIZE="6.4G"
"""

  PARTIAL_OUTPUT = """
NAME="sdc" RM="1" TYPE="disk" SIZE="7.4G"
NAME="sdc1" RM="1" TYPE="part" SIZE="1G"
NAME="sdc2" RM="1" TYPE="part" SIZE="6.4G"
"""

  def testListBlockDevices(self):
    """Tests that we can list all block devices correctly."""
    self.rc.AddCmdResult(partial_mock.Ignore(), output=self.FULL_OUTPUT)
    devices = osutils.ListBlockDevices()
    self.assertEqual(devices[0].NAME, 'sda')
    self.assertEqual(devices[0].RM, '0')
    self.assertEqual(devices[0].TYPE, 'disk')
    self.assertEqual(devices[0].SIZE, '128G')
    self.assertEqual(devices[3].NAME, 'sdc')
    self.assertEqual(devices[3].RM, '1')
    self.assertEqual(devices[3].TYPE, 'disk')
    self.assertEqual(devices[3].SIZE, '7.4G')

  def testGetDeviceSize(self):
    """Tests that we can get the size of a device."""
    self.rc.AddCmdResult(partial_mock.Ignore(), output=self.PARTIAL_OUTPUT)
    self.assertEqual(osutils.GetDeviceSize('/dev/sdc'), '7.4G')


class ChdirTests(cros_test_lib.MockTempDirTestCase):
  """Tests for ChdirContext."""

  def testChdir(self):
    current_dir = os.getcwd()
    self.assertNotEqual(self.tempdir, os.getcwd())
    with osutils.ChdirContext(self.tempdir):
      self.assertEqual(self.tempdir, os.getcwd())
    self.assertEqual(current_dir, os.getcwd())


class MountOverlayTest(cros_test_lib.MockTempDirTestCase):
  """Tests MountOverlayContext."""

  def setUp(self):
    self.upperdir = os.path.join(self.tempdir, 'first_level', 'upperdir')
    self.lowerdir = os.path.join(self.tempdir, 'lowerdir')
    self.mergeddir = os.path.join(self.tempdir, 'mergeddir')

    for path in [self.upperdir, self.lowerdir, self.mergeddir]:
      osutils.Touch(path, makedirs=True)

  def testMountWriteUnmountRead(self):
    mount_call = self.PatchObject(osutils, 'MountDir')
    umount_call = self.PatchObject(osutils, 'UmountDir')
    for cleanup in (True, False):
      with osutils.MountOverlayContext(self.lowerdir, self.upperdir,
                                       self.mergeddir, cleanup=cleanup):
        mount_call.assert_any_call(
            'overlay', self.mergeddir, fs_type='overlay', makedirs=False,
            mount_opts=('lowerdir=%s' % self.lowerdir,
                        'upperdir=%s' % self.upperdir,
                        mock.ANY),
            quiet=mock.ANY)
      umount_call.assert_any_call(self.mergeddir, cleanup=cleanup)

  def testMountFailFallback(self):
    """Test that mount failure with overlay fs_type fallsback to overlayfs."""
    def _FailOverlay(*_args, **kwargs):
      if kwargs['fs_type'] == 'overlay':
        raise cros_build_lib.RunCommandError(
            'Phony failure',
            cros_build_lib.CommandResult(cmd=['MounDir'], returncode=32))

    mount_call = self.PatchObject(osutils, 'MountDir')
    mount_call.side_effect = _FailOverlay
    umount_call = self.PatchObject(osutils, 'UmountDir')
    for cleanup in (True, False):
      with osutils.MountOverlayContext(self.lowerdir, self.upperdir,
                                       self.mergeddir, cleanup=cleanup):
        mount_call.assert_any_call(
            'overlay', self.mergeddir, fs_type='overlay', makedirs=False,
            mount_opts=('lowerdir=%s' % self.lowerdir,
                        'upperdir=%s' % self.upperdir,
                        mock.ANY),
            quiet=mock.ANY)
        mount_call.assert_any_call(
            'overlayfs', self.mergeddir, fs_type='overlayfs', makedirs=False,
            mount_opts=('lowerdir=%s' % self.lowerdir,
                        'upperdir=%s' % self.upperdir),
            quiet=mock.ANY)
      umount_call.assert_any_call(self.mergeddir, cleanup=cleanup)

  def testNoValidWorkdirFallback(self):
    """Test that we fallback to overlayfs when no valid workdir is found.."""
    def _FailFileSystemCheck(_path1, _path2):
      return False

    check_filesystem = self.PatchObject(osutils, '_SameFileSystem')
    check_filesystem.side_effect = _FailFileSystemCheck
    mount_call = self.PatchObject(osutils, 'MountDir')
    umount_call = self.PatchObject(osutils, 'UmountDir')

    for cleanup in (True, False):
      with osutils.MountOverlayContext(self.lowerdir, self.upperdir,
                                       self.mergeddir, cleanup=cleanup):
        mount_call.assert_any_call(
            'overlayfs', self.mergeddir, fs_type='overlayfs', makedirs=False,
            mount_opts=('lowerdir=%s' % self.lowerdir,
                        'upperdir=%s' % self.upperdir),
            quiet=mock.ANY)
      umount_call.assert_any_call(self.mergeddir, cleanup=cleanup)


class IterateMountPointsTests(cros_test_lib.TempDirTestCase):
  """Test for IterateMountPoints function."""

  def setUp(self):
    self.proc_mount = os.path.join(self.tempdir, 'mounts')
    osutils.WriteFile(
        self.proc_mount,
        r"""/dev/loop0 /mnt/dir_8 ext4 rw,relatime,data=ordered 0 0
/dev/loop2 /mnt/dir_1 ext4 rw,relatime,data=ordered 0 0
/dev/loop1 /mnt/dir_12 vfat rw 0 0
/dev/loop4 /mnt/dir_3 ext4 ro,relatime 0 0
weird\040system /mnt/weirdo unknown ro 0 0
tmpfs /mnt/spaced\040dir tmpfs ro 0 0
tmpfs /mnt/\134 tmpfs ro 0 0
"""
    )

  def testOkay(self):
    r = list(osutils.IterateMountPoints(self.proc_mount))
    self.assertEqual(len(r), 7)
    self.assertEqual(r[0].source, '/dev/loop0')
    self.assertEqual(r[1].destination, '/mnt/dir_1')
    self.assertEqual(r[2].filesystem, 'vfat')
    self.assertEqual(r[3].options, 'ro,relatime')

  def testEscape(self):
    r = list(osutils.IterateMountPoints(self.proc_mount))
    self.assertEqual(r[4].source, 'weird system')
    self.assertEqual(r[5].destination, '/mnt/spaced dir')
    self.assertEqual(r[6].destination, '/mnt/\\')


class ResolveSymlinkInRootTest(cros_test_lib.TempDirTestCase):
  """Tests for ResolveSymlinkInRoot."""

  def setUp(self):
    # Create symlinks in tempdir so they are cleaned up automatically.
    os.chdir(self.tempdir)

  def testRelativeLink(self):
    os.symlink('target', 'link')
    self.assertEqual(osutils.ResolveSymlinkInRoot('link', '/root'), 'target')

  def testAbsoluteLink(self):
    os.symlink('/target', 'link')
    self.assertEqual(osutils.ResolveSymlinkInRoot('link', '/root'),
                     '/root/target')

  def testRecursion(self):
    os.symlink('target', 'link1')
    os.symlink('link1', 'link2')
    self.assertEqual(osutils.ResolveSymlinkInRoot('link2', '/root'), 'target')

  def testRecursionWithAbsoluteLink(self):
    os.symlink('target', 'link1')
    os.symlink('/link1', 'link2')
    self.assertEqual(osutils.ResolveSymlinkInRoot('link2', '.'), './target')


class ResolveSymlinkTest(cros_test_lib.TempDirTestCase):
  """Tests for ResolveSymlink."""

  def setUp(self):
    self.file_path = os.path.join(self.tempdir, 'file')
    self.dir_path = os.path.join(self.tempdir, 'directory')
    osutils.Touch(self.file_path)
    osutils.SafeMakedirs(self.dir_path)

    os.chdir(self.tempdir)
    osutils.SafeSymlink(self.file_path, 'abs_file_symlink')
    osutils.SafeSymlink('./file', 'rel_file_symlink')
    osutils.SafeSymlink(self.dir_path, 'abs_dir_symlink')
    osutils.SafeSymlink('./directory', 'rel_dir_symlink')

    self.abs_file_symlink = os.path.join(self.tempdir, 'abs_file_symlink')
    self.rel_file_symlink = os.path.join(self.tempdir, 'rel_file_symlink')
    self.abs_dir_symlink = os.path.join(self.tempdir, 'abs_dir_symlink')
    self.rel_dir_symlink = os.path.join(self.tempdir, 'rel_dir_symlink')

  def testAbsoluteResolution(self):
    """Test absolute path resolutions."""
    self.assertEqual(self.file_path,
                     osutils.ResolveSymlink(self.abs_file_symlink))
    self.assertEqual(self.dir_path,
                     osutils.ResolveSymlink(self.abs_dir_symlink))

  def testRelativeResolution(self):
    """Test relative path resolutions."""
    self.assertEqual(self.file_path,
                     osutils.ResolveSymlink(self.rel_file_symlink))
    self.assertEqual(self.dir_path,
                     osutils.ResolveSymlink(self.rel_dir_symlink))


class IsInsideVmTest(cros_test_lib.MockTempDirTestCase):
  """Test osutils.IsInsideVmTest function."""

  def setUp(self):
    self.model_file = os.path.join(self.tempdir, 'sda', 'device', 'model')
    osutils.SafeMakedirs(os.path.dirname(self.model_file))
    self.mock_glob = self.PatchObject(
        glob, 'glob', return_value=[self.model_file])

  def testIsInsideVm(self):
    osutils.WriteFile(self.model_file, 'VBOX')
    self.assertTrue(osutils.IsInsideVm())
    self.assertEqual(self.mock_glob.call_args[0][0],
                     '/sys/block/*/device/model')

    osutils.WriteFile(self.model_file, 'VMware')
    self.assertTrue(osutils.IsInsideVm())

  def testIsNotInsideVm(self):
    osutils.WriteFile(self.model_file, 'ST1000DM000-1CH1')
    self.assertFalse(osutils.IsInsideVm())


class CopyDirContentsTestCase(cros_test_lib.TempDirTestCase):
  """Test CopyDirContents."""

  def testCopyEmptyDir(self):
    """Copy "empty" contents from a dir."""
    in_dir = os.path.join(self.tempdir, 'input')
    out_dir = os.path.join(self.tempdir, 'output')
    osutils.SafeMakedirsNonRoot(in_dir)
    osutils.SafeMakedirsNonRoot(out_dir)
    osutils.CopyDirContents(in_dir, out_dir)

  def testCopyFiles(self):
    """Copy from a dir that contains files."""
    in_dir = os.path.join(self.tempdir, 'input')
    out_dir = os.path.join(self.tempdir, 'output')
    osutils.SafeMakedirsNonRoot(in_dir)
    osutils.WriteFile(os.path.join(in_dir, 'a.txt'), 'aaa')
    osutils.WriteFile(os.path.join(in_dir, 'b.txt'), 'bbb')
    osutils.SafeMakedirsNonRoot(out_dir)
    osutils.CopyDirContents(in_dir, out_dir)
    self.assertEqual(
        osutils.ReadFile(os.path.join(out_dir, 'a.txt')).strip(), 'aaa')
    self.assertEqual(
        osutils.ReadFile(os.path.join(out_dir, 'b.txt')).strip(), 'bbb')

  def testCopyTree(self):
    """Copy from a dir that contains files."""
    in_dir = os.path.join(self.tempdir, 'input')
    out_dir = os.path.join(self.tempdir, 'output')
    osutils.SafeMakedirsNonRoot(in_dir)
    osutils.SafeMakedirsNonRoot(os.path.join(in_dir, 'a'))
    osutils.WriteFile(os.path.join(in_dir, 'a', 'b.txt'), 'bbb')
    osutils.SafeMakedirsNonRoot(out_dir)
    osutils.CopyDirContents(in_dir, out_dir)
    self.assertEqual(
        osutils.ReadFile(os.path.join(out_dir, 'a', 'b.txt')).strip(), 'bbb')

  def testSourceDirDoesNotExistRaises(self):
    """Coping from a non-existent source dir raises."""
    in_dir = os.path.join(self.tempdir, 'input')
    out_dir = os.path.join(self.tempdir, 'output')
    osutils.SafeMakedirsNonRoot(out_dir)
    with self.assertRaises(osutils.BadPathsException):
      osutils.CopyDirContents(in_dir, out_dir)

  def testDestinationDirDoesNotExistRaises(self):
    """Coping to a non-existent destination dir raises."""
    in_dir = os.path.join(self.tempdir, 'input')
    out_dir = os.path.join(self.tempdir, 'output')
    osutils.SafeMakedirsNonRoot(in_dir)
    with self.assertRaises(osutils.BadPathsException):
      osutils.CopyDirContents(in_dir, out_dir)

  def testDestinationDirNonEmptyRaises(self):
    """Coping to a non-empty destination dir raises."""
    in_dir = os.path.join(self.tempdir, 'input')
    out_dir = os.path.join(self.tempdir, 'output')
    osutils.SafeMakedirsNonRoot(in_dir)
    osutils.SafeMakedirsNonRoot(out_dir)
    osutils.SafeMakedirsNonRoot(os.path.join(out_dir, 'blah'))
    with self.assertRaises(osutils.BadPathsException):
      osutils.CopyDirContents(in_dir, out_dir)

  def testDestinationDirNonEmptyAllowNonEmptySet(self):
    """Copying to a non-empty destination with allow_nonempty does not raise."""
    in_dir = os.path.join(self.tempdir, 'input')
    out_dir = os.path.join(self.tempdir, 'output')
    osutils.SafeMakedirsNonRoot(in_dir)
    osutils.SafeMakedirsNonRoot(out_dir)
    osutils.SafeMakedirsNonRoot(os.path.join(out_dir, 'blah'))
    osutils.CopyDirContents(in_dir, out_dir, allow_nonempty=True)

  def testCopyingSymlinks(self):
    in_dir = os.path.join(self.tempdir, 'input')
    in_dir_link = os.path.join(in_dir, 'link')
    in_dir_symlinks_dir = os.path.join(in_dir, 'holding_symlink')
    in_dir_symlinks_dir_link = os.path.join(in_dir_symlinks_dir, 'link')

    out_dir = os.path.join(self.tempdir, 'output')
    out_dir_link = os.path.join(out_dir, 'link')
    out_dir_symlinks_dir = os.path.join(out_dir, 'holding_symlink')
    out_dir_symlinks_dir_link = os.path.join(out_dir_symlinks_dir, 'link')

    # Create directories and symlinks appropriately.
    osutils.SafeMakedirsNonRoot(in_dir)
    osutils.SafeMakedirsNonRoot(in_dir_symlinks_dir)
    os.symlink(self.tempdir, in_dir_link)
    os.symlink(self.tempdir, in_dir_symlinks_dir_link)

    osutils.SafeMakedirsNonRoot(out_dir)

    # Actual execution.
    osutils.CopyDirContents(in_dir, out_dir, symlinks=True)

    # Assertions.
    self.assertTrue(os.path.islink(out_dir_link))
    self.assertTrue(os.path.islink(out_dir_symlinks_dir_link))

  def testNotCopyingSymlinks(self):
    # Create temporary to symlink against.
    tmp_file = os.path.join(self.tempdir, 'a.txt')
    osutils.WriteFile(tmp_file, 'aaa')
    tmp_subdir = os.path.join(self.tempdir, 'tmp')
    osutils.SafeMakedirsNonRoot(tmp_subdir)
    tmp_subdir_file = os.path.join(tmp_subdir, 'b.txt')
    osutils.WriteFile(tmp_subdir_file, 'bbb')

    in_dir = os.path.join(self.tempdir, 'input')
    in_dir_link = os.path.join(in_dir, 'link')
    in_dir_symlinks_dir = os.path.join(in_dir, 'holding_symlink')
    in_dir_symlinks_dir_link = os.path.join(in_dir_symlinks_dir, 'link')

    out_dir = os.path.join(self.tempdir, 'output')
    out_dir_file = os.path.join(out_dir, 'link')
    out_dir_symlinks_dir = os.path.join(out_dir, 'holding_symlink')
    out_dir_symlinks_dir_subdir = os.path.join(out_dir_symlinks_dir, 'link')

    # Create directories and symlinks appropriately.
    osutils.SafeMakedirsNonRoot(in_dir)
    osutils.SafeMakedirsNonRoot(in_dir_symlinks_dir)
    os.symlink(tmp_file, in_dir_link)
    os.symlink(tmp_subdir, in_dir_symlinks_dir_link)

    osutils.SafeMakedirsNonRoot(out_dir)

    # Actual execution.
    osutils.CopyDirContents(in_dir, out_dir, symlinks=False)

    # Assertions.
    self.assertFalse(os.path.islink(out_dir_file))
    self.assertFalse(os.path.islink(out_dir_symlinks_dir_subdir))
    self.assertTrue(filecmp.cmp(out_dir_file, tmp_file))
    self.assertTrue(
        filecmp.cmp(os.path.join(out_dir_symlinks_dir_subdir, 'b.txt'),
                    tmp_subdir_file))

  @unittest.skipIf(sys.version_info.major < 3, 'Requires pathlib from py3')
  def testCopyingSymlinksAndFilesWithPathArgs(self):
    """Copying given |Path| arguments works properly for symlinks+files."""
    in_dir = Path(self.tempdir) / 'input'
    osutils.SafeMakedirs(in_dir)

    tmp_file = in_dir / 'a.txt'
    tmp_file.write_text('aaa', encoding='utf-8')
    tmp_file_link = tmp_file.with_suffix('.link')
    tmp_file_link.symlink_to(tmp_file)

    out_dir = Path(self.tempdir) / 'output'
    osutils.SafeMakedirs(out_dir)
    osutils.CopyDirContents(in_dir, out_dir, symlinks=True)

    out_tmp_file = out_dir / tmp_file.name
    self.assertEqual(out_tmp_file.read_text(encoding='utf-8'), 'aaa')
    out_tmp_file_link = out_dir / tmp_file_link.name
    self.assertEqual(Path(os.readlink(out_tmp_file_link)), tmp_file)

  @unittest.skipIf(sys.version_info.major < 3, 'Requires pathlib from py3')
  def testCopyingSubDirWithPathArgs(self):
    """Copying given |Path| arguments works properly for subdirectories."""
    in_dir = Path(self.tempdir) / 'input'
    osutils.SafeMakedirs(in_dir)

    tmp_file = in_dir / 'subdir' / 'a.txt'
    osutils.SafeMakedirs(tmp_file.parent)

    tmp_file.write_text('aaa', encoding='utf-8')

    out_dir = Path(self.tempdir) / 'output'
    osutils.SafeMakedirs(out_dir)
    osutils.CopyDirContents(in_dir, out_dir, symlinks=True)

    out_tmp_file = out_dir / 'subdir' / tmp_file.name
    self.assertEqual(out_tmp_file.read_text(encoding='utf-8'), 'aaa')


class WhichTests(cros_test_lib.TempDirTestCase):
  """Test Which."""

  def setUp(self):
    self.prog_path = os.path.join(self.tempdir, 'prog')
    osutils.Touch(self.prog_path, mode=0o755)
    self.text_path = os.path.join(self.tempdir, 'text')
    osutils.Touch(self.text_path, mode=0o644)

    # A random path for us to validate.
    os.environ['PATH'] = '/:%s' % (self.tempdir,)

  def testPath(self):
    """Check $PATH/path handling."""
    self.assertEqual(self.prog_path, osutils.Which('prog'))

    os.environ['PATH'] = ''
    self.assertEqual(None, osutils.Which('prog'))

    self.assertEqual(self.prog_path, osutils.Which('prog', path=self.tempdir))

  def testMode(self):
    """Check mode handling."""
    self.assertEqual(self.prog_path, osutils.Which('prog'))
    self.assertEqual(self.prog_path, osutils.Which('prog', mode=os.X_OK))
    self.assertEqual(self.prog_path, osutils.Which('prog', mode=os.R_OK))
    self.assertEqual(None, osutils.Which('text'))
    self.assertEqual(None, osutils.Which('text', mode=os.X_OK))
    self.assertEqual(self.text_path, osutils.Which('text', mode=os.F_OK))

  def testRoot(self):
    """Check root handling."""
    self.assertEqual(None, osutils.Which('prog', root='/.........'))
    self.assertEqual(self.prog_path, osutils.Which('prog', path='/',
                                                   root=self.tempdir))


class UmaskTests(cros_test_lib.TestCase):
  """Test Umask."""

  @staticmethod
  def getUmask():
    """Return the current umask setting."""
    # Testing this is messy because there is no syscall to look this up without
    # side-effects.  os.umask sets & queries at once.
    m = re.search(r'^Umask:\s+([0-7]+)', osutils.ReadFile('/proc/self/status'),
                  flags=re.M)
    assert m is not None
    return int(m.group(1), 8)

  def testBasic(self):
    """Verify umask is saved & restored."""
    os.umask(0o222)
    with osutils.UmaskContext(0o123) as old:
      assert self.getUmask() == 0o123
    assert self.getUmask() == old
    assert old == 0o222