<commit_before># Automated tests for the `executor' module.
#
# Author: <NAME> <<EMAIL>>
# Last Change: June 27, 2017
# URL: https://executor.readthedocs.io

"""
Automated tests for the `executor` package.

This test suite uses ``sudo`` in several tests. If you don't have passwordless
sudo configured you'll notice because you'll get interactive prompts when
running the test suite ...

Of course the idea behind a test suite is to run non-interactively, so in my
personal development environment I have added a custom sudo configuration file
``/etc/sudoers.d/executor-test-suite`` with the following contents::

    # /etc/sudoers.d/executor-test-suite:
    #
    # Configuration for sudo to run the executor test suite
    # without getting interactive sudo prompts.

    # To enable test_sudo_option.
    peter ALL=NOPASSWD:/bin/chmod 600 /tmp/executor-test-suite/*
    peter ALL=NOPASSWD:/bin/chown root\:root /tmp/executor-test-suite/*
    peter ALL=NOPASSWD:/bin/rm -R /tmp/executor-test-suite
    peter ALL=NOPASSWD:/usr/bin/stat --format=%a /tmp/executor-test-suite/*
    peter ALL=NOPASSWD:/usr/bin/stat --format=%G /tmp/executor-test-suite/*
    peter ALL=NOPASSWD:/usr/bin/stat --format=%U /tmp/executor-test-suite/*

    # To enable test_uid_option and test_user_option. The ALL=(ALL) tokens allow
    # running the command as any user (because the test suite picks user names
    # and user IDs more or less at random).
    peter ALL=(ALL) NOPASSWD:/usr/bin/id *

If you want to use this, make sure you change the username and sanity check the
locations of the executables whose pathnames have been expanded in the sudo
configuration. Happy testing!

By the way none of this is relevant on e.g. Travis CI because in that
environment passwordless sudo access has been configured.
"""

# Standard library modules.
import datetime
import logging
import os
import pwd
import random
import shlex
import socket
import sys
import tempfile
import time
import uuid

# External dependencies.
from humanfriendly import Timer, compact, dedent
from humanfriendly.testing import TemporaryDirectory, TestCase, retry, run_cli
from mock import MagicMock

# Modules included in our package.
from executor import (
    COMMAND_NOT_FOUND_STATUS,
    DEFAULT_SHELL,
    DEFAULT_WORKING_DIRECTORY,
    CommandNotFound,
    ExternalCommand,
    ExternalCommandFailed,
    execute,
    quote,
    which,
)
from executor.cli import main
from executor.concurrent import CommandPool, CommandPoolFailed
from executor.contexts import (
    ChangeRootContext,
    create_context,
    LocalContext,
    RemoteContext,
    SecureChangeRootContext,
)
from executor.process import ProcessTerminationFailed
from executor.chroot import CHROOT_PROGRAM_NAME
from executor.schroot import SCHROOT_PROGRAM_NAME
from executor.ssh.client import (
    DEFAULT_CONNECT_TIMEOUT,
    RemoteCommand,
    RemoteCommandFailed,
    RemoteCommandNotFound,
    RemoteConnectFailed,
    foreach,
    remote,
)
from executor.ssh.server import SSHServer

MISSING_COMMAND = 'a-program-name-that-no-one-would-ever-use'

# Initialize a logger for this module.
logger = logging.getLogger(__name__)


