# Copyright 2015 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.

"""Tests for the sysroot library."""

import os
from typing import Iterable, List, Optional, Tuple

from chromite.api.gen.chromiumos import common_pb2
from chromite.lib import chroot_lib
from chromite.lib import constants
from chromite.lib import cros_build_lib
from chromite.lib import cros_test_lib
from chromite.lib import osutils
from chromite.lib import sysroot_lib
from chromite.lib import toolchain
from chromite.lib import unittest_lib
from chromite.lib.parser import package_info


class SysrootLibTest(cros_test_lib.MockTempDirTestCase):
  """Unittest for sysroot_lib.py"""

  def setUp(self):
    """Setup the test environment."""
    self.PatchObject(cros_build_lib, 'IsInsideChroot', return_value=False)
    # Fake being root to avoid running all filesystem commands with sudo_run.
    self.PatchObject(osutils, 'IsRootUser', return_value=True)
    sysroot_path = os.path.join(self.tempdir, 'sysroot')
    osutils.SafeMakedirs(sysroot_path)
    self.sysroot = sysroot_lib.Sysroot(sysroot_path)
    self.relative_sysroot = sysroot_lib.Sysroot('sysroot')
    # make.conf needs to exist to correctly read back config.
    unittest_lib.create_stub_make_conf(sysroot_path)

  def _writeOverlays(self,
                     board_overlays: Optional[Iterable[str]] = None,
                     portdir_overlays: Optional[Iterable[str]] = None,
                     board: str = None) -> Tuple[List[str], List[str]]:
    """Helper function to write board and portdir overlays for the sysroot.

    By default uses one fake board overlay, and the chromiumos and portage
    stable overlays. Set the arguments to an empty list to set no values for
    that field. When not explicitly set, |portdir_overlays| includes all values
    in |board_overlays|.

    Returns:
      The board overlays, and the portdir overlays.
    """
    if board_overlays is None:
      board_overlays = ['overlay/board']
    if portdir_overlays is None:
      portdir_overlays = [
          constants.CHROMIUMOS_OVERLAY_DIR, constants.PORTAGE_STABLE_OVERLAY_DIR
      ] + board_overlays

    board_overlays_field = sysroot_lib.STANDARD_FIELD_BOARD_OVERLAY
    portdir_field = sysroot_lib.STANDARD_FIELD_PORTDIR_OVERLAY
    board_field = sysroot_lib.STANDARD_FIELD_BOARD_USE

    board_values = [
        f'{constants.CHROOT_SOURCE_ROOT}/{x}' for x in board_overlays
    ]
    board_value = '\n'.join(board_values)

    portdir_values = [
        f'{constants.CHROOT_SOURCE_ROOT}/{x}' for x in portdir_overlays
    ]
    portdir_value = '\n'.join(portdir_values)

    config_values = {}
    if board_values:
      config_values[board_overlays_field] = board_value
    if portdir_values:
      config_values[portdir_field] = portdir_value
    if board:
      config_values[board_field] = board

    config = '\n'.join(f'{k}="{v}"' for k, v in config_values.items())
    self.sysroot.WriteConfig(config)

    return board_values, portdir_values

  def testGetStandardField(self):
    """Tests that standard field can be fetched correctly."""
    self.sysroot.WriteConfig('FOO="bar"')
    self.assertEqual('bar', self.sysroot.GetStandardField('FOO'))

    # Works with multiline strings
    multiline = """foo
bar
baz
"""
    self.sysroot.WriteConfig('TEST="%s"' % multiline)
    self.assertEqual(multiline, self.sysroot.GetStandardField('TEST'))

  def testReadWriteCache(self):
    """Tests that we can write and read to the cache."""
    # If a field is not defined we get None.
    self.assertEqual(None, self.sysroot.GetCachedField('foo'))

    # If we set a field, we can get it.
    self.sysroot.SetCachedField('foo', 'bar')
    self.assertEqual('bar', self.sysroot.GetCachedField('foo'))

    # Setting a field in an existing cache preserve the previous values.
    self.sysroot.SetCachedField('hello', 'bonjour')
    self.assertEqual('bar', self.sysroot.GetCachedField('foo'))
    self.assertEqual('bonjour', self.sysroot.GetCachedField('hello'))

    # Setting a field to None unsets it.
    self.sysroot.SetCachedField('hello', None)
    self.assertEqual(None, self.sysroot.GetCachedField('hello'))

  def testErrorOnBadCachedValue(self):
    """Tests that we detect bad value for the sysroot cache."""
    forbidden = [
        'hello"bonjour',
        'hello\\bonjour',
        'hello\nbonjour',
        'hello$bonjour',
        'hello`bonjour',
    ]
    for value in forbidden:
      with self.assertRaises(ValueError):
        self.sysroot.SetCachedField('FOO', value)

  def testGenerateConfigNoToolchainRaisesError(self):
    """Tests _GenerateConfig() with no toolchain raises an error."""
    self.PatchObject(toolchain, 'FilterToolchains', autospec=True,
                     return_value={})

    with self.assertRaises(sysroot_lib.ConfigurationError):
      # pylint: disable=protected-access
      self.sysroot._GenerateConfig({}, ['foo_overlay'], ['foo_overlay'], '')

  def testExists(self):
    """Tests the Exists method."""
    self.assertTrue(self.sysroot.Exists())

    dne_sysroot = sysroot_lib.Sysroot(os.path.join(self.tempdir, 'DNE'))
    self.assertFalse(dne_sysroot.Exists())

  def testExistsInChroot(self):
    """Test the Exists method with a chroot."""
    chroot = chroot_lib.Chroot(self.tempdir)
    self.assertTrue(self.relative_sysroot.Exists(chroot=chroot))

  def testEquals(self):
    """Sanity check for the __eq__ methods."""
    sysroot1 = sysroot_lib.Sysroot(self.tempdir)
    sysroot2 = sysroot_lib.Sysroot(self.tempdir)
    self.assertEqual(sysroot1, sysroot2)

  def testProfileName(self):
    """Test the profile_name property when a value is set."""
    profile = 'foo'
    self.sysroot.SetCachedField(sysroot_lib.CACHED_FIELD_PROFILE_OVERRIDE,
                                profile)
    self.assertEqual(profile, self.sysroot.profile_name)

  def testProfileNameDefault(self):
    """Test the profile_name property when no value is set."""
    self.assertEqual(sysroot_lib.DEFAULT_PROFILE, self.sysroot.profile_name)

  def testBoardOverlay(self):
    """Test the board_overlay property."""
    board_overlays, _portdir_overlays = self._writeOverlays()

    self.assertEqual(sorted(board_overlays), sorted(self.sysroot.board_overlay))

  def testBuildTargetOverlays(self):
    """Tests for populated _build_target_overlay[s]."""
    private = '/path/to/overlay-x-private'
    expected = ['/path/to/overlay-x', private]
    overlays = expected + ['/path/to/chromeos-overlay']
    self._writeOverlays(overlays, board='x')

    # pylint: disable=protected-access
    results = [str(x) for x in self.sysroot._build_target_overlays]
    self.assertEqual(len(expected), len(results))
    for current in expected:
      self.assertTrue(any(result.endswith(current) for result in results))

    self.assertTrue(str(self.sysroot.build_target_overlay).endswith(private))

  def testNoBuildTargetOverlay(self):
    """Test for no standard build target overlay."""
    self._writeOverlays(['/path/to/chromeos-overlay', '/path/to/chipset-x'])

    # pylint: disable=protected-access
    self.assertEqual(0, len(self.sysroot._build_target_overlays))
    self.assertIsNone(self.sysroot.build_target_overlay)

  def testChipset(self):
    """Test for extracting a valid chipset."""
    expected = 'foo'
    chipsets = [
        f'/path/to/chipset-{expected}', f'/path/to/chipset-{expected}-private'
    ]
    all_overlays = chipsets + ['/path/to/chromeos-overlay']
    self._writeOverlays(all_overlays)

    self.assertEqual(expected, self.sysroot.chipset)

  def testNoChipset(self):
    """Test for handling no retrievable chipset value."""
    self._writeOverlays(['/path/to/chromeos-overlay', '/path/to/overlay-board'])
    self.assertIsNone(self.sysroot.chipset)

  def testOverlays(self):
    """Test the overlays property."""
    _board_overlays, portdir_overlays = self._writeOverlays()

    self.assertEqual(portdir_overlays, self.sysroot.portdir_overlay)

  def testGetOverlays(self):
    """Test the get_overlays function."""
    board_overlays, portdir_overlays = self._writeOverlays()

    self.assertEqual(
        board_overlays,
        [str(x) for x in self.sysroot.get_overlays(build_target_only=True)])
    self.assertEqual(portdir_overlays,
                     [str(x) for x in self.sysroot.get_overlays()])

  def testGetOverlaysRelative(self):
    portdir_overlays = [
        constants.CHROMIUMOS_OVERLAY_DIR, constants.PORTAGE_STABLE_OVERLAY_DIR
    ]
    self._writeOverlays(portdir_overlays=portdir_overlays)

    self.assertEqual(portdir_overlays,
                     [str(x) for x in self.sysroot.get_overlays(relative=True)])


