recipe_modules/file/api.py

# Copyright 2017 The LUCI Authors. All rights reserved.
# Use of this source code is governed under the Apache License, Version 2.0
# that can be found in the LICENSE file.

"""File manipulation (read/write/delete/glob) methods."""

from recipe_engine import recipe_api


import os
import fnmatch

# TODO(iannucci): Introduce the concept of a 'native step' and implement these
# directly in the current python interpreter without the need for a subprocess
# invocation.

class FileApi(recipe_api.RecipeApi):
  class Error(recipe_api.StepFailure):
    """Error is an InfraFailure, except that it also contains an errno field
    indicating the errno name (i.e. 'EEXIST') of the underlying error.
    """
    def __init__(self, step_name, errno_name, message):
      reason = 'Step(%r) failed %r with: %s' % (step_name, errno_name, message)
      super(FileApi.Error, self).__init__(reason)
      self.errno_name = errno_name

  def _assert_absolute_path_or_placeholder(self, path_or_placeholder):
    if isinstance(path_or_placeholder, recipe_api.Placeholder):
      # We assume that all Placeholder classes will render to an absolute path,
      # as this is part of their api contract.
      return True
    return self.m.path.assert_absolute(path_or_placeholder)

  def _run(self, name, args, step_test_data=None, stdout=None):
    if not step_test_data:
      step_test_data = self.test_api.errno
    args = [
      '--json-output', self.m.json.output(add_json_log=False)
    ] + args
    result = self.m.python(
      name, self.resource('fileutil.py'), args=args,
      step_test_data=step_test_data, stdout=stdout,
      infra_step=True)
    j = result.json.output
    if not j['ok']:
      result.presentation.status = self.m.step.FAILURE
      result.presentation.step_text = j['message']
      # pylint thinks this isn't a standard exception... silly pylint.
      # pylint: disable=nonstandard-exception
      raise self.Error(name, j['errno_name'], j['message'])
    return result

  def copy(self, name, source, dest):
    """Copies a file (including mode bits) from source to destination on the
    local filesystem.

    Behaves identically to shutil.copy.

    Args:
      * name (str) - The name of the step.
      * source (Path|Placeholder) - The path to the file you want to copy.
      * dest (Path|Placeholder) - The path to the destination file name. If this
        path exists and is a directory, the basename of `source` will be
        appended to derive a path to a destination file.

    Raises file.Error
    """
    self._assert_absolute_path_or_placeholder(source)
    self._assert_absolute_path_or_placeholder(dest)
    self._run(name, ['copy', source, dest])
    self.m.path.mock_copy_paths(source, dest)

  def copytree(self, name, source, dest, symlinks=False):
    """Recursively copies a directory tree.

    Behaves identically to shutil.copytree.
    `dest` must not exist.

    Args:
      * name (str) - The name of the step.
      * source (Path) - The path of the directory to copy.
      * dest (Path) - The place where you want the recursive copy to show up. This
        must not already exist.
      * symlinks (bool) - Preserve symlinks. No effect on Windows.

    Raises file.Error
    """
    self.m.path.assert_absolute(source)
    self.m.path.assert_absolute(dest)
    args = ['--symlinks'] if symlinks else []
    self._run(name, ['copytree'] + args + [source, dest])
    self.m.path.mock_copy_paths(source, dest)

  def move(self, name, source, dest):
    """Moves a file or directory.

    Behaves identically to shutil.move.

    Args:
      * name (str) - The name of the step.
      * source (Path) - The path of the item to move.
      * dest (Path) - The new name of the item.

    Raises file.Error
    """
    self.m.path.assert_absolute(source)
    self.m.path.assert_absolute(dest)
    self._run(name, ['move', source, dest])
    self.m.path.mock_copy_paths(source, dest)
    self.m.path.mock_remove_paths(source)

  def read_raw(self, name, source, test_data=''):
    """Reads a file as raw data.

    Args:
      * name (str) - The name of the step.
      * source (Path) - The path of the file to read.
      * test_data (str) - Some default data for this step to return when running
        under simulation.

    Returns (str) - The unencoded (binary) contents of the file.

    Raises file.Error
    """
    self.m.path.assert_absolute(source)
    step_test_data = lambda: self.test_api.read_raw(test_data)
    result = self._run(name, ['copy', source, self.m.raw_io.output()],
                       step_test_data=step_test_data)
    return result.raw_io.output

  def write_raw(self, name, dest, data):
    """Write the given `data` to `dest`.

    Args:
      * name (str) - The name of the step.
      * dest (Path) - The path of the file to write.
      * data (str) - The data to write.

    Raises file.Error.
    """
    self.m.path.assert_absolute(dest)
    self._run(name, ['copy', self.m.raw_io.input(data), dest])
    self.m.path.mock_add_paths(dest)

  def read_text(self, name, source, test_data=''):
    """Reads a file as UTF-8 encoded text.

    Args:
      * name (str) - The name of the step.
      * source (Path) - The path of the file to read.
      * test_data (str) - Some default data for this step to return when running
        under simulation.

    Returns (str) - The content of the file.

    Raises file.Error
    """
    self.m.path.assert_absolute(source)
    step_test_data = lambda: self.test_api.read_text(test_data)
    result = self._run(name, ['copy', source, self.m.raw_io.output_text()],
                       step_test_data=step_test_data)
    return result.raw_io.output_text

  def write_text(self, name, dest, text_data):
    """Write the given UTF-8 encoded `text_data` to `dest`.

    Args:
      * name (str) - The name of the step.
      * dest (Path) - The path of the file to write.
      * text_data (str) - The UTF-8 encoded data to write.

    Raises file.Error.
    """
    self.m.path.assert_absolute(dest)
    self._run(name, ['copy', self.m.raw_io.input_text(text_data), dest])
    self.m.path.mock_add_paths(dest)

  def glob_paths(self, name, source, pattern, test_data=()):
    """Performs glob expansion on `pattern`.

    glob rules for `pattern` follow the same syntax as for the python `glob`
    stdlib module.

    Args:
      * name (str) - The name of the step.
      * source (Path) - The directory whose contents should be globbed.
      * pattern (str) - The glob pattern to apply under `source`.
      * test_data (iterable[str]) - Some default data for this step to return when
        running under simulation. This should be the list of file items found
        in this directory.

    Returns list[Path] - All paths found.

    Raises file.Error.
    """
    result = self._run(
      name, ['glob', source, pattern],
      lambda: self.test_api.glob_paths(test_data),
      self.m.raw_io.output_text())
    ret = [source.join(*x.split(self.m.path.sep))
           for x in result.stdout.splitlines()]
    result.presentation.logs["glob"] = map(str, ret)
    return ret

  def remove(self, name, source):
    """Remove a file.

    Does not raise Error if the file doesn't exist.

    Args:
      * name (str) - The name of the step.
      * source (Path) - The file to remove.

    Raises file.Error.
    """
    self.m.path.assert_absolute(source)
    self._run(name, ['remove', source])
    self.m.path.mock_remove_paths(source)

  def listdir(self, name, source, test_data=()):
    """List all files inside a directory.

    Args:
      * name (str) - The name of the step.
      * source (Path) - The directory to list.
      * test_data (iterable[str]) - Some default data for this step to return when
        running under simulation. This should be the list of file items found
        in this directory.

    Returns list[Path]

    Raises file.Error.
    """
    self.m.path.assert_absolute(source)
    result = self._run(
      name, ['listdir', source],
      lambda: self.test_api.listdir(test_data),
      self.m.raw_io.output_text())
    ret = [source.join(x) for x in result.stdout.splitlines()]
    result.presentation.logs['listdir'] = map(str, ret)
    return ret

  def ensure_directory(self, name, dest, mode=0777):
    """Ensures that `dest` exists and is a directory.

    Args:
      * name (str) - The name of the step.
      * dest (Path) - The directory to ensure.
      * mode (int) - The mode to use if the directory doesn't exist. This method
        does not ensure the mode if the directory already exists (if you need
        that behaviour, file a bug).

    Raises file.Error if the path exists but is not a directory.
    """
    self.m.path.assert_absolute(dest)
    self._run(
      name, ['ensure-directory', '--mode', oct(mode), dest])
    self.m.path.mock_add_paths(dest)

  def filesizes(self, name, files, test_data=None):
    """Returns list of filesizes for the given files.

    Args:
      * name (str) - The name of the step.
      * files (list[Path]) - Paths to files.

    Returns list[int], size of each file in bytes.
    """
    if test_data is None:
      test_data = [111 * (i+1) + (i % 3 - 2) * i for i, _ in enumerate(files)]
    for f in files:
      self.m.path.assert_absolute(f)
    result = self._run(
      name, ['filesizes'] + list(files),
      lambda: self.test_api.filesizes(test_data),
      self.m.raw_io.output_text())
    ret = map(int, result.stdout.strip().splitlines())
    result.presentation.logs['filesizes'] = ['%s: \t%d' % fs
                                             for fs in zip(files, ret)]
    return ret

  def rmtree(self, name, source):
    """Recursively removes a directory.

    This uses a native python on Linux/Mac, and uses `rd` on Windows to avoid
    issues w.r.t. path lengths and read-only attributes. If the directory is
    gone already, this returns without error.

    Args:
      * name (str) - The name of the step.
      * source (Path) - The directory to remove.

    Raises file.Error.
    """
    self.m.path.assert_absolute(source)
    self._run(name, ['rmtree', source])
    self.m.path.mock_remove_paths(str(source))

  def rmcontents(self, name, source):
    """Similar to rmtree, but removes only contents not the directory.

    This is useful e.g. when removing contents of current working directory.
    Deleting current working directory makes all further getcwd calls fail
    until chdir is called. chdir would be tricky in recipes, so we provide
    a call that doesn't delete the directory itself.

    Args:
      * name (str) - The name of the step.
      * source (Path) - The directory whose contents should be removed.

    Raises file.Error.
    """
    self.m.path.assert_absolute(source)
    self._run(name, ['rmcontents', source])
    self.m.path.mock_remove_paths(str(source)+self.m.path.sep)

  def rmglob(self, name, source, pattern):
    """Removes all entries in `source` matching the glob `pattern`.

    Args:
      * name (str) - The name of the step.
      * source (Path) - The directory whose contents should be filtered and
        removed.
      * pattern (str) - The glob pattern to apply under `source`. Anything
        matching this pattern will be removed.

    Raises file.Error.
    """
    self.m.path.assert_absolute(source)
    self._run(name, ['rmglob', source, pattern])

    src = str(source)
    def filt(p):
      assert p.startswith(src), (src, p)
      return fnmatch.fnmatch(p[len(src)+1:].split(os.path.sep)[0], pattern)
    self.m.path.mock_remove_paths(str(source), filt)

  def symlink(self, name, source, link):
    """Creates a symlink from link to source on the local filesystem.

    Behaves identically to os.symlink.

    Args:
      * name (str) - The name of the step.
      * source (Path|Placeholder) - The path to link to.
      * link (Path|Placeholder) - The link to create.

    Raises file.Error
    """
    self._assert_absolute_path_or_placeholder(source)
    self._assert_absolute_path_or_placeholder(link)
    self._run(name, ['symlink', source, link])
    self.m.path.mock_copy_paths(source, link)