class ExecutorTestCase(TestCase):

    """Container for the `executor` test suite."""

    def setUp(self):
        """Set up logging for subprocesses and initialize test directories."""
        # Set up our superclass.
        super(ExecutorTestCase, self).setUp()
        # Enable verbose logging to the terminal for subprocesses.
        os.environ['COLOREDLOGS_LOG_LEVEL'] = 'DEBUG'
        # Create the directory where superuser privileges are tested.
        self.sudo_enabled_directory = os.path.join(tempfile.gettempdir(), 'executor-test-suite')
        if not os.path.isdir(self.sudo_enabled_directory):
            os.makedirs(self.sudo_enabled_directory)

    def test_graceful_termination(self):
        """Test graceful termination of processes."""
        self.check_termination(method='terminate')

    def test_forceful_termination(self):
        """Test forceful termination of processes."""
        self.check_termination(method='kill')

    def test_graceful_to_forceful_fallback(self):
        """Test that graceful termination falls back to forceful termination."""
        timer = Timer()
        expected_lifetime = 60
        with NonGracefulCommand('sleep', str(expected_lifetime), check=False) as cmd:
            # Request graceful termination even though we know it will fail.
            cmd.terminate(timeout=1)
            # Verify that the process terminated even though our graceful
            # termination request was ignored.
            assert not cmd.is_running
            # Verify that the process actually terminated due to the fall back
            # and not because its expected life time simply ran out.
            assert timer.elapsed_time < expected_lifetime

    def test_process_termination_failure(self):
        """Test handling of forceful termination failures."""
        with NonForcefulCommand('sleep', '60', check=False) as cmd:
            # Request forceful termination even though we know it will fail.
            self.assertRaises(ProcessTerminationFailed, cmd.kill, timeout=1)
            # Verify that the process is indeed still running :-).
            assert cmd.is_running
            # Bypass the overrides to get rid of the process.
            ExternalCommand.terminate_helper(cmd)

    def check_termination(self, method):
        """Helper method for process termination tests."""
        with ExternalCommand('sleep', '60', check=False) as cmd:
            timer = Timer()
            # We use a positive but very low timeout so that all of the code
            # involved gets a chance to run, but without slowing us down.
            getattr(cmd, method)(timeout=0.1)
            # Gotcha: Call wait() so that the process (our own subprocess) is
            # reclaimed because until we do so proc.is_running will be True!
            cmd.wait()
            # Now we can verify our assertions.
            assert not cmd.is_running, "Child still running despite graceful termination request!"
            assert timer.elapsed_time < 10, "It look too long to terminate the child!"

    def test_program_searching(self):
        """Make sure which() works as expected."""
        assert which('python')
        assert not which(MISSING_COMMAND)

    def test_status_code_checking(self):
        """Make sure that status code handling is sane."""
        assert execute('true') is True
        assert execute('false', check=False) is False
        # Make sure execute('false') raises an exception.
        self.assertRaises(ExternalCommandFailed, execute, 'false')
        # Make sure execute('exit 42') raises an exception.
        e = self.assertRaises(ExternalCommandFailed, execute, 'exit 42')
        # Make sure the exception has the expected properties.
        self.assertEqual(e.command.command_line, ['bash', '-c', 'exit 42'])
        self.assertEqual(e.returncode, 42)
        # Make sure the CommandNotFound exception is raised consistently
        # regardless of the values of the `shell' and `async' options.
        for async in True, False:
            for shell in True, False:
                cmd = ExternalCommand(MISSING_COMMAND, async=async, shell=shell)
                self.assertRaises(CommandNotFound, cmd.wait)
                assert cmd.returncode == COMMAND_NOT_FOUND_STATUS

    def test_shell_opt_out(self):
        """Test that callers can always opt out of shell evaluation."""
        # A command consisting of a single string implies shell evaluation but
        # you can change that default.
        assert DEFAULT_SHELL in ExternalCommand('echo 42').command_line
        assert DEFAULT_SHELL not in ExternalCommand('echo 42', shell=False).command_line
        # A command consisting of more than one string implies no shell
        # evaluation but you can change that default.
        assert DEFAULT_SHELL not in ExternalCommand('echo', '42').command_line
        assert DEFAULT_SHELL in ExternalCommand('echo', '42', shell=True).command_line
        # Confirm that opting out of shell evaluation really bypasses all shells.
        cmd = ExternalCommand(
            'echo this will never match an executable name',
            shell=False, check=False,
        )
        cmd.start()
        cmd.wait()
        assert cmd.error_type is CommandNotFound

    def test_commands_on_stdin(self):
        """Test that callers can opt in to shell evaluation for local commands given on standard input."""
        random_string = uuid.uuid4().hex
        output = execute(capture=True, shell=True, input='echo %s' % quote(random_string))
        assert output == random_string

    def test_remote_commands_on_stdin(self):
        """Test that callers can opt in to shell evaluation for remote commands given on standard input."""
        random_string = uuid.uuid4().hex
        with SSHServer() as server:
            output = remote('127.0.0.1',
                            capture=True, shell=True,
                            input='echo %s' % quote(random_string),
                            **server.client_options)
            assert output == random_string

    def test_stdin(self):
        """Make sure standard input can be provided to external commands."""
        assert execute('tr', 'a-z', 'A-Z', input='test', capture=True) == 'TEST'

    def test_stdout(self):
        """Make sure standard output of external commands can be captured."""
        assert execute('echo', 'this is a test', capture=True) == 'this is a test'
        assert execute('echo', '-e', r'line 1\nline 2', capture=True) == 'line 1\nline 2\n'
        # I don't know how to test for the effect of silent=True in a practical
        # way without creating the largest test in this test suite :-). The
        # least I can do is make sure the keyword argument is accepted and the
        # code runs without exceptions in supported environments.
        assert execute('echo', 'this is a test', silent=True) is True

    def test_stderr(self):
        """Make sure standard error of external commands can be captured."""
        stdout_value = 'this goes to standard output'
        stderr_value = 'and this goes to the standard error stream'
        shell_command = 'echo %s; echo %s >&2' % (stdout_value, stderr_value)
        cmd = ExternalCommand(shell_command, capture=True, capture_stderr=True)
        cmd.start()
        assert stdout_value in cmd.decoded_stdout
        assert stderr_value in cmd.decoded_stderr

    def test_merged_streams(self):
        """Make sure standard output/error of external commands can be captured together."""
        stdout_value = 'this goes to standard output'
        stderr_value = 'and this goes to the standard error stream'
        shell_command = 'echo %s; echo %s >&2' % (stdout_value, stderr_value)
        cmd = ExternalCommand(shell_command, capture=True, merge_streams=True)
        cmd.start()
        assert stdout_value in cmd.decoded_stdout
        assert stderr_value in cmd.decoded_stdout
        assert stdout_value not in (cmd.decoded_stderr or '')
        assert stderr_value not in (cmd.decoded_stderr or '')

    def test_stdout_to_file(self):
        """Make sure the standard output stream of external commands can be redirected and appended to a file."""
        fd, filename = tempfile.mkstemp(prefix='executor-', suffix='-stdout.txt')
        with open(filename, 'w') as handle:
            handle.write('existing contents\n')
        with open(filename, 'a') as handle:<commit_msg>Expose more OSError inconsistencies (std{out,err}=None)<commit_after># Automated tests for the `executor' module.
