api/field_handler_unittest.py

# -*- coding: utf-8 -*-
# Copyright 2019 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.

"""field_handler module tests."""

from __future__ import print_function

import os
import sys

from chromite.api import field_handler
from chromite.api.gen.chromite.api import build_api_test_pb2
from chromite.api.gen.chromiumos import common_pb2
from chromite.lib import chroot_lib
from chromite.lib import cros_test_lib
from chromite.lib import osutils


assert sys.version_info >= (3, 6), 'This module requires Python 3.6+'


class ChrootHandlerTest(cros_test_lib.TestCase):
  """ChrootHandler tests."""

  def setUp(self):
    self.path = '/chroot/dir'
    self.cache_dir = '/cache/dir'
    self.chrome_dir = '/chrome/dir'
    self.env = {'FEATURES': 'thing', 'CHROME_ORIGIN': 'LOCAL_SOURCE'}
    self.expected_chroot = chroot_lib.Chroot(path=self.path,
                                             cache_dir=self.cache_dir,
                                             chrome_root=self.chrome_dir,
                                             env=self.env)

  def test_parse_chroot_success(self):
    """Test successful Chroot message parse."""
    chroot_msg = common_pb2.Chroot()
    chroot_msg.path = self.path
    chroot_msg.cache_dir = self.cache_dir
    chroot_msg.chrome_dir = self.chrome_dir
    chroot_msg.env.features.add().feature = 'thing'

    chroot_handler = field_handler.ChrootHandler(clear_field=False)
    parsed_chroot = chroot_handler.parse_chroot(chroot_msg)

    self.assertEqual(self.expected_chroot, parsed_chroot)

  def test_handle_success(self):
    """Test a successful Chroot message parse from a parent message."""
    message = build_api_test_pb2.TestRequestMessage()
    message.chroot.path = self.path
    message.chroot.cache_dir = self.cache_dir
    message.chroot.chrome_dir = self.chrome_dir
    message.chroot.env.features.add().feature = 'thing'

    # First a no-clear parse.
    chroot_handler = field_handler.ChrootHandler(clear_field=False)
    chroot = chroot_handler.handle(message)

    self.assertEqual(self.expected_chroot, chroot)
    self.assertEqual(message.chroot.path, self.path)

    # A clear field parse.
    clear_chroot_handler = field_handler.ChrootHandler(clear_field=True)
    chroot = clear_chroot_handler.handle(message)

    self.assertEqual(self.expected_chroot, chroot)
    self.assertFalse(message.chroot.path)

  def test_handle_empty_chroot_message(self):
    """Test handling of an empty chroot message."""
    message = build_api_test_pb2.TestRequestMessage()
    empty_chroot = chroot_lib.Chroot()

    chroot_handler = field_handler.ChrootHandler(clear_field=False)
    chroot = chroot_handler.handle(message)

    self.assertEqual(empty_chroot, chroot)


