Complete the transfer of maintainer hat to myself.

[dput.git] / test / test_methods.py

# -*- coding: utf-8; -*-
#
# test/test_methods.py
# Part of ‘dput’, a Debian package upload toolkit.
#
# Copyright © 2015–2016 Ben Finney <ben+python@benfinney.id.au>
#
# This is free software: you may copy, modify, and/or distribute this work
# under the terms of the GNU General Public License as published by the
# Free Software Foundation; version 3 of that license or any later version.
# No warranty expressed or implied. See the file ‘LICENSE.GPL-3’ for details.

""" Unit tests for upload method behaviour. """

from __future__ import (absolute_import, unicode_literals)

import sys
import os
import os.path
import subprocess
import io
import stat
import pkgutil
import importlib
import collections
import tempfile
import textwrap
import doctest
import getpass
import ftplib

if sys.version_info >= (3, 3):
    import urllib.parse as urlparse
elif sys.version_info >= (3, 0):
    raise RuntimeError("Python 3 earlier than 3.3 is not supported.")
else:
    import urlparse

import testtools
import testscenarios
import httpretty

__package__ = str("test")
__import__(__package__)

sys.path.insert(1, os.path.dirname(os.path.dirname(__file__)))
import dput.dput
import dput.helper.dputhelper
import dput.methods
import dput.methods.local
import dput.methods.ftp
import dput.methods.http
import dput.methods.https
import dput.methods.scp
import dput.methods.rsync

from .helper import (
        mock,
        FakeSystemExit,
        patch_system_interfaces,
        EXIT_STATUS_FAILURE,
        FileDouble,
        patch_os_stat,
        patch_os_lstat,
        setup_file_double_behaviour,
        patch_subprocess_check_call,
        ARG_ANY,
        SubprocessDouble,
        )
from .test_dputhelper import (
        patch_filewithprogress,
        )

\f
class import_upload_functions_TestCase(
        testscenarios.WithScenarios,
        testtools.TestCase):
    """ Test cases for `import_upload_functions` function. """

    scenarios = [
            ('empty', {
                'module_names': [],
                }),
            ('one', {
                'module_names': ["foo"],
                }),
            ('three', {
                'module_names': ["foo", "bar", "baz"],
                }),
            ]

    for (scenario_name, scenario) in scenarios:
        modules_by_name = collections.OrderedDict()
        iter_modules_result = []
        for module_name in scenario['module_names']:
            module = mock.Mock()
            module.__name__ = module_name
            module.upload = mock.Mock()
            modules_by_name[module_name] = module
            module_entry = (module, module_name, False)
            iter_modules_result.append(module_entry)
        scenario['modules_by_name'] = modules_by_name
        scenario['iter_modules_result'] = iter_modules_result
    del scenario_name, scenario

    def setUp(self):
        """ Set up test fixtures. """
        super(import_upload_functions_TestCase, self).setUp()
        patch_system_interfaces(self)

        self.patch_import_functions()

    def patch_import_functions(self):
        """ Patch import functions used by the function. """
        self.patch_pkgutil_iter_modules()
        self.patch_importlib_import_module()

    def patch_pkgutil_iter_modules(self):
        """ Patch `pkgutil.iter_modules` function for this test case. """
        func_patcher = mock.patch.object(
                pkgutil, "iter_modules", autospec=True)
        func_patcher.start()
        self.addCleanup(func_patcher.stop)
        pkgutil.iter_modules.return_value = self.iter_modules_result

    def patch_importlib_import_module(self):
        """ Patch `importlib.import_module` function for this test case. """
        func_patcher = mock.patch.object(
                importlib, "import_module", autospec=True)
        func_patcher.start()
        self.addCleanup(func_patcher.stop)

        def fake_import_module(full_name):
            module_name = full_name.split(".")[-1]
            module = self.modules_by_name[module_name]
            return module

        importlib.import_module.side_effect = fake_import_module

    @mock.patch.object(dput.dput, 'debug', new=True)
    def test_emits_debug_message_for_modules_found(self):
        """ Should emit a debug message for the modules found. """
        expected_message = "D: modules_found: {names!r}".format(
                names=self.module_names)
        dput.dput.import_upload_functions()
        self.assertIn(expected_message, sys.stdout.getvalue())

    @mock.patch.object(dput.dput, 'debug', new=True)
    def test_emits_debug_message_for_each_import(self):
        """ Should emit a debug message for each module imported. """
        dput.dput.import_upload_functions()
        for (module_name, module) in self.modules_by_name.items():
            expected_message = "D: Module: {name} ({module!r})".format(
                    name=module_name, module=module)
            self.assertIn(expected_message, sys.stdout.getvalue())

    @mock.patch.object(dput.dput, 'debug', new=True)
    def test_emits_debug_message_for_each_upload_method(self):
        """ Should emit a debug message for each upload method. """
        dput.dput.import_upload_functions()
        for module_name in self.module_names:
            expected_message = "D: Method name: {name}".format(
                    name=module_name)
            self.assertIn(expected_message, sys.stdout.getvalue())

    def test_returns_expected_function_mapping(self):
        """ Should return expected mapping of upload functions. """
        result = dput.dput.import_upload_functions()
        expected_result = {
                name: self.modules_by_name[name].upload
                for name in self.module_names}
        self.assertEqual(expected_result, result)

\f
def patch_getpass_getpass(testcase):
    """ Patch the `getpass.getpass` function for the test case. """
    func_patcher = mock.patch.object(getpass, "getpass", autospec=True)
    func_patcher.start()
    testcase.addCleanup(func_patcher.stop)

\f
class upload_TestCase(
        testscenarios.WithScenarios,
        testtools.TestCase):
    """ Base for test cases for method modules `upload` functions. """

    files_scenarios = [
            ('file-list-empty', {
                'paths_to_upload': [],
                }),
            ('file-list-one', {
                'paths_to_upload': [tempfile.mktemp()],
                }),
            ('file-list-three', {
                'paths_to_upload': [tempfile.mktemp() for __ in range(3)],
                }),
            ]

    check_call_scenarios = [
            ('check-call-success', {
                'check_call_scenario_name': 'success',
                }),
            ('check-call-failure', {
                'check_call_scenario_name': 'failure',
                'check_call_error': subprocess.CalledProcessError,
                }),
            ]

    check_call_success_scenarios = [
            (name, params) for (name, params) in check_call_scenarios
            if 'check_call_error' not in params]

    incoming_scenarios = [
            ('incoming-simple', {
                'incoming_path': tempfile.mktemp(),
                }),
            ('incoming-no-leading-slash', {
                'incoming_path': tempfile.mktemp().lstrip("/"),
                }),
            ('incoming-has-trailing-slash', {
                'incoming_path': tempfile.mktemp() + "/",
                }),
            ]

    def setUp(self):
        """ Set up test fixtures. """
        super(upload_TestCase, self).setUp()
        patch_system_interfaces(self)

        patch_subprocess_check_call(self)

        patch_os_stat(self)
        self.set_file_doubles()
        setup_file_double_behaviour(self)

        patch_filewithprogress(self)

        self.set_test_args()

    def set_file_doubles(self):
        """ Set the file doubles for this test case. """
        for path in self.paths_to_upload:
            fake_file = getattr(self, 'fake_file', None)
            double = FileDouble(path, fake_file)
            double.set_os_stat_scenario(
                    getattr(self, 'os_stat_scenario_name', "okay"))
            double.register_for_testcase(self)

            func_patcher = mock.patch.object(
                    double.fake_file, "close", autospec=True)
            func_patcher.start()
            self.addCleanup(func_patcher.stop)

    def set_test_args(self):
        """ Set the arguments for the test call to the function. """
        raise NotImplementedError

    def get_command_args_from_latest_check_call(self):
        """ Get command line arguments from latest `subprocess.check_call`. """
        latest_call = subprocess.check_call.call_args
        (args, kwargs) = latest_call
        command_args = args[0]
        return command_args