#
# Author: <NAME> <<EMAIL>>
# Last Change: June 27, 2017
# URL: https://executor.readthedocs.io

"""
Automated tests for the `executor` package.

This test suite uses ``sudo`` in several tests. If you don't have passwordless
sudo configured you'll notice because you'll get interactive prompts when
running the test suite ...

Of course the idea behind a test suite is to run non-interactively, so in my
personal development environment I have added a custom sudo configuration file
``/etc/sudoers.d/executor-test-suite`` with the following contents::

    # /etc/sudoers.d/executor-test-suite:
    #
    # Configuration for sudo to run the executor test suite
    # without getting interactive sudo prompts.

    # To enable test_sudo_option.
    peter ALL=NOPASSWD:/bin/chmod 600 /tmp/executor-test-suite/*
    peter ALL=NOPASSWD:/bin/chown root\:root /tmp/executor-test-suite/*
    peter ALL=NOPASSWD:/bin/rm -R /tmp/executor-test-suite
    peter ALL=NOPASSWD:/usr/bin/stat --format=%a /tmp/executor-test-suite/*
    peter ALL=NOPASSWD:/usr/bin/stat --format=%G /tmp/executor-test-suite/*
    peter ALL=NOPASSWD:/usr/bin/stat --format=%U /tmp/executor-test-suite/*

    # To enable test_uid_option and test_user_option. The ALL=(ALL) tokens allow
    # running the command as any user (because the test suite picks user names
    # and user IDs more or less at random).
    peter ALL=(ALL) NOPASSWD:/usr/bin/id *

If you want to use this, make sure you change the username and sanity check the
locations of the executables whose pathnames have been expanded in the sudo
configuration. Happy testing!

By the way none of this is relevant on e.g. Travis CI because in that
environment passwordless sudo access has been configured.
"""

# Standard library modules.
import datetime
import logging
import os
import pwd
import random
import shlex
import socket
import sys
import tempfile
import time
import uuid

# External dependencies.
from humanfriendly import Timer, compact, dedent
from humanfriendly.testing import TemporaryDirectory, TestCase, retry, run_cli
from mock import MagicMock