class CopyPathInTest(cros_test_lib.TempDirTestCase):
  """PathHandler tests."""

  def setUp(self):
    self.source_dir = os.path.join(self.tempdir, 'source')
    self.dest_dir = os.path.join(self.tempdir, 'destination')
    osutils.SafeMakedirs(self.source_dir)
    osutils.SafeMakedirs(self.dest_dir)

    self.source_file1 = os.path.join(self.source_dir, 'file1')
    self.file1_contents = 'file 1'
    osutils.WriteFile(self.source_file1, self.file1_contents)

    self.file2_contents = 'some data'
    self.source_file2 = os.path.join(self.source_dir, 'file2')
    osutils.WriteFile(self.source_file2, self.file2_contents)

  def _path_checks(self, source_file, dest_file, contents=None):
    """Set of common checks for the copied files/directories."""
    # Message should now reflect the new path.
    self.assertNotEqual(source_file, dest_file)
    # The new path should be in the destination directory.
    self.assertStartsWith(dest_file, self.dest_dir)
    # The new file should exist.
    self.assertExists(dest_file)

    if contents:
      # The contents should be the same as the source file.
      self.assertFileContents(dest_file, contents)

  def test_handle_file(self):
    """Test handling of a single file."""
    message = build_api_test_pb2.TestRequestMessage()
    message.path.path = self.source_file1
    message.path.location = common_pb2.Path.OUTSIDE

    with field_handler.copy_paths_in(message, self.dest_dir, delete=True):
      new_path = message.path.path
      self._path_checks(self.source_file1, new_path, self.file1_contents)

    # The file should have been deleted on exit with delete=True.
    self.assertNotExists(new_path)
    # The original should still exist.
    self.assertExists(self.source_file1)
    # The path should get reset.
    self.assertEqual(message.path.path, self.source_file1)

  def test_handle_files(self):
    """Test handling of multiple files."""
    message = build_api_test_pb2.TestRequestMessage()
    message.path.path = self.source_file1
    message.path.location = common_pb2.Path.OUTSIDE
    message.another_path.path = self.source_file2
    message.another_path.location = common_pb2.Path.OUTSIDE

    with field_handler.copy_paths_in(message, self.dest_dir, delete=False):
      new_path1 = message.path.path
      new_path2 = message.another_path.path

      self._path_checks(self.source_file1, new_path1, self.file1_contents)
      self._path_checks(self.source_file2, new_path2, self.file2_contents)

    # The files should still exist with delete=False.
    self.assertExists(new_path1)
    self.assertExists(new_path2)

  def test_handle_nested_file(self):
    """Test the nested path handling."""
    message = build_api_test_pb2.TestRequestMessage()
    message.nested_path.path.path = self.source_file1
    message.nested_path.path.location = common_pb2.Path.OUTSIDE

    with field_handler.copy_paths_in(message, self.dest_dir):
      new_path = message.nested_path.path.path
      self._path_checks(self.source_file1, new_path, self.file1_contents)

  def test_handle_directory(self):
    """Test handling of a directory."""
    message = build_api_test_pb2.TestRequestMessage()
    message.path.path = self.source_dir
    message.path.location = common_pb2.Path.OUTSIDE

    with field_handler.copy_paths_in(message, self.dest_dir):
      new_path = message.path.path

      self._path_checks(self.source_dir, self.dest_dir)
      # Make sure both directories have the same files.
      self.assertCountEqual(os.listdir(self.source_dir), os.listdir(new_path))

  def test_direction(self):
    """Test the direction argument preventing copies."""
    message = build_api_test_pb2.TestRequestMessage()
    message.path.path = self.source_file1
    message.path.location = common_pb2.Path.INSIDE

    with field_handler.copy_paths_in(message, self.dest_dir, delete=True):
      self.assertEqual(self.source_file1, message.path.path)

    # It should not be deleting the file when it doesn't need to copy it even
    # with delete=True.
    self.assertExists(self.source_file1)

  def test_prefix_inside(self):
    """Test the transfer inside prefix handling."""
    message = build_api_test_pb2.TestRequestMessage()
    message.path.path = self.source_dir
    message.path.location = common_pb2.Path.OUTSIDE

    with field_handler.copy_paths_in(message, self.dest_dir,
                                     prefix=self.tempdir):
      new_path = message.path.path
      # The prefix should be removed.
      self.assertFalse(new_path.startswith(self.tempdir))


