OperatingSystem.py

Go to the documentation of this file.

#  Copyright 2008-2015 Nokia Networks
#  Copyright 2016-     Robot Framework Foundation
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
 
import fnmatch
import glob
import io
import os
import shutil
import sys
import tempfile
import time
 
from robotide.lib.robot.version import get_version
from robotide.lib.robot.api import logger
from robotide.lib.robot.utils import (abspath, ConnectionCache, console_decode, del_env_var,
                         get_env_var, get_env_vars, get_time, is_truthy,
                         is_unicode, normpath, parse_time, plural_or_not,
                         secs_to_timestamp, secs_to_timestr, seq2str,
                         set_env_var, timestr_to_secs, unic, CONSOLE_ENCODING,
                         IRONPYTHON, JYTHON, PY2, PY3, SYSTEM_ENCODING, WINDOWS)
 
__version__ = get_version()
PROCESSES = ConnectionCache('No active processes.')
 
 
 
class OperatingSystem():
    ROBOT_LIBRARY_SCOPE = 'GLOBAL'
    ROBOT_LIBRARY_VERSION = __version__
 
    
    def run(self, command):
        return self._run_run(command)[1]
 
    
    def run_and_return_rc(self, command):
        return self._run_run(command)[0]
 
    
    def run_and_return_rc_and_output(self, command):
        return self._run_run(command)
 
    def _run(self, command):
        process = _Process(command)
        self._info_info("Running command '%s'." % process)
        stdout = process.read()
        rc = process.close()
        return rc, stdout
 
    
    def get_file(self, path, encoding='UTF-8', encoding_errors='strict'):
        path = self._absnorm_absnorm(path)
        self._link_link("Getting file '%s'.", path)
        encoding = self._map_encoding_map_encoding(encoding)
        if IRONPYTHON:
            # https://github.com/IronLanguages/main/issues/1233
            with open(path) as f:
                content = f.read().decode(encoding, encoding_errors)
        else:
            with io.open(path, encoding=encoding, errors=encoding_errors,
                         newline='') as f:
                content = f.read()
        return content.replace('\r\n', '\n')
 
    def _map_encoding(self, encoding):
        # Python 3 opens files in native system encoding by default.
        if PY3 and encoding.upper() == 'SYSTEM':
            return None
        return {'SYSTEM': SYSTEM_ENCODING,
                'CONSOLE': CONSOLE_ENCODING}.get(encoding.upper(), encoding)
 
    
    def get_binary_file(self, path):
        path = self._absnorm_absnorm(path)
        self._link_link("Getting file '%s'.", path)
        with open(path, 'rb') as f:
            return bytes(f.read())
 
    
    def grep_file(self, path, pattern, encoding='UTF-8', encoding_errors='strict'):
        pattern = '*%s*' % pattern
        path = self._absnorm_absnorm(path)
        lines = []
        total_lines = 0
        self._link_link("Reading file '%s'.", path)
        with io.open(path, encoding=encoding, errors=encoding_errors) as f:
            for line in f.readlines():
                total_lines += 1
                line = line.rstrip('\r\n')
                if fnmatch.fnmatchcase(line, pattern):
                    lines.append(line)
            self._info_info('%d out of %d lines matched' % (len(lines), total_lines))
            return '\n'.join(lines)
 
    
    def log_file(self, path, encoding='UTF-8', encoding_errors='strict'):
        content = self.get_fileget_file(path, encoding, encoding_errors)
        self._info_info(content)
        return content
 
    # File and directory existence
 
    
    def should_exist(self, path, msg=None):
        path = self._absnorm_absnorm(path)
        if not self._glob_glob(path):
            self._fail_fail(msg, "Path '%s' does not exist." % path)
        self._link_link("Path '%s' exists.", path)
 
    
    def should_not_exist(self, path, msg=None):
        path = self._absnorm_absnorm(path)
        matches = self._glob_glob(path)
        if matches:
            self._fail_fail(msg, self._get_matches_error_get_matches_error('Path', path, matches))
        self._link_link("Path '%s' does not exist.", path)
 
    def _glob(self, path):
        return glob.glob(path) if not os.path.exists(path) else [path]
 
    def _get_matches_error(self, what, path, matches):
        if not self._is_glob_path_is_glob_path(path):
            return "%s '%s' exists." % (what, path)
        return "%s '%s' matches %s." % (what, path, seq2str(sorted(matches)))
 
    def _is_glob_path(self, path):
        return '*' in path or '?' in path or ('[' in path and ']' in path)
 
    
    def file_should_exist(self, path, msg=None):
        path = self._absnorm_absnorm(path)
        matches = [p for p in self._glob_glob(path) if os.path.isfile(p)]
        if not matches:
            self._fail_fail(msg, "File '%s' does not exist." % path)
        self._link_link("File '%s' exists.", path)
 
    
    def file_should_not_exist(self, path, msg=None):
        path = self._absnorm_absnorm(path)
        matches = [p for p in self._glob_glob(path) if os.path.isfile(p)]
        if matches:
            self._fail_fail(msg, self._get_matches_error_get_matches_error('File', path, matches))
        self._link_link("File '%s' does not exist.", path)
 
    
    def directory_should_exist(self, path, msg=None):
        path = self._absnorm_absnorm(path)
        matches = [p for p in self._glob_glob(path) if os.path.isdir(p)]
        if not matches:
            self._fail_fail(msg, "Directory '%s' does not exist." % path)
        self._link_link("Directory '%s' exists.", path)
 
    
    def directory_should_not_exist(self, path, msg=None):
        path = self._absnorm_absnorm(path)
        matches = [p for p in self._glob_glob(path) if os.path.isdir(p)]
        if matches:
            self._fail_fail(msg, self._get_matches_error_get_matches_error('Directory', path, matches))
        self._link_link("Directory '%s' does not exist.", path)
 
    # Waiting file/dir to appear/disappear
 
    
    def wait_until_removed(self, path, timeout='1 minute'):
        path = self._absnorm_absnorm(path)
        timeout = timestr_to_secs(timeout)
        maxtime = time.time() + timeout
        while self._glob_glob(path):
            if timeout >= 0 and time.time() > maxtime:
                self._fail_fail("'%s' was not removed in %s."
                           % (path, secs_to_timestr(timeout)))
            time.sleep(0.1)
        self._link_link("'%s' was removed.", path)
 
    
    def wait_until_created(self, path, timeout='1 minute'):
        path = self._absnorm_absnorm(path)
        timeout = timestr_to_secs(timeout)
        maxtime = time.time() + timeout
        while not self._glob_glob(path):
            if timeout >= 0 and time.time() > maxtime:
                self._fail_fail("'%s' was not created in %s."
                           % (path, secs_to_timestr(timeout)))
            time.sleep(0.1)
        self._link_link("'%s' was created.", path)
 
    # Dir/file empty
 
    
    def directory_should_be_empty(self, path, msg=None):
        path = self._absnorm_absnorm(path)
        items = self._list_dir_list_dir(path)
        if items:
            self._fail_fail(msg, "Directory '%s' is not empty. Contents: %s."
                            % (path, seq2str(items, lastsep=', ')))
        self._link_link("Directory '%s' is empty.", path)
 
    
    def directory_should_not_be_empty(self, path, msg=None):
        path = self._absnorm_absnorm(path)
        items = self._list_dir_list_dir(path)
        if not items:
            self._fail_fail(msg, "Directory '%s' is empty." % path)
        self._link_link("Directory '%%s' contains %d item%s."
                   % (len(items), plural_or_not(items)), path)
 
    
    def file_should_be_empty(self, path, msg=None):
        path = self._absnorm_absnorm(path)
        if not os.path.isfile(path):
            self._error_error("File '%s' does not exist." % path)
        size = os.stat(path).st_size
        if size > 0:
            self._fail_fail(msg,
                       "File '%s' is not empty. Size: %d bytes." % (path, size))
        self._link_link("File '%s' is empty.", path)
 
    
    def file_should_not_be_empty(self, path, msg=None):
        path = self._absnorm_absnorm(path)
        if not os.path.isfile(path):
            self._error_error("File '%s' does not exist." % path)
        size = os.stat(path).st_size
        if size == 0:
            self._fail_fail(msg, "File '%s' is empty." % path)
        self._link_link("File '%%s' contains %d bytes." % size, path)
 
    # Creating and removing files and directory
 
    
    def create_file(self, path, content='', encoding='UTF-8'):
        path = self._write_to_file_write_to_file(path, content, encoding)
        self._link_link("Created file '%s'.", path)
 
    def _write_to_file(self, path, content, encoding=None, mode='w'):
        path = self._absnorm_absnorm(path)
        parent = os.path.dirname(path)
        if not os.path.exists(parent):
            os.makedirs(parent)
        # io.open() only accepts Unicode, not byte-strings, in text mode.
        # We expect possible byte-strings to be all ASCII.
        if PY2 and isinstance(content, str) and 'b' not in mode:
            content = unicode(content)
        if encoding:
            encoding = self._map_encoding_map_encoding(encoding)
        with io.open(path, mode, encoding=encoding) as f:
            f.write(content)
        return path
 
    
    def create_binary_file(self, path, content):
        if is_unicode(content):
            content = bytes(bytearray(ord(c) for c in content))
        path = self._write_to_file_write_to_file(path, content, mode='wb')
        self._link_link("Created binary file '%s'.", path)
 
    
    def append_to_file(self, path, content, encoding='UTF-8'):
        path = self._write_to_file_write_to_file(path, content, encoding, mode='a')
        self._link_link("Appended to file '%s'.", path)
 
    
    def remove_file(self, path):
        path = self._absnorm_absnorm(path)
        matches = self._glob_glob(path)
        if not matches:
            self._link_link("File '%s' does not exist.", path)
        for match in matches:
            if not os.path.isfile(match):
                self._error_error("Path '%s' is not a file." % match)
            os.remove(match)
            self._link_link("Removed file '%s'.", match)
 
    
    def remove_files(self, *paths):
        for path in paths:
            self.remove_fileremove_file(path)
 
    
    def empty_directory(self, path):
        path = self._absnorm_absnorm(path)
        for item in self._list_dir_list_dir(path, absolute=True):
            if os.path.isdir(item):
                shutil.rmtree(item)
            else:
                os.remove(item)
        self._link_link("Emptied directory '%s'.", path)
 
    
    def create_directory(self, path):
        path = self._absnorm_absnorm(path)
        if os.path.isdir(path):
            self._link_link("Directory '%s' already exists.", path )
        elif os.path.exists(path):
            self._error_error("Path '%s' is not a directory." % path)
        else:
            os.makedirs(path)
            self._link_link("Created directory '%s'.", path)
 
    
    def remove_directory(self, path, recursive=False):
        path = self._absnorm_absnorm(path)
        if not os.path.exists(path):
            self._link_link("Directory '%s' does not exist.", path)
        elif not os.path.isdir(path):
            self._error_error("Path '%s' is not a directory." % path)
        else:
            if is_truthy(recursive):
                shutil.rmtree(path)
            else:
                self.directory_should_be_emptydirectory_should_be_empty(
                    path, "Directory '%s' is not empty." % path)
                os.rmdir(path)
            self._link_link("Removed directory '%s'.", path)
 
    # Moving and copying files and directories
 
    
    def copy_file(self, source, destination):
        source, destination = \
            self._prepare_copy_and_move_file_prepare_copy_and_move_file(source, destination)
        if not self._are_source_and_destination_same_file_are_source_and_destination_same_file(source, destination):
            source, destination = self._atomic_copy_atomic_copy(source, destination)
            self._link_link("Copied file from '%s' to '%s'.", source, destination)
        return destination
 
    def _prepare_copy_and_move_file(self, source, destination):
        source = self._normalize_copy_and_move_source_normalize_copy_and_move_source(source)
        destination = self._normalize_copy_and_move_destination_normalize_copy_and_move_destination(destination)
        if os.path.isdir(destination):
            destination = os.path.join(destination, os.path.basename(source))
        return source, destination
 
    def _normalize_copy_and_move_source(self, source):
        source = self._absnorm_absnorm(source)
        sources = self._glob_glob(source)
        if len(sources) > 1:
            self._error_error("Multiple matches with source pattern '%s'." % source)
        if sources:
            source = sources[0]
        if not os.path.exists(source):
            self._error_error("Source file '%s' does not exist." % source)
        if not os.path.isfile(source):
            self._error_error("Source file '%s' is not a regular file." % source)
        return source
 
    def _normalize_copy_and_move_destination(self, destination):
        is_dir = os.path.isdir(destination) or destination.endswith(('/', '\\'))
        destination = self._absnorm_absnorm(destination)
        directory = destination if is_dir else os.path.dirname(destination)
        self._ensure_destination_directory_exists_ensure_destination_directory_exists(directory)
        return destination
 
    def _ensure_destination_directory_exists(self, path):
        if not os.path.exists(path):
            os.makedirs(path)
        elif not os.path.isdir(path):
            self._error_error("Destination '%s' exists and is not a directory." % path)
 
    def _are_source_and_destination_same_file(self, source, destination):
        if self._force_normalize_force_normalize(source) == self._force_normalize_force_normalize(destination):
            self._link_link("Source '%s' and destination '%s' point to the same "
                       "file.", source, destination)
            return True
        return False
 
    def _force_normalize(self, path):
        # TODO: Should normalize_path also support link normalization?
        # TODO: Should we handle dos paths like 'exampl~1.txt'?
        return os.path.realpath(normpath(path, case_normalize=True))
 
    
    def _atomic_copy(self, source, destination):
        temp_directory = tempfile.mkdtemp(dir=os.path.dirname(destination))
        temp_file = os.path.join(temp_directory, os.path.basename(source))
        try:
            shutil.copy(source, temp_file)
            if os.path.exists(destination):
                os.remove(destination)
            shutil.move(temp_file, destination)
        finally:
            shutil.rmtree(temp_directory)
        return source, destination
 
    
    def move_file(self, source, destination):
        source, destination = \
            self._prepare_copy_and_move_file_prepare_copy_and_move_file(source, destination)
        if not self._are_source_and_destination_same_file_are_source_and_destination_same_file(destination, source):
            shutil.move(source, destination)
            self._link_link("Moved file from '%s' to '%s'.", source, destination)
        return destination
 
    
    def copy_files(self, *sources_and_destination):
        sources, destination \
            = self._prepare_copy_and_move_files_prepare_copy_and_move_files(sources_and_destination)
        for source in sources:
            self.copy_filecopy_file(source, destination)
 
    def _prepare_copy_and_move_files(self, items):
        if len(items) < 2:
            self._error_error('Must contain destination and at least one source.')
        sources = self._glob_files_glob_files(items[:-1])
        destination = self._absnorm_absnorm(items[-1])
        self._ensure_destination_directory_exists_ensure_destination_directory_exists(destination)
        return sources, destination
 
    def _glob_files(self, patterns):
        files = []
        for pattern in patterns:
            files.extend(self._glob_glob(self._absnorm_absnorm(pattern)))
        return files
 
    
    def move_files(self, *sources_and_destination):
        sources, destination \
            = self._prepare_copy_and_move_files_prepare_copy_and_move_files(sources_and_destination)
        for source in sources:
            self.move_filemove_file(source, destination)
 
    
    def copy_directory(self, source, destination):
        source, destination \
            = self._prepare_copy_and_move_directory_prepare_copy_and_move_directory(source, destination)
        try:
            shutil.copytree(source, destination)
        except shutil.Error:
            # https://github.com/robotframework/robotframework/issues/2321
            if not (WINDOWS and JYTHON):
                raise
        self._link_link("Copied directory from '%s' to '%s'.", source, destination)
 
    def _prepare_copy_and_move_directory(self, source, destination):
        source = self._absnorm_absnorm(source)
        destination = self._absnorm_absnorm(destination)
        if not os.path.exists(source):
            self._error_error("Source '%s' does not exist." % source)
        if not os.path.isdir(source):
            self._error_error("Source '%s' is not a directory." % source)
        if os.path.exists(destination) and not os.path.isdir(destination):
            self._error_error("Destination '%s' is not a directory." % destination)
        if os.path.exists(destination):
            base = os.path.basename(source)
            destination = os.path.join(destination, base)
        else:
            parent = os.path.dirname(destination)
            if not os.path.exists(parent):
                os.makedirs(parent)
        return source, destination
 
    
    def move_directory(self, source, destination):
        source, destination \
            = self._prepare_copy_and_move_directory_prepare_copy_and_move_directory(source, destination)
        shutil.move(source, destination)
        self._link_link("Moved directory from '%s' to '%s'.", source, destination)
 
    # Environment Variables
 
    
    def get_environment_variable(self, name, default=None):
        value = get_env_var(name, default)
        if value is None:
            self._error_error("Environment variable '%s' does not exist." % name)
        return value
 
    
    def set_environment_variable(self, name, value):
        set_env_var(name, value)
        self._info_info("Environment variable '%s' set to value '%s'."
                   % (name, value))
 
    
    def append_to_environment_variable(self, name, *values, **config):
        sentinel = object()
        initial = self.get_environment_variableget_environment_variable(name, sentinel)
        if initial is not sentinel:
            values = (initial,) + values
        separator = config.pop('separator', os.pathsep)
        if config:
            config = ['='.join(i) for i in sorted(config.items())]
            self._error_error('Configuration %s not accepted.'
                        % seq2str(config, lastsep=' or '))
        self.set_environment_variableset_environment_variable(name, separator.join(values))
 
    
    def remove_environment_variable(self, *names):
        for name in names:
            value = del_env_var(name)
            if value:
                self._info_info("Environment variable '%s' deleted." % name)
            else:
                self._info_info("Environment variable '%s' does not exist." % name)
 
    
    def environment_variable_should_be_set(self, name, msg=None):
        value = get_env_var(name)
        if not value:
            self._fail_fail(msg, "Environment variable '%s' is not set." % name)
        self._info_info("Environment variable '%s' is set to '%s'." % (name, value))
 
    
    def environment_variable_should_not_be_set(self, name, msg=None):
        value = get_env_var(name)
        if value:
            self._fail_fail(msg, "Environment variable '%s' is set to '%s'."
                            % (name, value))
        self._info_info("Environment variable '%s' is not set." % name)
 
    
    def get_environment_variables(self):
        return get_env_vars()
 
    
    def log_environment_variables(self, level='INFO'):
        variables = get_env_vars()
        for name in sorted(variables, key=lambda item: item.lower()):
            self._log_log('%s = %s' % (name, variables[name]), level)
        return variables
 
    # Path
 
    
    def join_path(self, base, *parts):
        base = base.replace('/', os.sep)
        parts = [p.replace('/', os.sep) for p in parts]
        return self.normalize_pathnormalize_path(os.path.join(base, *parts))
 
    
    def join_paths(self, base, *paths):
        return [self.join_pathjoin_path(base, path) for path in paths]
 
    
    def normalize_path(self, path, case_normalize=False):
        path = os.path.normpath(os.path.expanduser(path.replace('/', os.sep)))
        # os.path.normcase doesn't normalize on OSX which also, by default,
        # has case-insensitive file system. Our robot.utils.normpath would
        # do that, but it's not certain would that, or other things that the
        # utility do, desirable.
        if case_normalize:
            path = os.path.normcase(path)
        return path or '.'
 
    
    def split_path(self, path):
        return os.path.split(self.normalize_pathnormalize_path(path))
 
    
    def split_extension(self, path):
        path = self.normalize_pathnormalize_path(path)
        basename = os.path.basename(path)
        if basename.startswith('.' * basename.count('.')):
            return path, ''
        if path.endswith('.'):
            path2 = path.rstrip('.')
            trailing_dots = '.' * (len(path) - len(path2))
            path = path2
        else:
            trailing_dots = ''
        basepath, extension = os.path.splitext(path)
        if extension.startswith('.'):
            extension = extension[1:]
        if extension:
            extension += trailing_dots
        else:
            basepath += trailing_dots
        return basepath, extension
 
    # Misc
 
    
    def get_modified_time(self, path, format='timestamp'):
        path = self._absnorm_absnorm(path)
        if not os.path.exists(path):
            self._error_error("Path '%s' does not exist." % path)
        mtime = get_time(format, os.stat(path).st_mtime)
        self._link_link("Last modified time of '%%s' is %s." % mtime, path)
        return mtime
 
    
    def set_modified_time(self, path, mtime):
        mtime = parse_time(mtime)
        path = self._absnorm_absnorm(path)
        if not os.path.exists(path):
            self._error_error("File '%s' does not exist." % path)
        if not os.path.isfile(path):
            self._error_error("Path '%s' is not a regular file." % path)
        os.utime(path, (mtime, mtime))
        time.sleep(0.1)  # Give os some time to really set these times
        tstamp = secs_to_timestamp(mtime, seps=('-', ' ', ':'))
        self._link_link("Set modified time of '%%s' to %s." % tstamp, path)
 
    
    def get_file_size(self, path):
        path = self._absnorm_absnorm(path)
        if not os.path.isfile(path):
            self._error_error("File '%s' does not exist." % path)
        size = os.stat(path).st_size
        plural = plural_or_not(size)
        self._link_link("Size of file '%%s' is %d byte%s." % (size, plural), path)
        return size
 
    
    def list_directory(self, path, pattern=None, absolute=False):
        items = self._list_dir_list_dir(path, pattern, absolute)
        self._info_info('%d item%s:\n%s' % (len(items), plural_or_not(items),
                                       '\n'.join(items)))
        return items
 
    
    def list_files_in_directory(self, path, pattern=None, absolute=False):
        files = self._list_files_in_dir_list_files_in_dir(path, pattern, absolute)
        self._info_info('%d file%s:\n%s' % (len(files), plural_or_not(files),
                                       '\n'.join(files)))
        return files
 
    
    def list_directories_in_directory(self, path, pattern=None, absolute=False):
        dirs = self._list_dirs_in_dir_list_dirs_in_dir(path, pattern, absolute)
        self._info_info('%d director%s:\n%s' % (len(dirs),
                                           'y' if len(dirs) == 1 else 'ies',
                                           '\n'.join(dirs)))
        return dirs
 
    
    def count_items_in_directory(self, path, pattern=None):
        count = len(self._list_dir_list_dir(path, pattern))
        self._info_info("%s item%s." % (count, plural_or_not(count)))
        return count
 
    
    def count_files_in_directory(self, path, pattern=None):
        count = len(self._list_files_in_dir_list_files_in_dir(path, pattern))
        self._info_info("%s file%s." % (count, plural_or_not(count)))
        return count
 
    
    def count_directories_in_directory(self, path, pattern=None):
        count = len(self._list_dirs_in_dir_list_dirs_in_dir(path, pattern))
        self._info_info("%s director%s." % (count, 'y' if count == 1 else 'ies'))
        return count
 
    def _list_dir(self, path, pattern=None, absolute=False):
        path = self._absnorm_absnorm(path)
        self._link_link("Listing contents of directory '%s'.", path)
        if not os.path.isdir(path):
            self._error_error("Directory '%s' does not exist." % path)
        # result is already unicode but unic also handles NFC normalization
        items = sorted(unic(item) for item in os.listdir(path))
        if pattern:
            items = [i for i in items if fnmatch.fnmatchcase(i, pattern)]
        if is_truthy(absolute):
            path = os.path.normpath(path)
            items = [os.path.join(path, item) for item in items]
        return items
 
    def _list_files_in_dir(self, path, pattern=None, absolute=False):
        return [item for item in self._list_dir_list_dir(path, pattern, absolute)
                if os.path.isfile(os.path.join(path, item))]
 
    def _list_dirs_in_dir(self, path, pattern=None, absolute=False):
        return [item for item in self._list_dir_list_dir(path, pattern, absolute)
                if os.path.isdir(os.path.join(path, item))]
 
    
    def touch(self, path):
        path = self._absnorm_absnorm(path)
        if os.path.isdir(path):
            self._error_error("Cannot touch '%s' because it is a directory." % path)
        if not os.path.exists(os.path.dirname(path)):
            self._error_error("Cannot touch '%s' because its parent directory does "
                        "not exist." % path)
        if os.path.exists(path):
            mtime = round(time.time())
            os.utime(path, (mtime, mtime))
            self._link_link("Touched existing file '%s'.", path)
        else:
            open(path, 'w').close()
            self._link_link("Touched new file '%s'.", path)
 
    def _absnorm(self, path):
        path = self.normalize_pathnormalize_path(path)
        try:
            return abspath(path)
        except ValueError:  # http://ironpython.codeplex.com/workitem/29489
            return path
 
    def _fail(self, *messages):
        raise AssertionError(next(msg for msg in messages if msg))
 
    def _error(self, msg):
        raise RuntimeError(msg)
 
    def _info(self, msg):
        self._log_log(msg, 'INFO')
 
    def _link(self, msg, *paths):
        paths = tuple('<a href="file://%s">%s</a>' % (p, p) for p in paths)
        self._log_log(msg % paths, 'HTML')
 
    def _warn(self, msg):
        self._log_log(msg, 'WARN')
 
    def _log(self, msg, level):
        logger.write(msg, level)
 
 