\f
class local_upload_TestCase(upload_TestCase):
    """ Test cases for `methods.local.upload` function. """

    scenarios = testscenarios.multiply_scenarios(
            upload_TestCase.incoming_scenarios,
            upload_TestCase.files_scenarios,
            upload_TestCase.check_call_success_scenarios)

    command_file_path = os.path.join("/usr/bin", "install")

    def setUp(self):
        """ Set up test fixtures. """
        super(local_upload_TestCase, self).setUp()

        self.set_subprocess_double()

    def set_test_args(self):
        """ Set the arguments for the test call to the function. """
        self.test_args = dict(
                fqdn=object(),
                login=object(),
                incoming=self.incoming_path,
                files_to_upload=self.paths_to_upload,
                debug=None,
                compress=object(),
                progress=object(),
                )

    def set_subprocess_double(self):
        """ Set the test double for the subprocess. """
        argv = [self.command_file_path, "-m", ARG_ANY, ARG_ANY]
        double = SubprocessDouble(self.command_file_path, argv=argv)
        double.register_for_testcase(self)
        double.set_subprocess_check_call_scenario(
                self.check_call_scenario_name)
        self.subprocess_double = double

    def test_calls_check_call_with_install_command(self):
        """ Should call `subprocess.check_call` to invoke ‘install’. """
        dput.methods.local.upload(**self.test_args)
        command_args = self.get_command_args_from_latest_check_call()
        expected_command = self.command_file_path
        self.assertEqual(expected_command, command_args[0])

    def test_calls_check_call_with_mode_option_in_command(self):
        """ Should call `subprocess.check_call`, invoke command with mode. """
        dput.methods.local.upload(**self.test_args)
        command_args = self.get_command_args_from_latest_check_call()
        expected_mode = (
                stat.S_IRUSR | stat.S_IWUSR
                | stat.S_IRGRP
                | stat.S_IROTH)
        expected_mode_text = "{mode:04o}".format(mode=expected_mode)[-3:]
        expected_option_args = ["-m", expected_mode_text]
        self.assertEqual(expected_option_args, command_args[1:3])

    def test_calls_check_call_with_file_paths_in_command(self):
        """ Should call `subprocess.check_call` with file paths in command. """
        dput.methods.local.upload(**self.test_args)
        command_args = self.get_command_args_from_latest_check_call()
        self.assertEqual(self.paths_to_upload, command_args[3:-1])

    def test_calls_check_call_with_incoming_path_as_final_arg(self):
        """ Should call `subprocess.check_call` with incoming path. """
        dput.methods.local.upload(**self.test_args)
        command_args = self.get_command_args_from_latest_check_call()
        self.assertEqual(self.incoming_path, command_args[-1])

    def test_emits_debug_message_for_upload_command(self):
        """ Should emit a debug message for the upload command. """
        self.test_args['debug'] = True
        dput.methods.local.upload(**self.test_args)
        expected_message = textwrap.dedent("""\
                D: Uploading with cp to {dir_path}
                D: ...
                """).format(dir_path=self.incoming_path)
        self.assertThat(
                sys.stdout.getvalue(),
                testtools.matchers.DocTestMatches(
                    expected_message, flags=doctest.ELLIPSIS))

    def test_calls_sys_exit_if_check_call_returns_nonzero(self):
        """ Should call `sys.exit` if `subprocess.check_call` fails. """
        self.subprocess_double.set_subprocess_check_call_scenario('failure')
        with testtools.ExpectedException(FakeSystemExit):
            dput.methods.local.upload(**self.test_args)
        expected_output = textwrap.dedent("""\
                Error while uploading.
                """)
        self.assertIn(expected_output, sys.stdout.getvalue())
        sys.exit.assert_called_with(EXIT_STATUS_FAILURE)

\f
class ftp_upload_TestCase(upload_TestCase):
    """ Test cases for `methods.ftp.upload` function. """

    scenarios = NotImplemented

    def setUp(self):
        """ Set up test fixtures. """
        super(ftp_upload_TestCase, self).setUp()

        self.set_ftp_client()
        self.patch_ftplib_ftp()

        patch_getpass_getpass(self)
        self.fake_password = self.getUniqueString()
        getpass.getpass.return_value = self.fake_password
        if not hasattr(self, 'expected_password'):
            self.expected_password = self.fake_password

    def set_test_args(self):
        """ Set the arguments for the test call to the function. """
        if not hasattr(self, 'progress_type'):
            self.progress_type = 0
        self.test_args = dict(
                fqdn=self.getUniqueString(),
                login=self.login,
                incoming=self.incoming_path,
                files_to_upload=self.paths_to_upload,
                debug=False,
                ftp_mode=self.ftp_mode,
                progress=self.progress_type,
                port=object(),
                )

    def set_ftp_client(self):
        """ Set the FTP client double. """
        self.ftp_client = mock.MagicMock(name="FTP")

    def patch_ftplib_ftp(self):
        """ Patch `ftplib.FTP` class for this test case. """
        patcher = mock.patch.object(ftplib, "FTP", autospec=True)
        patcher.start()
        self.addCleanup(patcher.stop)

        ftplib.FTP.return_value = self.ftp_client