class SyncDirsTest(cros_test_lib.TempDirTestCase):
  """Tests for sync_dirs."""

  def setUp(self):
    D = cros_test_lib.Directory
    filesystem = (
        D('chroot', (
            D('tmp', (
                D('tempdir', ()),
            )),
        )),
        D('sources', (
            D('single_file', ('single_file.txt',)),
            D('nested_directories', (
                'basedir_file.log',
                D('nested1', (
                    'nested1.txt',
                    D('nested2', ('nested2.txt',)),
                )),
            )),
        )),
    )
    cros_test_lib.CreateOnDiskHierarchy(self.tempdir, filesystem)

    self.chroot = os.path.join(self.tempdir, 'chroot')
    self.chroot_tmp = os.path.join(self.chroot, 'tmp')
    self.destination = os.path.join(self.chroot_tmp, 'tempdir')
    self.inside_path = '/tmp/tempdir'

    self.single_file_src = os.path.join(self.tempdir, 'sources', 'single_file')
    self.sf_src_file = os.path.join(self.single_file_src, 'single_file.txt')
    self.sf_dest_file = os.path.join(self.destination, 'single_file.txt')

    self.nested_dirs_src = (
        os.path.join(self.tempdir, 'sources', 'nested_directories'))
    self.nested_src_files = (
        os.path.join(self.nested_dirs_src, 'basedir_file.log'),
        os.path.join(self.nested_dirs_src, 'nested1', 'nested1.txt'),
        os.path.join(self.nested_dirs_src, 'nested1', 'nested2', 'nested2.txt'),
    )
    self.nested_dest_files = (
        os.path.join(self.destination, 'basedir_file.log'),
        os.path.join(self.destination, 'nested1', 'nested1.txt'),
        os.path.join(self.destination, 'nested1', 'nested2', 'nested2.txt'),
    )

    self.message = build_api_test_pb2.TestRequestMessage()

  def _assertExist(self, files):
    for f in files:
      self.assertExists(f)

  def _assertNotExist(self, files):
    for f in files:
      self.assertNotExists(f)

  def testSingleFileTransfer(self):
    """Single source file syncs."""
    self.message.synced_dir.dir = self.single_file_src

    # Verify source files exist and destination files do not.
    self.assertExists(self.sf_src_file)
    self.assertNotExists(self.sf_dest_file)

    with field_handler.sync_dirs(self.message, self.destination, self.chroot):
      # Verify the prefix is getting correctly stripped.
      self.assertEqual(self.message.synced_dir.dir, self.inside_path)
      # Verify the files have all been correctly copied in.
      self.assertExists(self.sf_dest_file)

    self.assertEqual(self.message.synced_dir.dir, self.single_file_src)
    # Verify the files have all been copied out.
    self.assertExists(self.sf_src_file)

  def testNestedFileSync(self):
    """Nested directories and files sync."""
    self.message.synced_dir.dir = self.nested_dirs_src

    self._assertExist(self.nested_src_files)
    self._assertNotExist(self.nested_dest_files)

    with field_handler.sync_dirs(self.message, self.destination, self.chroot):
      self.assertEqual(self.message.synced_dir.dir, self.inside_path)
      self._assertExist(self.nested_dest_files)

    self.assertEqual(self.message.synced_dir.dir, self.nested_dirs_src)
    self._assertExist(self.nested_src_files)

  def testDeletion(self):
    """Test file deletions are exported correctly."""
    self.message.synced_dir.dir = self.nested_dirs_src

    deleted_src = os.path.join(self.nested_dirs_src, 'nested1', 'nested1.txt')
    deleted_dest = os.path.join(self.destination, 'nested1', 'nested1.txt')

    self._assertExist(self.nested_src_files)
    self._assertNotExist(self.nested_dest_files)

    with field_handler.sync_dirs(self.message, self.destination, self.chroot):
      self._assertExist(self.nested_dest_files)
      osutils.SafeUnlink(deleted_dest)

    self._assertExist(set(self.nested_src_files) - {deleted_src})
    self.assertNotExists(deleted_src)

  def testCreation(self):
    """Test file creations are exported correctly."""
    self.message.synced_dir.dir = self.nested_dirs_src

    new_src = os.path.join(self.nested_dirs_src, 'new_dir', 'new_file')
    new_dest = os.path.join(self.destination, 'new_dir', 'new_file')

    self._assertExist(self.nested_src_files)
    self._assertNotExist(self.nested_dest_files)

    with field_handler.sync_dirs(self.message, self.destination, self.chroot):
      self._assertExist(self.nested_dest_files)
      osutils.Touch(new_dest, makedirs=True)

    self._assertExist(self.nested_src_files)
    self.assertExists(new_src)

  def testModification(self):
    """Test file modifications are exported correctly."""
    self.message.synced_dir.dir = self.single_file_src

    self.assertExists(self.sf_src_file)
    self.assertNotExists(self.sf_dest_file)

    self.assertEqual('', osutils.ReadFile(self.sf_src_file))
    file_content = 'Content!'

    with field_handler.sync_dirs(self.message, self.destination, self.chroot):
      self.assertExists(self.sf_dest_file)
      osutils.WriteFile(self.sf_dest_file, file_content)

    self.assertExists(self.sf_src_file)
    self.assertEqual(file_content, osutils.ReadFile(self.sf_src_file))