class ProfileTest(cros_test_lib.MockTempDirTestCase):
  """Unittest for sysroot_lib.py"""

  def testConversion(self):
    """Test converting to/from protobuf."""
    profile1 = sysroot_lib.Profile('profile')
    profile2 = sysroot_lib.Profile()

    proto1 = common_pb2.Profile(name='profile')
    proto2 = common_pb2.Profile()

    self.assertEqual(profile1.as_protobuf, proto1)
    self.assertEqual(profile2.as_protobuf, proto2)

    self.assertEqual(profile1, sysroot_lib.Profile.from_protobuf(proto1))
    self.assertEqual(profile2, sysroot_lib.Profile.from_protobuf(proto2))

  def testEquality(self):
    """Test that equality functions work."""

    profile = sysroot_lib.Profile('profile')
    self.assertEqual(profile, sysroot_lib.Profile('profile'))
    self.assertNotEqual(profile, sysroot_lib.Profile('other'))
    self.assertNotEqual(profile, sysroot_lib.Profile(''))


class SysrootLibInstallConfigTest(cros_test_lib.MockTempDirTestCase):
  """Unittest for sysroot_lib.py"""

  # pylint: disable=protected-access

  def setUp(self):
    """Setup the test environment."""
    # Fake being root to avoid running all filesystem commands with sudo_run.
    self.PatchObject(osutils, 'IsRootUser', return_value=True)
    self.sysroot = sysroot_lib.Sysroot(self.tempdir)
    self.make_conf_generic_target = os.path.join(self.tempdir,
                                                 'make.conf.generic-target')
    self.make_conf_user = os.path.join(self.tempdir, 'make.conf.user')

    D = cros_test_lib.Directory
    filesystem = (
        D('etc', ()),
        'make.conf.generic-target',
        'make.conf.user',
    )

    cros_test_lib.CreateOnDiskHierarchy(self.tempdir, filesystem)

  def testInstallMakeConf(self):
    """Test make.conf installation."""
    self.PatchObject(sysroot_lib, '_GetMakeConfGenericPath',
                     return_value=self.make_conf_generic_target)

    self.sysroot.InstallMakeConf()

    filepath = os.path.join(self.tempdir, sysroot_lib._MAKE_CONF)
    self.assertExists(filepath)

  def testInstallMakeConfBoard(self):
    """Test make.conf.board installation."""
    self.PatchObject(self.sysroot, 'GenerateBoardMakeConf', return_value='#foo')
    self.PatchObject(self.sysroot, 'GenerateBinhostConf', return_value='#bar')

    self.sysroot.InstallMakeConfBoard()

    filepath = os.path.join(self.tempdir, sysroot_lib._MAKE_CONF_BOARD)
    content = '#foo\n#bar\n'
    self.assertExists(filepath)
    self.assertFileContents(filepath, content)

  def testInstallMakeConfBoardSetup(self):
    """Test make.conf.board_setup installation."""
    self.PatchObject(self.sysroot, 'GenerateBoardSetupConfig',
                     return_value='#foo')

    self.sysroot.InstallMakeConfBoardSetup('board')

    filepath = os.path.join(self.tempdir, sysroot_lib._MAKE_CONF_BOARD_SETUP)
    content = '#foo'
    self.assertExists(filepath)
    self.assertFileContents(filepath, content)

  def testInstallMakeConfUser(self):
    """Test make.conf.user installation."""
    self.PatchObject(sysroot_lib, '_GetChrootMakeConfUserPath',
                     return_value=self.make_conf_user)

    self.sysroot.InstallMakeConfUser()

    filepath = os.path.join(self.tempdir, sysroot_lib._MAKE_CONF_USER)
    self.assertExists(filepath)