class ftp_upload_NormalFilesTestCase(ftp_upload_TestCase):
    """ Test cases for `methods.ftp.upload` function, upload normal files. """

    login_scenarios = [
            ('anonymous', {
                'login': "anonymous",
                'expected_password': "dput@packages.debian.org",
                }),
            ('username', {
                'login': "lorem",
                }),
            ]

    ftp_client_scenarios = [
            ('default', {
                'ftp_mode': False,
                }),
            ('passive-mode', {
                'ftp_mode': True,
                }),
            ]

    scenarios = testscenarios.multiply_scenarios(
            upload_TestCase.incoming_scenarios,
            upload_TestCase.files_scenarios,
            login_scenarios, ftp_client_scenarios)

    def test_emits_debug_message_for_connect(self):
        """ Should emit debug message for successful connect. """
        self.test_args['debug'] = True
        dput.methods.ftp.upload(**self.test_args)
        expected_fqdn = self.test_args['fqdn']
        expected_output = textwrap.dedent("""\
                D: FTP-Connection to host: {fqdn}
                """).format(fqdn=expected_fqdn)
        self.assertIn(expected_output, sys.stdout.getvalue())

    def test_calls_ftp_connect_with_expected_args(self):
        """ Should call `FTP.connect` with expected args. """
        dput.methods.ftp.upload(**self.test_args)
        expected_args = (
                self.test_args['fqdn'],
                self.test_args['port'],
                )
        self.ftp_client.connect.assert_called_with(*expected_args)

    def test_emits_error_message_when_ftp_connect_error(self):
        """ Should emit error message when `FTP.connect` raises error. """
        self.ftp_client.connect.side_effect = ftplib.error_temp
        try:
            dput.methods.ftp.upload(**self.test_args)
        except FakeSystemExit:
            pass
        expected_output = "Connection failed, aborting"
        self.assertIn(expected_output, sys.stdout.getvalue())

    def test_calls_sys_exit_when_ftp_connect_permission_error(self):
        """ Should call `sys.exit` when `FTP.connect` raises error. """
        self.ftp_client.connect.side_effect = ftplib.error_temp
        with testtools.ExpectedException(FakeSystemExit):
            dput.methods.ftp.upload(**self.test_args)
        sys.exit.assert_called_with(EXIT_STATUS_FAILURE)

    def test_calls_ftp_login_with_expected_args(self):
        """ Should call `FTP.login` with expected args. """
        dput.methods.ftp.upload(**self.test_args)
        expected_args = (
                self.test_args['login'],
                self.expected_password,
                )
        self.ftp_client.login.assert_called_with(*expected_args)

    def test_emits_error_message_when_ftp_login_permission_error(self):
        """ Should emit error message when `FTP.login` permission error. """
        self.ftp_client.login.side_effect = ftplib.error_perm
        try:
            dput.methods.ftp.upload(**self.test_args)
        except FakeSystemExit:
            pass
        expected_output = "Wrong Password"
        self.assertIn(expected_output, sys.stdout.getvalue())

    def test_calls_sys_exit_when_ftp_login_permission_error(self):
        """ Should call `sys.exit` when `FTP.login` permission error. """
        self.ftp_client.login.side_effect = ftplib.error_perm
        with testtools.ExpectedException(FakeSystemExit):
            dput.methods.ftp.upload(**self.test_args)
        sys.exit.assert_called_with(EXIT_STATUS_FAILURE)

    def test_emits_error_message_when_ftp_login_eof_error(self):
        """ Should emit error message when `FTP.login` EOF error. """
        self.ftp_client.login.side_effect = EOFError
        try:
            dput.methods.ftp.upload(**self.test_args)
        except FakeSystemExit:
            pass
        expected_output = "Server closed the connection"
        self.assertIn(expected_output, sys.stdout.getvalue())

    def test_calls_sys_exit_when_ftp_login_eof_error(self):
        """ Should call `sys.exit` when `FTP.login` EOF error. """
        self.ftp_client.login.side_effect = EOFError
        with testtools.ExpectedException(FakeSystemExit):
            dput.methods.ftp.upload(**self.test_args)
        sys.exit.assert_called_with(EXIT_STATUS_FAILURE)

    def test_calls_ftp_set_pasv_with_expected_args(self):
        """ Should call `FTP.set_pasv` with expected args. """
        dput.methods.ftp.upload(**self.test_args)
        expected_mode = bool(self.test_args['ftp_mode'])
        expected_args = (expected_mode,)
        self.ftp_client.set_pasv.assert_called_with(*expected_args)

    def test_calls_ftp_cwd_with_expected_args(self):
        """ Should call `FTP.cwd` with expected args. """
        dput.methods.ftp.upload(**self.test_args)
        expected_path = self.incoming_path
        expected_args = (expected_path,)
        self.ftp_client.cwd.assert_called_with(*expected_args)

    def test_emits_debug_message_for_cwd(self):
        """ Should emit debug message for successful `FTP.cwd`. """
        self.test_args['debug'] = True
        dput.methods.ftp.upload(**self.test_args)
        expected_output = textwrap.dedent("""\
                D: Directory to upload to: {path}
                """).format(path=self.incoming_path)
        self.assertIn(expected_output, sys.stdout.getvalue())

    def test_emits_error_message_when_destination_directory_not_found(self):
        """ Should emit error message when destination directory not found. """
        error = ftplib.error_perm("550 Not Found")
        self.ftp_client.cwd.side_effect = error
        try:
            dput.methods.ftp.upload(**self.test_args)
        except FakeSystemExit:
            pass
        expected_output = "Directory to upload to does not exist."
        self.assertIn(expected_output, sys.stdout.getvalue())

    def test_calls_sys_exit_when_ftp_cwd_permission_error(self):
        """ Should call `sys.exit` when `FTP.cwd` permission error. """
        error = ftplib.error_perm("550 Not Found")
        self.ftp_client.cwd.side_effect = error
        with testtools.ExpectedException(FakeSystemExit):
            dput.methods.ftp.upload(**self.test_args)
        sys.exit.assert_called_with(EXIT_STATUS_FAILURE)

    def test_propagates_exception_when_ftp_cwd_permission_error(self):
        """ Should call `sys.exit` when `FTP.cwd` permission error. """
        error = ftplib.error_perm("500 Bad Stuff Happened")
        self.ftp_client.cwd.side_effect = error
        with testtools.ExpectedException(error.__class__):
            dput.methods.ftp.upload(**self.test_args)

    def test_propagates_exception_when_ftp_cwd_eof_error(self):
        """ Should call `sys.exit` when `FTP.cwd` EOF error. """
        error = EOFError()
        self.ftp_client.cwd.side_effect = error
        with testtools.ExpectedException(error.__class__):
            dput.methods.ftp.upload(**self.test_args)

    def test_emits_debug_message_for_each_file(self):
        """ Should emit debug message for each file to upload. """
        self.test_args['debug'] = True
        dput.methods.ftp.upload(**self.test_args)
        expected_output = "".join(textwrap.dedent("""\
                D: Uploading File: {path}
                  Uploading {filename}: done.
                """).format(path=path, filename=os.path.basename(path))
                for path in self.paths_to_upload)
        self.assertIn(expected_output, sys.stdout.getvalue())

    def test_calls_ftp_storbinary_for_each_file(self):
        """ Should call `FTP.storbinary` for each file to upload. """
        dput.methods.ftp.upload(**self.test_args)
        registry = FileDouble.get_registry_for_testcase(self)
        expected_blocksize = 1024
        expected_calls = [
                mock.call(
                    "STOR {filename}".format(filename=os.path.basename(path)),
                    registry[path].fake_file, expected_blocksize)
                for path in self.paths_to_upload]
        self.ftp_client.storbinary.assert_has_calls(
                expected_calls, any_order=True)

    def test_calls_close_for_each_file(self):
        """ Should call `file.close` for each file to upload. """
        dput.methods.ftp.upload(**self.test_args)
        registry = FileDouble.get_registry_for_testcase(self)
        for path in self.paths_to_upload:
            fake_file = registry[path].fake_file
            fake_file.close.assert_called_with()