class ExtractResultsTest(cros_test_lib.TempDirTestCase):
  """Tests for extract_results."""

  def setUp(self):
    # Setup the directories.
    self.chroot_dir = os.path.join(self.tempdir, 'chroot')
    self.source_dir = '/source'
    self.chroot_source = os.path.join(self.chroot_dir,
                                      self.source_dir.lstrip(os.sep))
    self.source_dir2 = '/source2'
    self.chroot_source2 = os.path.join(self.chroot_dir,
                                       self.source_dir2.lstrip(os.sep))
    self.dest_dir = os.path.join(self.tempdir, 'destination')
    osutils.SafeMakedirs(self.chroot_source)
    osutils.SafeMakedirs(self.chroot_source2)
    osutils.SafeMakedirs(self.dest_dir)

    # Two files in the same directory inside the chroot.
    self.source_file1 = os.path.join(self.chroot_source, 'file1')
    self.source_file1_inside = os.path.join(self.source_dir, 'file1')
    self.file1_contents = 'file 1'
    osutils.WriteFile(self.source_file1, self.file1_contents)

    self.file2_contents = 'some data'
    self.source_file2 = os.path.join(self.chroot_source, 'file2')
    self.source_file2_inside = os.path.join(self.source_dir, 'file2')
    osutils.WriteFile(self.source_file2, self.file2_contents)

    # Third file in a different location.
    self.file3_contents = 'another file'
    self.source_file3 = os.path.join(self.chroot_source2, 'file3')
    self.source_file3_inside = os.path.join(self.source_dir2, 'file3')
    osutils.WriteFile(self.source_file3, self.file3_contents)

    self.request = build_api_test_pb2.TestRequestMessage()
    self.request.result_path.path.path = self.dest_dir
    self.request.result_path.path.location = common_pb2.Path.OUTSIDE
    self.response = build_api_test_pb2.TestResultMessage()
    self.chroot = chroot_lib.Chroot(path=self.chroot_dir)

  def _path_checks(self, path, destination, contents=None):
    self.assertTrue(path)
    self.assertStartsWith(path, destination)
    self.assertExists(path)
    if contents:
      self.assertFileContents(path, contents)

  def test_single_file(self):
    """Test a single file."""
    self.response.artifact.path = self.source_file1_inside
    self.response.artifact.location = common_pb2.Path.INSIDE

    field_handler.extract_results(self.request, self.response, self.chroot)

    self._path_checks(self.response.artifact.path, self.dest_dir,
                      contents=self.file1_contents)

  def test_single_directory(self):
    """Test a single directory."""
    self.response.artifact.path = self.source_dir
    self.response.artifact.location = common_pb2.Path.INSIDE

    field_handler.extract_results(self.request, self.response, self.chroot)

    self._path_checks(self.response.artifact.path, self.dest_dir)
    self.assertCountEqual(os.listdir(self.chroot_source),
                          os.listdir(self.response.artifact.path))

  def test_multiple_files(self):
    """Test multiple files."""
    self.response.artifact.path = self.source_file1_inside
    self.response.artifact.location = common_pb2.Path.INSIDE
    self.response.nested_artifact.path.path = self.source_file2_inside
    self.response.nested_artifact.path.location = common_pb2.Path.INSIDE

    artifact3 = self.response.artifacts.add()
    artifact3.path = self.source_file3_inside
    artifact3.location = common_pb2.Path.INSIDE

    field_handler.extract_results(self.request, self.response, self.chroot)

    self._path_checks(self.response.artifact.path, self.dest_dir,
                      contents=self.file1_contents)
    self._path_checks(self.response.nested_artifact.path.path, self.dest_dir,
                      contents=self.file2_contents)

    self.assertEqual(1, len(self.response.artifacts))
    for artifact in self.response.artifacts:
      self._path_checks(artifact.path, self.dest_dir,
                        contents=self.file3_contents)

  def test_multiple_directories(self):
    """Test multiple directories."""
    self.response.artifact.path = self.source_dir
    self.response.artifact.location = common_pb2.Path.INSIDE
    self.response.nested_artifact.path.path = self.source_dir2
    self.response.nested_artifact.path.location = common_pb2.Path.INSIDE

    field_handler.extract_results(self.request, self.response, self.chroot)

    self._path_checks(self.response.artifact.path, self.dest_dir)
    self._path_checks(self.response.nested_artifact.path.path, self.dest_dir)

    expected = os.listdir(self.chroot_source)
    expected.extend(os.listdir(self.chroot_source2))
    self.assertCountEqual(expected, os.listdir(self.response.artifact.path))