# Modules included in our package.
from executor import (
    COMMAND_NOT_FOUND_STATUS,
    DEFAULT_SHELL,
    DEFAULT_WORKING_DIRECTORY,
    CommandNotFound,
    ExternalCommand,
    ExternalCommandFailed,
    execute,
    quote,
    which,
)
from executor.cli import main
from executor.concurrent import CommandPool, CommandPoolFailed
from executor.contexts import (
    ChangeRootContext,
    create_context,
    LocalContext,
    RemoteContext,
    SecureChangeRootContext,
)
from executor.process import ProcessTerminationFailed
from executor.chroot import CHROOT_PROGRAM_NAME
from executor.schroot import SCHROOT_PROGRAM_NAME
from executor.ssh.client import (
    DEFAULT_CONNECT_TIMEOUT,
    RemoteCommand,
    RemoteCommandFailed,
    RemoteCommandNotFound,
    RemoteConnectFailed,
    foreach,
    remote,
)
from executor.ssh.server import SSHServer

MISSING_COMMAND = 'a-program-name-that-no-one-would-ever-use'

# Initialize a logger for this module.
logger = logging.getLogger(__name__)


class ExecutorTestCase(TestCase):

    """Container for the `executor` test suite."""

    def setUp(self):
        """Set up logging for subprocesses and initialize test directories."""
        # Set up our superclass.
        super(ExecutorTestCase, self).setUp()
        # Enable verbose logging to the terminal for subprocesses.
        os.environ['COLOREDLOGS_LOG_LEVEL'] = 'DEBUG'
        # Create the directory where superuser privileges are tested.
        self.sudo_enabled_directory = os.path.join(tempfile.gettempdir(), 'executor-test-suite')
        if not os.path.isdir(self.sudo_enabled_directory):
            os.makedirs(self.sudo_enabled_directory)

    def test_graceful_termination(self):
        """Test graceful termination of processes."""
        self.check_termination(method='terminate')

    def test_forceful_termination(self):
        """Test forceful termination of processes."""
        self.check_termination(method='kill')

    def test_graceful_to_forceful_fallback(self):
        """Test that graceful termination falls back to forceful termination."""
        timer = Timer()
        expected_lifetime = 60
        with NonGracefulCommand('sleep', str(expected_lifetime), check=False) as cmd:
            # Request graceful termination even though we know it will fail.
            cmd.terminate(timeout=1)
            # Verify that the process terminated even though our graceful
            # termination request was ignored.
            assert not cmd.is_running
            # Verify that the process actually terminated due to the fall back
            # and not because its expected life time simply ran out.
            assert timer.elapsed_time < expected_lifetime

    def test_process_termination_failure(self):
        """Test handling of forceful termination failures."""
        with NonForcefulCommand('sleep', '60', check=False) as cmd:
            # Request forceful termination even though we know it will fail.
            self.assertRaises(ProcessTerminationFailed, cmd.kill, timeout=1)
            # Verify that the process is indeed still running :-).
            assert cmd.is_running
            # Bypass the overrides to get rid of the process.
            ExternalCommand.terminate_helper(cmd)

    def check_termination(self, method):
        """Helper method for process termination tests."""
        with ExternalCommand('sleep', '60', check=False) as cmd:
            timer = Timer()
            # We use a positive but very low timeout so that all of the code
            # involved gets a chance to run, but without slowing us down.
            getattr(cmd, method)(timeout=0.1)
            # Gotcha: Call wait() so that the process (our own subprocess) is
            # reclaimed because until we do so proc.is_running will be True!
            cmd.wait()
            # Now we can verify our assertions.
            assert not cmd.is_running, "Child still running despite graceful termination request!"
            assert timer.elapsed_time < 10, "It look too long to terminate the child!"

    def test_program_searching(self):
        """Make sure which() works as expected."""
        assert which('python')
        assert not which(MISSING_COMMAND)

    def test_status_code_checking(self):
        """Make sure that status code handling is sane."""
        assert execute('true') is True
        assert execute('false', check=False) is False
        # Make sure execute('false') raises an exception.
        self.assertRaises(ExternalCommandFailed, execute, 'false')
        # Make sure execute('exit 42') raises an exception.
        e = self.assertRaises(ExternalCommandFailed, execute, 'exit 42')
        # Make sure the exception has the expected properties.
        self.assertEqual(e.command.command_line, ['bash', '-c', 'exit 42'])
        self.assertEqual(e.returncode, 42)
        # Make sure the CommandNotFound exception is raised consistently
        # regardless of the values of the `shell' and `async' options.
        for async in True, False:
            for shell in True, False:
                cmd = ExternalCommand(MISSING_COMMAND, async=async, shell=shell)
                self.assertRaises(CommandNotFound, cmd.wait)
                assert cmd.returncode == COMMAND_NOT_FOUND_STATUS

    def test_shell_opt_out(self):
        """Test that callers can always opt out of shell evaluation."""
        # A command consisting of a single string implies shell evaluation but
        # you can change that default.
        assert DEFAULT_SHELL in ExternalCommand('echo 42').command_line
        assert DEFAULT_SHELL not in ExternalCommand('echo 42', shell=False).command_line
        # A command consisting of more than one string implies no shell
        # evaluation but you can change that default.
        assert DEFAULT_SHELL not in ExternalCommand('echo', '42').command_line
        assert DEFAULT_SHELL in ExternalCommand('echo', '42', shell=True).command_line
        # Confirm that opting out of shell evaluation really bypasses all shells.
        cmd = ExternalCommand(
            'echo this will never match an executable name',
            shell=False, check=False,
        )
        cmd.start()
        cmd.wait()
        assert cmd.error_type is CommandNotFound

    def test_commands_on_stdin(self):
        """Test that callers can opt in to shell evaluation for local commands given on standard input."""
        random_string = uuid.uuid4().hex
        output = execute(capture=True, shell=True, input='echo %s' % quote(random_string))
        assert output == random_string

    def test_remote_commands_on_stdin(self):
        """Test that callers can opt in to shell evaluation for remote commands given on standard input."""
        random_string = uuid.uuid4().hex
        with SSHServer() as server:
            output = remote('127.0.0.1',
                            capture=True, shell=True,
                            input='echo %s' % quote(random_string),
                            **server.client_options)
            assert output == random_string

    def test_stdin(self):
        """Make sure standard input can be provided to external commands."""
        assert execute('tr', 'a-z', 'A-Z', input='test', capture=True) == 'TEST'

    def test_stdout(self):
        """Make sure standard output of external commands can be captured."""
        assert execute('echo', 'this is a test', capture=True) == 'this is a test'
        assert execute('echo', '-e', r'line 1\nline 2', capture=True) == 'line 1\nline 2\n'
        # I don't know how to test for the effect of silent=True in a practical
        # way without creating the largest test in this test suite :-). The
        # least I can do is make sure the keyword argument is accepted and the
        # code runs without exceptions in supported environments.
        assert execute('echo', 'this is a test', silent=True) is True

    def test_stderr(self):
        """Make sure standard error of external commands can be captured."""
        stdout_value = 'this goes to standard output'
        stderr_value = 'and this goes to the standard error stream'
        shell_command = 'echo %s; echo %s >&2' % (stdout_value, stderr_value)
        cmd = ExternalCommand(shell_command, capture=True, capture_stderr=True)
        cmd.start()
        assert stdout_value in cmd.decoded_stdout
        assert stderr_value in cmd.decoded_stderr

    def test_output_on_error(self):
        """Check the combination of output capturing and error handling."""
        for shell in True, False:
            cmd = ExternalCommand(MISSING_COMMAND, capture=True, check=False, shell=shell)
            cmd.start()
            cmd.wait()
            assert cmd.returncode == COMMAND_NOT_FOUND_STATUS
            assert cmd.stdout == ''

    def test_merged_streams(self):
        """Make sure standard output/error of external commands can be captured together."""
        stdout_value = 'this goes to standard output'
        stderr_value = 'and this goes to the standard error stream'
        shell_command = 'echo %s; echo %s >&2' % (stdout_value, stderr_value)
        cmd = ExternalCommand(shell_command, capture=True, merge_streams=True)
        cmd.start()
        assert stdout_value in cmd.decoded_stdout
        assert stderr_value in cmd.decoded_stdout
        assert stdout_value not in (cmd.decoded_stderr or '')
        assert stderr_value not in (cmd.decoded_stderr or '')

    def test_stdout_to_file(self):
        """Make sure the standard output stream of external commands can be redirected and appended to a file."""
        fd, filename = tempfile.mkstemp(prefix='executor-', suffix='-stdout.txt')
        with open(filename, 'w') as handle:
            handle.write('existing contents\n')
        with open(filename, 'a') as handle:
{'fraction_non_alphanumeric': 0.08318164500698472, 'fraction_numerical': 0.00431782584769081, 'mean_word_length': 2.9804549283909014, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1307908', 'n_tokens_mistral': 6446, 'n_tokens_neox': 6138, 'n_words': 2271}