class ftp_upload_ErrorTestCase(ftp_upload_TestCase):
    """ Test cases for `methods.ftp.upload` function, error conditions. """

    login_scenarios = [
            ('anonymous', {
                'login': "anonymous",
                'expected_password': "dput@packages.debian.org",
                }),
            ]

    ftp_client_scenarios = [
            ('default', {
                'ftp_mode': False,
                }),
            ]

    progress_scenarios = [
            ('progress-type-0', {
                'progress_type': 0,
                }),
            ('progress-type-1', {
                'progress_type': 1,
                }),
            ('progress-type-2', {
                'progress_type': 2,
                }),
            ]

    files_scenarios = list(
            (scenario_name, scenario) for (scenario_name, scenario)
            in upload_TestCase.files_scenarios
            if scenario['paths_to_upload'])

    scenarios = testscenarios.multiply_scenarios(
            upload_TestCase.incoming_scenarios,
            files_scenarios,
            login_scenarios, ftp_client_scenarios, progress_scenarios)

    def test_emits_warning_when_remote_file_exists(self):
        """ Should emit a warning message when remote file exists. """
        error = ftplib.error_perm("553 Exists")
        self.ftp_client.storbinary.side_effect = error
        dput.methods.ftp.upload(**self.test_args)
        for path in self.paths_to_upload:
            expected_output = textwrap.dedent("""\
                    Leaving existing {path} on the server and continuing
                    """).format(path=os.path.basename(path))
            self.expectThat(
                    sys.stdout.getvalue(),
                    testtools.matchers.Contains(expected_output))

    def test_omits_sys_exit_when_remote_file_exists(self):
        """ Should omit call to `sys.exit` when remote file exists. """
        error = ftplib.error_perm("553 Exists")
        self.ftp_client.storbinary.side_effect = error
        dput.methods.ftp.upload(**self.test_args)
        self.assertFalse(sys.exit.called)

    def test_emits_error_message_when_storbinary_failure(self):
        """ Should emit an error message when `FTP.storbinary` failure. """
        error = ftplib.error_perm("504 Weird Stuff Happened")
        self.ftp_client.storbinary.side_effect = error
        try:
            dput.methods.ftp.upload(**self.test_args)
        except FakeSystemExit:
            pass
        for path in self.paths_to_upload[:1]:
            expected_output = (
                    "Note: This error might indicate a problem with"
                    " your passive_ftp setting.\n")
            self.expectThat(
                    sys.stdout.getvalue(),
                    testtools.matchers.Contains(expected_output))

    def test_calls_sys_exit_when_storbinary_failure(self):
        """ Should call `sys.exit` when `FTP.storbinary` failure. """
        error = ftplib.error_perm("504 Weird Stuff Happened")
        self.ftp_client.storbinary.side_effect = error
        with testtools.ExpectedException(FakeSystemExit):
            dput.methods.ftp.upload(**self.test_args)
        sys.exit.assert_called_with(EXIT_STATUS_FAILURE)

    def test_emits_debug_message_when_open_failure(self):
        """ Should emit a debug message when `builtins.open` failure. """
        self.test_args['debug'] = True
        registry = FileDouble.get_registry_for_testcase(self)
        for path in self.paths_to_upload:
            double = registry[path]
            double.set_open_scenario('nonexist')
        try:
            dput.methods.ftp.upload(**self.test_args)
        except EnvironmentError:
            pass
        expected_output = (
                "D: Should exit silently now, but"
                " will throw exception for debug.")
        self.assertIn(expected_output, sys.stdout.getvalue())

    def test_propagates_error_from_storbinary_for_debug(self):
        """ Should propagate error from `FTP.storbinary` when debug. """
        self.test_args['debug'] = True
        error = ftplib.error_perm("504 Weird Stuff Happened")
        self.ftp_client.storbinary.side_effect = error
        with testtools.ExpectedException(error.__class__):
            dput.methods.ftp.upload(**self.test_args)

    def test_propagates_error_from_quit_for_debug(self):
        """ Should propagate error from `FTP.quit` when debug. """
        self.test_args['debug'] = True
        error = ftplib.error_perm("504 Weird Stuff Happened")
        self.ftp_client.quit.side_effect = error
        with testtools.ExpectedException(error.__class__):
            dput.methods.ftp.upload(**self.test_args)


def make_expected_filewithprogress_attributes_by_path(testcase, attrs):
    """ Make a mapping from path to expected FileWithProgress attribs. """
    expected_attributes_by_path = {}
    registry = FileDouble.get_registry_for_testcase(testcase)
    for path in testcase.paths_to_upload:
        file_double = registry[path]
        expected_attributes = {
                'f': file_double.fake_file,
                'size': file_double.stat_result.st_size,
                }
        expected_attributes.update(attrs)
        expected_attributes_by_path[path] = expected_attributes

    return expected_attributes_by_path


class ftp_upload_ProgressTestCase(ftp_upload_TestCase):
    """ Test cases for `methods.ftp.upload` function, with progress meter. """

    login_scenarios = [
            ('anonymous', {
                'login': "anonymous",
                'expected_password': "dput@packages.debian.org",
                }),
            ]

    ftp_client_scenarios = [
            ('default', {
                'ftp_mode': False,
                }),
            ]

    progress_scenarios = [
            ('progress-type-1', {
                'progress_type': 1,
                }),
            ('progress-type-2', {
                'progress_type': 2,
                }),
            ]

    scenarios = testscenarios.multiply_scenarios(
            upload_TestCase.incoming_scenarios,
            upload_TestCase.files_scenarios,
            login_scenarios, ftp_client_scenarios, progress_scenarios)

    def test_calls_storbinary_with_filewithprogress(self):
        """ Should use a `FileWithProgress` to call `FTP.storbinary`. """
        dput.methods.ftp.upload(**self.test_args)
        expected_calls = [
                mock.call(
                    mock.ANY, self.fake_filewithprogress,
                    mock.ANY)
                for path in self.paths_to_upload]
        self.ftp_client.storbinary.assert_has_calls(
                expected_calls, any_order=True)

    def test_filewithprogress_has_expected_attributes(self):
        """ Should have expected attributes on the `FileWithProgress`. """
        expected_attributes_by_path = (
                make_expected_filewithprogress_attributes_by_path(
                    self, {'ptype': self.progress_type}))
        dput.methods.ftp.upload(**self.test_args)
        for call in self.ftp_client.storbinary.mock_calls:
            (__, call_args, call_kwargs) = call
            (__, stor_file, __) = call_args
            path = stor_file.f.name
            expected_attributes = expected_attributes_by_path[path]
            stor_file_attributes = {
                    name: getattr(stor_file, name)
                    for name in expected_attributes}
            self.expectThat(
                    expected_attributes,
                    testtools.matchers.Equals(stor_file_attributes))

    def test_filewithprogress_has_sentinel_size_when_stat_failure(self):
        """ Should have sentinel `size` value when `os.stat` failure. """
        expected_attributes_by_path = (
                make_expected_filewithprogress_attributes_by_path(
                    self, {'size': -1}))
        registry = FileDouble.get_registry_for_testcase(self)
        for path in self.paths_to_upload:
            double = registry[path]
            double.set_os_stat_scenario('notfound_error')
        dput.methods.ftp.upload(**self.test_args)
        for call in self.ftp_client.storbinary.mock_calls:
            (__, call_args, call_kwargs) = call
            (__, stor_file, __) = call_args
            path = stor_file.f.name
            expected_attributes = expected_attributes_by_path[path]
            stor_file_attributes = {
                    name: getattr(stor_file, name)
                    for name in expected_attributes}
            self.expectThat(
                    expected_attributes,
                    testtools.matchers.Equals(stor_file_attributes))

    def test_emits_debug_message_when_stat_failure(self):
        """ Should have sentinel `size` value when `os.stat` failure. """
        self.test_args['debug'] = True
        registry = FileDouble.get_registry_for_testcase(self)
        for path in self.paths_to_upload:
            double = registry[path]
            double.set_os_stat_scenario('notfound_error')
        dput.methods.ftp.upload(**self.test_args)
        for path in self.paths_to_upload:
            expected_output = textwrap.dedent("""\
                    D: Determining size of file '{path}' failed
                    """).format(path=path)
            self.expectThat(
                    sys.stdout.getvalue(),
                    testtools.matchers.Contains(expected_output))