class SysrootLibToolchainUpdateTest(cros_test_lib.RunCommandTempDirTestCase):
  """Sysroot.ToolchanUpdate tests."""

  def setUp(self):
    """Setup the test environment."""
    # Fake being root to avoid running commands with sudo_run.
    self.PatchObject(osutils, 'IsRootUser', return_value=True)

    self.sysroot = sysroot_lib.Sysroot(self.tempdir)
    self.emerge = os.path.join(constants.CHROMITE_BIN_DIR, 'parallel_emerge')

  def testDefaultUpdateToolchain(self):
    """Test the default path."""
    self.PatchObject(toolchain, 'InstallToolchain')

    self.sysroot.UpdateToolchain('board')
    self.assertCommandContains(
        [self.emerge, '--board=board', '--getbinpkg', '--usepkg'])

  def testNoLocalInitUpdateToolchain(self):
    """Test the nousepkg and not local case."""
    self.PatchObject(toolchain, 'InstallToolchain')

    self.sysroot.UpdateToolchain('board', local_init=False)
    self.assertCommandContains(['--getbinpkg', '--usepkg'], expected=False)
    self.assertCommandContains([self.emerge, '--board=board'])

  def testReUpdateToolchain(self):
    """Test behavior when not running for the first time."""
    self.PatchObject(toolchain, 'InstallToolchain')

    self.PatchObject(self.sysroot, 'IsToolchainInstalled', return_value=True)
    self.sysroot.UpdateToolchain('board')
    self.assertCommandContains([self.emerge], expected=False)

  def testInstallToolchainError(self):
    """Test error handling from the libc install."""
    failed = ['cat/pkg', 'cat/pkg2']
    failed_pkgs = [package_info.parse(pkg) for pkg in failed]
    result = cros_build_lib.CompletedProcess(returncode=1)
    error = toolchain.ToolchainInstallError('Error', result=result,
                                            tc_info=failed_pkgs)
    self.PatchObject(toolchain, 'InstallToolchain', side_effect=error)

    try:
      self.sysroot.UpdateToolchain('board')
    except sysroot_lib.ToolchainInstallError as e:
      self.assertTrue(e.failed_toolchain_info)
      self.assertEqual(failed_pkgs, e.failed_toolchain_info)
    except Exception as e:
      self.fail('Unexpected exception raised: %s' % type(e))
    else:
      self.fail('Expected an exception.')

  def testEmergeError(self):
    """Test the emerge error handling."""
    self.PatchObject(toolchain, 'InstallToolchain')
    # pylint: disable=protected-access
    command = self.sysroot._UpdateToolchainCommand('board', True)

    err = cros_build_lib.RunCommandError(
        'Error', cros_build_lib.CompletedProcess(returncode=1))
    self.rc.AddCmdResult(command, side_effect=err)

    with self.assertRaises(sysroot_lib.ToolchainInstallError):
      self.sysroot.UpdateToolchain('board', local_init=True)


def test_get_sdk_provided_packages(simple_sysroot):
  pkg_provided = simple_sysroot.path / 'etc/portage/profile/package.provided'
  content = """
foo/bar-2-r3

# Comment line.
cat/pkg-1.0.0 # Comment after package.
"""
  osutils.WriteFile(pkg_provided, content, makedirs=True)
  pkgs = list(sysroot_lib.get_sdk_provided_packages(simple_sysroot.path))
  expected = [package_info.parse(p) for p in ('foo/bar-2-r3', 'cat/pkg-1.0.0')]
  assert pkgs == expected