class _Process:
 
    def __init__(self, command):
        self._command_command = self._process_command_process_command(command)
        self._process_process = os.popen(self._command_command)
 
    def __str__(self):
        return self._command_command
 
    def read(self):
        return self._process_output_process_output(self._process_process.read())
 
    def close(self):
        try:
            rc = self._process_process.close()
        except IOError:  # Has occurred sometimes in Windows
            return 255
        if rc is None:
            return 0
        # In Windows (Python and Jython) return code is value returned by
        # command (can be almost anything)
        # In other OS:
        #   In Jython return code can be between '-255' - '255'
        #   In Python return code must be converted with 'rc >> 8' and it is
        #   between 0-255 after conversion
        if WINDOWS or JYTHON:
            return rc % 256
        return rc >> 8
 
    def _process_command(self, command):
        if '>' not in command:
            if command.endswith('&'):
                command = command[:-1] + ' 2>&1 &'
            else:
                command += ' 2>&1'
        return self._encode_to_file_system_encode_to_file_system(command)
 
    def _encode_to_file_system(self, string):
        enc = sys.getfilesystemencoding() if PY2 else None
        return string.encode(enc) if enc else string
 
    def _process_output(self, output):
        if '\r\n' in output:
            output = output.replace('\r\n', '\n')
        if output.endswith('\n'):
            output = output[:-1]
        return console_decode(output, force=True)