\f
class http_upload_TestCase(upload_TestCase):
    """ Base for test cases for `methods.http.upload` function. """

    scenarios = NotImplemented

    protocol_scenarios = [
            ('http', {
                'function_to_test': dput.methods.http.upload,
                'protocol': "http",
                'protocol_version': "HTTP/1.0",
                }),
            ('https', {
                'function_to_test': dput.methods.https.upload,
                'protocol': "https",
                'protocol_version': "HTTP/1.0",
                }),
            ]

    login_scenarios = [
            ('username', {
                'login': "lorem",
                }),
            ]

    def setUp(self):
        """ Set up test fixtures. """
        super(http_upload_TestCase, self).setUp()

        httpretty.enable()
        self.addCleanup(httpretty.disable)

        self.set_response_header_fields()
        self.patch_put_requests()

        patch_getpass_getpass(self)
        self.fake_password = self.getUniqueString()
        getpass.getpass.return_value = self.fake_password
        if not hasattr(self, 'expected_password'):
            self.expected_password = self.fake_password

    def set_test_args(self):
        """ Set the arguments for the test call to the function. """
        if not hasattr(self, 'progress_type'):
            self.progress_type = 0
        self.test_args = dict(
                fqdn=self.getUniqueString(),
                login=self.login,
                incoming=self.incoming_path,
                files_to_upload=self.paths_to_upload,
                debug=False,
                dummy=object(),
                progress=self.progress_type,
                )
        if self.function_to_test is dput.methods.http.upload:
            self.test_args['protocol'] = self.protocol

    def make_upload_uri(self, file_name):
        """ Make the URI for a file for upload. """
        uri = urlparse.urlunsplit([
                self.protocol, self.test_args['fqdn'],
                os.path.join(os.path.sep, self.incoming_path, file_name),
                None, None])
        return uri

    def set_response_header_fields(self):
        """ Set the header fields for the HTTP response. """
        if not hasattr(self, 'response_header_fields'):
            self.response_header_fields = {}

    def patch_put_requests(self):
        """ Patch the HTTP PUT requests. """
        self.path_by_request_uri = {}
        for path in self.paths_to_upload:
            upload_uri = self.make_upload_uri(os.path.basename(path))
            self.path_by_request_uri[upload_uri] = path
            response_body = ""
            httpretty.register_uri(
                    httpretty.PUT, upload_uri,
                    status=self.status_code, body=response_body)


class http_upload_SuccessTestCase(http_upload_TestCase):
    """ Success test cases for `methods.http.upload` function. """

    response_scenarios = [
            ('okay', {
                'status_code': 200,
                'status_reason': "Okay",
                }),
            ('chatter', {
                'status_code': 203,
                'status_reason': "Non-Authoritative Information",
                }),
            ]

    auth_scenarios = [
            ('auth-accepted', {
                'auth_response_status_code': 200,
                'auth_response_status_reason': "Okay",
                }),
            ]

    size_scenarios = [
            ('size-empty', {
                'fake_file': io.BytesIO(),
                }),
            ('size-1k', {
                'fake_file': io.BytesIO(
                    b"Lorem ipsum, dolor sit amet.___\n" * 32),
                }),
            ('size-100k', {
                'fake_file': io.BytesIO(
                    b"Lorem ipsum, dolor sit amet.___\n" * 3200),
                }),
            ]

    incoming_scenarios = list(upload_TestCase.incoming_scenarios)
    for (scenario_name, scenario) in incoming_scenarios:
        scenario['expected_url_path_prefix'] = os.path.join(
                os.path.sep, scenario['incoming_path'])
    del scenario_name, scenario

    scenarios = testscenarios.multiply_scenarios(
            upload_TestCase.files_scenarios,
            size_scenarios,
            upload_TestCase.incoming_scenarios,
            http_upload_TestCase.protocol_scenarios,
            http_upload_TestCase.login_scenarios,
            response_scenarios, auth_scenarios)

    def test_emits_debug_message_for_upload(self):
        """ Should emit debug message for upload. """
        self.test_args['debug'] = True
        self.function_to_test(**self.test_args)
        for path in self.paths_to_upload:
            expected_uri = self.make_upload_uri(os.path.basename(path))
            expected_output = textwrap.dedent("""\
                    D: HTTP-PUT to URL: {uri}
                    """).format(uri=expected_uri)
            self.expectThat(
                    sys.stdout.getvalue(),
                    testtools.matchers.Contains(expected_output))

    def test_request_has_expected_fields(self):
        """ Should send request with expected fields in header. """
        if not self.paths_to_upload:
            self.skipTest("No files to upload")
        self.function_to_test(**self.test_args)
        registry = FileDouble.get_registry_for_testcase(self)
        path = self.paths_to_upload[-1]
        double = registry[path]
        request = httpretty.last_request()
        expected_fields = {
                'User-Agent': "dput",
                'Connection': "close",
                'Content-Length': "{size:d}".format(
                    size=len(double.fake_file.getvalue())),
                }
        for (name, value) in expected_fields.items():
            self.expectThat(
                    request.headers.get(name),
                    testtools.matchers.Equals(value))


class http_upload_ProgressTestCase(http_upload_TestCase):
    """ Test cases for `methods.http.upload` function, with progress meter. """

    files_scenarios = list(
            (scenario_name, scenario) for (scenario_name, scenario)
            in upload_TestCase.files_scenarios
            if scenario['paths_to_upload'])

    response_scenarios = [
            ('okay', {
                'status_code': 200,
                'status_reason': "Okay",
                }),
            ]

    progress_scenarios = [
            ('progress-type-1', {
                'progress_type': 1,
                }),
            ('progress-type-2', {
                'progress_type': 2,
                }),
            ]

    scenarios = testscenarios.multiply_scenarios(
            files_scenarios,
            progress_scenarios,
            upload_TestCase.incoming_scenarios,
            http_upload_TestCase.protocol_scenarios,
            http_upload_TestCase.login_scenarios,
            http_upload_SuccessTestCase.auth_scenarios,
            response_scenarios)

    def test_filewithprogress_has_expected_attributes(self):
        """ Should have expected attributes on the `FileWithProgress`. """
        expected_attributes_by_path = (
                make_expected_filewithprogress_attributes_by_path(
                    self, {'ptype': self.progress_type}))
        self.function_to_test(**self.test_args)
        path = self.paths_to_upload[-1]
        expected_attributes = expected_attributes_by_path[path]
        fake_file_attributes = {
                name: getattr(self.fake_filewithprogress, name)
                for name in expected_attributes}
        self.expectThat(
                expected_attributes,
                testtools.matchers.Equals(fake_file_attributes))


class http_upload_UnknownProtocolTestCase(http_upload_TestCase):
    """ Test cases for `methods.http.upload` function, unknown protocol. """

    files_scenarios = list(
            (scenario_name, scenario) for (scenario_name, scenario)
            in upload_TestCase.files_scenarios
            if scenario['paths_to_upload'])

    protocol_scenarios = [
            ('protocol-bogus', {
                'function_to_test': dput.methods.http.upload,
                'protocol': "b0gUs",
                'protocol_version': "b0gUs",
                'expected_exit_status': EXIT_STATUS_FAILURE,
                }),
            ]

    response_scenarios = [
            (scenario_name, scenario) for (scenario_name, scenario)
            in http_upload_SuccessTestCase.response_scenarios
            if scenario['status_code'] == 200]

    scenarios = testscenarios.multiply_scenarios(
            files_scenarios,
            upload_TestCase.incoming_scenarios,
            protocol_scenarios,
            http_upload_TestCase.login_scenarios,
            response_scenarios,
            http_upload_SuccessTestCase.auth_scenarios)

    def test_emits_error_message_when_unknown_protocol(self):
        """ Should emit error message when unknown protocol. """
        try:
            self.function_to_test(**self.test_args)
        except FakeSystemExit:
            pass
        expected_output = "Wrong protocol for upload "
        self.assertIn(expected_output, sys.stderr.getvalue())

    def test_calls_sys_exit_when_unknown_protocol(self):
        """ Should call `sys.exit` when unknown protocol. """
        with testtools.ExpectedException(FakeSystemExit):
            self.function_to_test(**self.test_args)
        sys.exit.assert_called_with(self.expected_exit_status)


class http_upload_FileStatFailureTestCase(http_upload_TestCase):
    """ Test cases for `methods.http.upload` function, `os.stat` failure. """

    files_scenarios = list(
            (scenario_name, scenario) for (scenario_name, scenario)
            in upload_TestCase.files_scenarios
            if scenario['paths_to_upload'])

    os_stat_scenarios = [
            ('os-stat-notfound', {
                'os_stat_scenario_name': "notfound_error",
                'expected_exit_status': EXIT_STATUS_FAILURE,
                }),
            ('os-stat-denied', {
                'os_stat_scenario_name': "denied_error",
                'expected_exit_status': EXIT_STATUS_FAILURE,
                }),
            ]

    response_scenarios = list(
            (scenario_name, scenario) for (scenario_name, scenario)
            in http_upload_SuccessTestCase.response_scenarios
            if scenario['status_code'] == 200)

    scenarios = testscenarios.multiply_scenarios(
            files_scenarios,
            os_stat_scenarios,
            upload_TestCase.incoming_scenarios,
            http_upload_TestCase.protocol_scenarios,
            http_upload_TestCase.login_scenarios,
            response_scenarios,
            http_upload_SuccessTestCase.auth_scenarios)

    def test_emits_error_message(self):
        """ Should emit error message when `os.stat` failure. """
        try:
            self.function_to_test(**self.test_args)
        except FakeSystemExit:
            pass
        expected_output = textwrap.dedent("""\
                Determining size of file '{path}' failed
                """).format(path=self.paths_to_upload[0])
        self.assertIn(expected_output, sys.stderr.getvalue())

    def test_calls_sys_exit_with_expected_exit_status(self):
        """ Should call `sys.exit` with expected exit status. """
        with testtools.ExpectedException(FakeSystemExit):
            self.function_to_test(**self.test_args)
        sys.exit.assert_called_with(self.expected_exit_status)


class http_upload_ResponseErrorTestCase(http_upload_TestCase):
    """ Error test cases for `methods.http.upload` function. """

    files_scenarios = list(
            (scenario_name, scenario) for (scenario_name, scenario)
            in upload_TestCase.files_scenarios
            if scenario['paths_to_upload'])

    response_scenarios = [
            ('server-error', {
                'status_code': 500,
                'status_reason': "Internal Server Error",
                'auth_response_status_code': 200,
                'auth_response_status_reason': "Okay",
                'expected_exit_status': EXIT_STATUS_FAILURE,
                }),
            ]

    scenarios = testscenarios.multiply_scenarios(
            files_scenarios,
            upload_TestCase.incoming_scenarios,
            http_upload_TestCase.protocol_scenarios,
            http_upload_TestCase.login_scenarios,
            response_scenarios)

    def test_emits_error_message_when_response_status_error(self):
        """ Should emit debug message when response status is error. """
        try:
            self.function_to_test(**self.test_args)
        except FakeSystemExit:
            pass
        expected_output = textwrap.dedent("""\
                Upload failed: {status} {reason}
                """).format(status=self.status_code, reason=self.status_reason)
        self.assertIn(expected_output, sys.stdout.getvalue())

    def test_calls_sys_exit_when_response_status_error(self):
        """ Should call `sys.exit` when response status is error. """
        with testtools.ExpectedException(FakeSystemExit):
            self.function_to_test(**self.test_args)
        sys.exit.assert_called_with(self.expected_exit_status)

\f
def make_host_spec(username, host):
    """ Make an SSH host specification. """
    host_spec = host
    if username != "*":
        host_spec = "{user}@{fqdn}".format(user=username, fqdn=host)
    return host_spec


def make_remote_spec(username, host, dir_path):
    """ Make an SCP remote specification. """
    host_spec = make_host_spec(username, host)
    remote_spec = "{host}:{dir}".format(host=host_spec, dir=dir_path)
    return remote_spec


class ssh_channel_upload_TestCase(upload_TestCase):
    """ Base for test cases for upload over SSH channel. """

    function_to_test = NotImplemented

    scenarios = NotImplemented

    login_scenarios = [
            ('login-username', {
                'login': "lorem",
                }),
            ('login-wildcard', {
                'login': "*",
                }),
            ]

    stat_mode_scenarios = [
            ('stat-mode-default', {}),
            ('stat-mode-0620', {
                'stat_mode': 0o0620,
                'expected_ssh_chmod': True,
                }),
            ]

    def set_upload_file_modes(self):
        """ Set filesystem modes for upload files. """
        registry = FileDouble.get_registry_for_testcase(self)
        if hasattr(self, 'stat_mode'):
            for path in self.paths_to_upload:
                file_double = registry[path]
                file_double.stat_result = file_double.stat_result._replace(
                        st_mode=self.stat_mode)

    def set_ssh_chmod_subprocess_double(self):
        """ Set the ‘ssh … chmod’ test double for the subprocess. """
        command_file_path = "/usr/bin/ssh"
        argv = [os.path.basename(command_file_path)]
        argv.extend(self.expected_ssh_options)
        argv.append(make_host_spec(
                username=self.login, host=self.test_args['fqdn']))
        argv.extend(["chmod", "0644"])
        argv.extend(
                os.path.join(self.incoming_path, os.path.basename(path))
                for path in self.paths_to_upload)
        double = SubprocessDouble(command_file_path, argv=argv)
        double.register_for_testcase(self)
        check_call_scenario_name = getattr(
                self, 'ssh_chmod_check_call_scenario_name', "success")
        double.set_subprocess_check_call_scenario(check_call_scenario_name)
        self.ssh_chmod_subprocess_double = double

\f
class scp_upload_TestCase(ssh_channel_upload_TestCase):
    """ Test cases for `methods.scp.upload` function. """

    function_to_test = staticmethod(dput.methods.scp.upload)

    scenarios = NotImplemented

    ssh_config_scenarios = [
            ('ssh-opts-none', {
                'ssh_config_options': [],
                'expected_ssh_options': [],
                }),
            ('ssh-opts-one', {
                'ssh_config_options': ["foo"],
                'expected_ssh_options': ["-o", "foo"],
                }),
            ('ssh-opts-three', {
                'ssh_config_options': ["foo", "bar", "baz"],
                'expected_ssh_options': [
                    "-o", "foo", "-o", "bar", "-o", "baz"],
                }),
            ]

    def setUp(self):
        """ Set up test fixtures. """
        super(scp_upload_TestCase, self).setUp()

        patch_os_lstat(self)
        self.set_upload_file_modes()

        self.set_scp_subprocess_double()
        self.set_ssh_chmod_subprocess_double()

    def set_test_args(self):
        """ Set the arguments for the test call to the function. """
        self.test_args = dict(
                fqdn=self.getUniqueString(),
                login=self.login,
                incoming=self.incoming_path,
                files_to_upload=self.paths_to_upload,
                debug=None,
                compress=self.compress,
                ssh_config_options=self.ssh_config_options,
                progress=object(),
                )

    def set_scp_subprocess_double(self):
        """ Set the ‘scp’ test double for the subprocess. """
        command_file_path = "/usr/bin/scp"
        argv = [os.path.basename(command_file_path), "-p"]
        argv.extend(self.scp_compress_options)
        argv.extend(self.expected_ssh_options)
        argv.extend(self.paths_to_upload)
        argv.append(make_remote_spec(
                username=self.login, host=self.test_args['fqdn'],
                dir_path=self.incoming_path))
        double = SubprocessDouble(command_file_path, argv=argv)
        double.register_for_testcase(self)
        check_call_scenario_name = getattr(
                self, 'scp_subprocess_check_call_scenario_name', "success")
        double.set_subprocess_check_call_scenario(check_call_scenario_name)
        self.scp_subprocess_double = double


class scp_upload_ScpTestCase(scp_upload_TestCase):
    """ Test cases for `methods.scp.upload` function, with ‘scp’ command. """

    compress_scenarios = [
            ('compress-false', {
                'compress': False,
                'scp_compress_options': [],
                }),
            ('compress-true', {
                'compress': True,
                'scp_compress_options': ["-C"],
                }),
            ]

    scenarios = testscenarios.multiply_scenarios(
            upload_TestCase.files_scenarios,
            upload_TestCase.incoming_scenarios,
            ssh_channel_upload_TestCase.login_scenarios,
            ssh_channel_upload_TestCase.stat_mode_scenarios,
            compress_scenarios,
            scp_upload_TestCase.ssh_config_scenarios)

    def test_emits_debug_message_for_upload(self):
        """ Should emit debug message for files upload. """
        self.test_args['debug'] = True
        self.function_to_test(**self.test_args)
        expected_output = textwrap.dedent("""\
                D: Uploading with scp to {host}:{incoming}
                """).format(
                    host=make_host_spec(
                        username=self.login, host=self.test_args['fqdn']),
                    incoming=self.incoming_path)
        self.assertIn(expected_output, sys.stdout.getvalue())

    def test_calls_check_call_with_expected_scp_command(self):
        """ Should call `subprocess.check_call` with ‘scp’ command. """
        self.function_to_test(**self.test_args)
        expected_call = mock.call(self.scp_subprocess_double.argv)
        self.assertIn(expected_call, subprocess.check_call.mock_calls)

    def test_emits_error_message_when_scp_failure(self):
        """ Should emit error message when ‘scp’ command fails. """
        double = self.scp_subprocess_double
        double.set_subprocess_check_call_scenario("failure")
        try:
            self.function_to_test(**self.test_args)
        except FakeSystemExit:
            pass
        expected_output = "Error while uploading."
        self.assertIn(expected_output, sys.stdout.getvalue())

    def test_calls_sys_exit_when_scp_failure(self):
        """ Should call `sys.exit` when ‘scp’ command fails. """
        double = self.scp_subprocess_double
        double.set_subprocess_check_call_scenario("failure")
        with testtools.ExpectedException(FakeSystemExit):
            self.function_to_test(**self.test_args)
        sys.exit.assert_called_with(EXIT_STATUS_FAILURE)


class scp_upload_ChmodTestCase(scp_upload_TestCase):
    """ Test cases for `methods.scp.upload` function, with ‘ssh … chmod’. """

    files_scenarios = list(
            (scenario_name, scenario) for (scenario_name, scenario)
            in upload_TestCase.files_scenarios
            if scenario['paths_to_upload'])

    stat_mode_scenarios = list(
            (scenario_name, scenario) for (scenario_name, scenario)
            in ssh_channel_upload_TestCase.stat_mode_scenarios
            if 'expected_ssh_chmod' in scenario)

    compress_scenarios = [
            ('compress-false', {
                'compress': False,
                'scp_compress_options': [],
                }),
            ]

    scenarios = testscenarios.multiply_scenarios(
            files_scenarios,
            upload_TestCase.incoming_scenarios,
            ssh_channel_upload_TestCase.login_scenarios,
            stat_mode_scenarios,
            compress_scenarios,
            scp_upload_TestCase.ssh_config_scenarios)

    def test_emits_debug_message_for_fixing_permissions(self):
        """ Should emit debug message for fixing file permissions . """
        self.test_args['debug'] = True
        self.function_to_test(**self.test_args)
        expected_output = "D: Fixing some permissions"
        self.assertIn(expected_output, sys.stdout.getvalue())

    def test_calls_check_call_with_expected_ssh_chmod_command(self):
        """ Should call `subprocess.check_call` with ‘ssh … chmod’ command. """
        self.function_to_test(**self.test_args)
        expected_call = mock.call(self.ssh_chmod_subprocess_double.argv)
        self.assertIn(expected_call, subprocess.check_call.mock_calls)

    def test_emits_error_message_when_ssh_chmod_failure(self):
        """ Should emit error message when ‘ssh … chmod’ command fails. """
        double = self.ssh_chmod_subprocess_double
        double.set_subprocess_check_call_scenario("failure")
        try:
            self.function_to_test(**self.test_args)
        except FakeSystemExit:
            pass
        expected_output = "Error while fixing permissions."
        self.assertIn(expected_output, sys.stdout.getvalue())

    def test_calls_sys_exit_when_ssh_chmod_failure(self):
        """ Should call `sys.exit` when ‘ssh … chmod’ command fails. """
        double = self.ssh_chmod_subprocess_double
        double.set_subprocess_check_call_scenario("failure")
        with testtools.ExpectedException(FakeSystemExit):
            self.function_to_test(**self.test_args)
        sys.exit.assert_called_with(EXIT_STATUS_FAILURE)

\f
class rsync_upload_TestCase(ssh_channel_upload_TestCase):
    """ Test cases for `methods.rsync.upload` function. """

    function_to_test = staticmethod(dput.methods.rsync.upload)

    scenarios = testscenarios.multiply_scenarios(
            upload_TestCase.files_scenarios,
            upload_TestCase.incoming_scenarios,
            ssh_channel_upload_TestCase.login_scenarios,
            ssh_channel_upload_TestCase.stat_mode_scenarios)

    def setUp(self):
        """ Set up test fixtures. """
        super(rsync_upload_TestCase, self).setUp()

        self.set_rsync_subprocess_double()

        self.expected_ssh_options = []
        self.set_ssh_chmod_subprocess_double()

    def set_test_args(self):
        """ Set the arguments for the test call to the function. """
        self.test_args = dict(
                fqdn=self.getUniqueString(),
                login=self.login,
                incoming=self.incoming_path,
                files_to_upload=self.paths_to_upload,
                debug=False,
                dummy=object(),
                progress=object(),
                )

    def set_rsync_subprocess_double(self):
        """ Set the ‘rsync’ test double for the subprocess. """
        command_file_path = "/usr/bin/rsync"
        argv = [os.path.basename(command_file_path)]
        argv.extend(self.paths_to_upload)
        argv.extend([
                "--copy-links", "--progress", "--partial",
                "-zave", "ssh -x"])
        argv.append(make_remote_spec(
                username=self.login, host=self.test_args['fqdn'],
                dir_path=self.incoming_path))
        double = SubprocessDouble(command_file_path, argv=argv)
        double.register_for_testcase(self)
        check_call_scenario_name = getattr(
                self, 'rsync_check_call_scenario_name', "success")
        double.set_subprocess_check_call_scenario(check_call_scenario_name)
        self.rsync_subprocess_double = double

    def test_emits_debug_message_for_upload(self):
        """ Should emit debug message for files upload. """
        self.test_args['debug'] = True
        self.function_to_test(**self.test_args)
        expected_output = textwrap.dedent("""\
                D: Uploading with rsync to {host}:{incoming}
                """).format(
                    host=make_host_spec(
                        username=self.login, host=self.test_args['fqdn']),
                    incoming=self.incoming_path)
        self.assertIn(expected_output, sys.stdout.getvalue())

    def test_calls_check_call_with_expected_rsync_command(self):
        """ Should call `subprocess.check_call` with ‘rsync’ command. """
        self.function_to_test(**self.test_args)
        expected_call = mock.call(self.rsync_subprocess_double.argv)
        self.assertIn(expected_call, subprocess.check_call.mock_calls)

    def test_emits_error_message_when_rsync_failure(self):
        """ Should emit error message when ‘rsync’ command fails. """
        double = self.rsync_subprocess_double
        double.set_subprocess_check_call_scenario("failure")
        try:
            self.function_to_test(**self.test_args)
        except FakeSystemExit:
            pass
        expected_output = "Error while uploading."
        self.assertIn(expected_output, sys.stdout.getvalue())

    def test_calls_sys_exit_when_rsync_failure(self):
        """ Should call `sys.exit` when ‘rsync’ command fails. """
        double = self.rsync_subprocess_double
        double.set_subprocess_check_call_scenario("failure")
        with testtools.ExpectedException(FakeSystemExit):
            self.function_to_test(**self.test_args)
        sys.exit.assert_called_with(EXIT_STATUS_FAILURE)

    def test_emits_debug_message_for_fixing_permissions(self):
        """ Should emit debug message for fixing file permissions . """
        self.test_args['debug'] = True
        self.function_to_test(**self.test_args)
        expected_output = textwrap.dedent("""\
                D: Fixing file permissions with {host}
                """).format(
                    host=make_host_spec(
                        username=self.login, host=self.test_args['fqdn']))
        self.assertIn(expected_output, sys.stdout.getvalue())

    def test_calls_check_call_with_expected_ssh_chmod_command(self):
        """ Should call `subprocess.check_call` with ‘ssh … chmod’ command. """
        self.function_to_test(**self.test_args)
        expected_call = mock.call(list(self.ssh_chmod_subprocess_double.argv))
        self.assertIn(expected_call, subprocess.check_call.mock_calls)

    def test_emits_error_message_when_ssh_chmod_failure(self):
        """ Should emit error message when ‘ssh … chmod’ command fails. """
        double = self.ssh_chmod_subprocess_double
        double.set_subprocess_check_call_scenario("failure")
        try:
            self.function_to_test(**self.test_args)
        except FakeSystemExit:
            pass
        expected_output = "Error while fixing permission."
        self.assertIn(expected_output, sys.stdout.getvalue())

    def test_calls_sys_exit_when_ssh_chmod_failure(self):
        """ Should call `sys.exit` when ‘ssh … chmod’ command fails. """
        double = self.ssh_chmod_subprocess_double
        double.set_subprocess_check_call_scenario("failure")
        with testtools.ExpectedException(FakeSystemExit):
            self.function_to_test(**self.test_args)
        sys.exit.assert_called_with(EXIT_STATUS_FAILURE)

\f
# Local variables:
# coding: utf-8
# mode: python
# End:
# vim: fileencoding=utf-8 filetype=python :