# test/test_methods.py
# Part of ‘dput’, a Debian package upload toolkit.
#
# This is free software, and you are welcome to redistribute it under
# certain conditions; see the end of this file for copyright
# information, grant of license, and disclaimer of warranty.

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

import collections
import contextlib
import doctest
import ftplib
import getpass
import http.client
import importlib
import io
import itertools
import os
import os.path
import pkgutil
import stat
import subprocess
import sys
import tempfile
import textwrap
import unittest.mock
import urllib.parse
import urllib.request

import testscenarios
import testtools

import dput.dput
import dput.helper.dputhelper
import dput.methods
import dput.methods.ftp
import dput.methods.http
import dput.methods.https
import dput.methods.local
import dput.methods.rsync
import dput.methods.scp

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


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 = unittest.mock.Mock()
            module.__name__ = module_name
            module.upload = unittest.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().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 = unittest.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 = unittest.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

    @unittest.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())

    @unittest.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())

    @unittest.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)

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


class upload_BaseTestCase(
        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': "{}/".format(tempfile.mktemp()),
                }),
            ]

    def setUp(self):
        """ Set up test fixtures. """
        super().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 = unittest.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


class local_upload_TestCase(upload_BaseTestCase):
    """ Test cases for `methods.local.upload` function. """

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

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

    def setUp(self):
        """ Set up test fixtures. """
        super().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)


class ftp_upload_TestCase(upload_BaseTestCase):
    """ Test cases for `methods.ftp.upload` function. """

    scenarios = NotImplemented

    def setUp(self):
        """ Set up test fixtures. """
        super().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=make_fake_fqdn(self),
                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 = unittest.mock.MagicMock(name="FTP")

    def patch_ftplib_ftp(self):
        """ Patch `ftplib.FTP` class for this test case. """
        patcher = unittest.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_BaseTestCase.incoming_scenarios,
            upload_BaseTestCase.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_emits_no_debug_message_when_debug_false(self):
        """ Should emit no debug messages when `debug` is false. """
        self.test_args['debug'] = False
        dput.methods.ftp.upload(**self.test_args)
        debug_message_lines = [
                line for line in sys.stdout.getvalue().split()
                if line.startswith("D: ")]
        self.assertEqual([], debug_message_lines)

    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
        with contextlib.suppress(FakeSystemExit):
            dput.methods.ftp.upload(**self.test_args)
        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
        with contextlib.suppress(FakeSystemExit):
            dput.methods.ftp.upload(**self.test_args)
        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
        with contextlib.suppress(FakeSystemExit):
            dput.methods.ftp.upload(**self.test_args)
        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
        with contextlib.suppress(FakeSystemExit):
            dput.methods.ftp.upload(**self.test_args)
        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 = [
                unittest.mock.call(
                    "STOR {filename}".format(filename=os.path.basename(path)),
                    open(path, mode='b'),
                    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 = open(path, 'b')
            self.assertTrue(fake_file.closed)


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

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

    ftp_client_scenarios = [
            ('client-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_BaseTestCase.files_scenarios
            if scenario['paths_to_upload'])

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

    def test_emits_no_debug_message_when_debug_false(self):
        """ Should emit no debug messages when `debug` is false. """
        self.test_args['debug'] = False
        error = ftplib.error_perm("553 Exists")
        self.ftp_client.storbinary.side_effect = error
        dput.methods.ftp.upload(**self.test_args)
        debug_message_lines = [
                line for line in sys.stdout.getvalue().split()
                if line.startswith("D: ")]
        self.assertEqual([], debug_message_lines)

    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
        with contextlib.suppress(FakeSystemExit):
            dput.methods.ftp.upload(**self.test_args)
        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')
        with contextlib.suppress(EnvironmentError):
            dput.methods.ftp.upload(**self.test_args)
        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 test_emits_debug_message_from_quit_when_debug_true(self):
        """ Should emit text of error from `FTP.quit`. """
        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)
        expected_output = str(error)
        self.assertIn(expected_output, sys.stdout.getvalue())

    def test_emits_no_debug_message_from_quit_when_debug_false(self):
        """ Should not raise nor emit any error from `FTP.quit`. """
        self.test_args['debug'] = False
        error = ftplib.error_perm("504 Weird Stuff Happened")
        self.ftp_client.quit.side_effect = error
        dput.methods.ftp.upload(**self.test_args)
        unwanted_output = str(error)
        self.assertNotIn(unwanted_output, sys.stdout.getvalue())


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]
        fake_file = open(path, mode='b')
        expected_attributes = {
                'f': 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_BaseTestCase.incoming_scenarios,
            upload_BaseTestCase.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 = [
                unittest.mock.call(
                    unittest.mock.ANY, self.fake_filewithprogress,
                    unittest.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))


def make_mock_httprequest(
        *,
        method="GET",
        url=None,
        header_fields=None,
        body=None):
    """ Make an instrumented mock for `urllib.request.Request`. """
    request = unittest.mock.Mock(spec=urllib.request.Request)
    request.url = url
    request.headers = header_fields
    if body is None:
        body = ""
    body_data = body.encode(dput.methods.http.MESSAGE_BODY_ENCODING)
    request.data = body_data

    return request


reason_by_http_status = {
    http.client.OK: "Okay",
    http.client.NON_AUTHORITATIVE_INFORMATION: "Non-Authoritative Information",
    http.client.INTERNAL_SERVER_ERROR: "Internal Server Error",
    http.client.UNAUTHORIZED: "Unauthorized",
}


def make_mock_httpresponse(
        *,
        version="HTTP/1.1",
        status=http.client.OK,
        header_fields=None,
        body=None):
    """ Make an instrumented mock for `http.client.HTTPResponse`. """
    response = unittest.mock.Mock(spec=http.client.HTTPResponse)
    response.version = version
    response.status = status
    response.reason = reason_by_http_status[status]

    if header_fields is None:
        header_fields = {}
    response.headers = {name: value for (name, value) in header_fields.items()}
    if body is None:
        body = ""
    body_data = body.encode(dput.methods.http.MESSAGE_BODY_ENCODING)
    response.msg = body_data
    response.read.return_value = body_data
    response.getheader.side_effect = (lambda n: header_fields[n])
    response.getheaders.return_value = list(header_fields.items())

    return response


class http_upload_TestCase(upload_BaseTestCase):
    """ 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().setUp()

        self.set_response_header_fields()
        self.patch_connection()

        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=make_fake_fqdn(self),
                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 = urllib.parse.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_connection(self):
        """ Patch the connection instance. """
        mock_connection = unittest.mock.Mock(spec=http.client.HTTPConnection)
        mock_connection.getresponse.side_effect = itertools.chain(
                self.responses,
                itertools.repeat(self.responses[-1]))

        def putrequest(method, url, *, skip_accept_encoding=True):
            request = make_mock_httprequest(
                method=method,
                header_fields={},
                url=url)
            mock_connection.latest_request = request

        mock_connection.putrequest = unittest.mock.Mock(
            spec=http.client.HTTPConnection.putrequest,
            wraps=putrequest)

        def putheader(header, *values):
            values = list(values)
            field_value = "\r\n".join(values)
            mock_connection.latest_request.headers[header] = field_value

        mock_connection.putheader = unittest.mock.Mock(
            spec=http.client.HTTPConnection.putheader,
            wraps=putheader)

        def send(data):
            mock_connection.latest_request.data = data

        mock_connection.send = unittest.mock.Mock(
            spec=http.client.HTTPConnection.send,
            wraps=send)
        self.mock_connection = mock_connection
        self.mock_connection_class = unittest.mock.Mock(
                spec=http.client.HTTPConnection,
                return_value=self.mock_connection)
        self.test_args['connection_class'] = self.mock_connection_class


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

    response_scenarios = [
            ('okay', {
                'responses': [
                    make_mock_httpresponse(
                        status=http.client.OK,
                        body=textwrap.dedent("""\
                            Lorem ipsum, dolor sit amet.
                            """),
                        ),
                    ],
                }),
            ('chatter', {
                'responses': [
                    make_mock_httpresponse(
                        status=http.client.NON_AUTHORITATIVE_INFORMATION,
                        body=textwrap.dedent("""\
                            Lorem ipsum, dolor sit amet.
                            """),
                        ),
                    ],
                }),
            ]

    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_BaseTestCase.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_BaseTestCase.files_scenarios,
            size_scenarios,
            upload_BaseTestCase.incoming_scenarios,
            http_upload_TestCase.protocol_scenarios,
            http_upload_TestCase.login_scenarios,
            response_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 = self.mock_connection.latest_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_BaseTestCase.files_scenarios
            if scenario['paths_to_upload'])

    response_scenarios = [
            ('okay', {
                'responses': [
                    make_mock_httpresponse(
                        status=http.client.OK,
                        body=textwrap.dedent("""\
                            Lorem ipsum, dolor sit amet.
                            """),
                        ),
                    ],
                }),
            ]

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

    scenarios = testscenarios.multiply_scenarios(
            files_scenarios,
            progress_scenarios,
            upload_BaseTestCase.incoming_scenarios,
            http_upload_TestCase.protocol_scenarios,
            http_upload_TestCase.login_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_BaseTestCase.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]

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

    def test_emits_error_message_when_unknown_protocol(self):
        """ Should emit error message when unknown protocol. """
        with contextlib.suppress(FakeSystemExit):
            self.function_to_test(**self.test_args)
        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_BaseTestCase.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)

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

    def test_emits_error_message(self):
        """ Should emit error message when `os.stat` failure. """
        with contextlib.suppress(FakeSystemExit):
            self.function_to_test(**self.test_args)
        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_BaseTestCase.files_scenarios
            if scenario['paths_to_upload'])

    response_scenarios = [
            ('server-error body-text', {
                'responses': [
                    make_mock_httpresponse(
                        status=http.client.INTERNAL_SERVER_ERROR,
                        body=textwrap.dedent("""\
                            Lorem ipsum, dolor sit amet.
                            """),
                        ),
                    ],
                'expected_final_status': (
                    http.client.INTERNAL_SERVER_ERROR,
                    "Internal Server Error"),
                'expected_debug_output': textwrap.dedent("""\
                    Lorem ipsum, dolor sit amet.
                    """),
                'expected_exit_status': EXIT_STATUS_FAILURE,
                }),
            ('server-error body-empty', {
                'responses': [
                    make_mock_httpresponse(
                        status=http.client.INTERNAL_SERVER_ERROR,
                        body=""),
                    ],
                'expected_final_status': (
                    http.client.INTERNAL_SERVER_ERROR,
                    "Internal Server Error"),
                'expected_exit_status': EXIT_STATUS_FAILURE,
                }),
            ('authorization-rejected', {
                'responses': [
                    make_mock_httpresponse(
                        status=http.client.UNAUTHORIZED,
                        header_fields={
                            'WWW-Authenticate': "Basic realm=\"pellentesque\"",
                            },
                        body=textwrap.dedent("""\
                            Lorem ipsum, dolor sit amet.
                            """),
                        ),
                    make_mock_httpresponse(
                        status=http.client.UNAUTHORIZED,
                        header_fields={
                            'WWW-Authenticate': "Basic realm=\"pellentesque\"",
                            },
                        body=textwrap.dedent("""\
                            Lorem ipsum, dolor sit amet.
                            """),
                        ),
                    ],
                'expected_final_status': (
                    http.client.UNAUTHORIZED, "Unauthorized"),
                'expected_debug_output': textwrap.dedent("""\
                    Lorem ipsum, dolor sit amet.
                    """),
                'expected_exit_status': EXIT_STATUS_FAILURE,
                }),
            ]

    scenarios = testscenarios.multiply_scenarios(
            files_scenarios,
            upload_BaseTestCase.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. """
        if self.expected_final_status[0] == http.client.UNAUTHORIZED:
            self.skipTest("FIXME: can't see how this ever worked?")
        with contextlib.suppress(FakeSystemExit):
            self.function_to_test(**self.test_args)
        (code, reason) = self.expected_final_status
        expected_output = textwrap.dedent("""\
                Upload failed: {status} {reason}
                """).format(status=code, reason=reason)
        self.assertIn(expected_output, sys.stdout.getvalue())

    def test_emits_error_response_body_text_when_debug_true(self):
        """ Should emit error response body text when debug is true. """
        self.test_args['debug'] = True
        if not hasattr(self, 'expected_debug_output'):
            self.skipTest("no response body expected")
        if self.expected_final_status[0] == http.client.UNAUTHORIZED:
            self.skipTest("FIXME: can't see how this ever worked?")
        with contextlib.suppress(FakeSystemExit):
            self.function_to_test(**self.test_args)
        expected_output = self.expected_debug_output
        self.assertIn(expected_output, sys.stdout.getvalue())

    def test_emits_no_response_body_text_when_debug_false(self):
        """ Should emit no response body text when ‘debug’ is false. """
        self.test_args['debug'] = False
        if not hasattr(self, 'expected_debug_output'):
            self.skipTest("no response body expected")
        if self.expected_final_status[0] == http.client.UNAUTHORIZED:
            self.skipTest("FIXME: can't see how this ever worked?")
        with contextlib.suppress(FakeSystemExit):
            self.function_to_test(**self.test_args)
        unwanted_output = self.expected_debug_output
        self.assertNotIn(unwanted_output, sys.stdout.getvalue())

    def test_calls_sys_exit_when_response_status_error(self):
        """ Should call `sys.exit` when response status is error. """
        if self.expected_final_status[0] == http.client.UNAUTHORIZED:
            self.skipTest("FIXME: can't see how this ever worked?")
        with testtools.ExpectedException(FakeSystemExit):
            self.function_to_test(**self.test_args)
        sys.exit.assert_called_with(self.expected_exit_status)


class http_upload_ResponseUnauthorizedTestCase(http_upload_TestCase):
    """ Unauthorized test cases for `methods.http.upload` function. """

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

    response_scenarios = [
            ('authorization-accepted', {
                'responses': [
                    make_mock_httpresponse(
                        status=http.client.UNAUTHORIZED,
                        header_fields={
                            'WWW-Authenticate': "Basic realm=\"pellentesque\"",
                            },
                        body=textwrap.dedent("""\
                            Lorem ipsum, dolor sit amet.
                            """),
                        ),
                    make_mock_httpresponse(
                        status=http.client.OK,
                        body=textwrap.dedent("""\
                            Lorem ipsum, dolor sit amet.
                            """),
                        ),
                    ],
                'expected_final_status': (http.client.OK, "Okay"),
                'expected_exit_status': EXIT_STATUS_SUCCESS,
                }),
            ]

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

    def test_calls_getpass(self):
        """ Should call `getpass.getpass` when status is unauthorized. """
        self.skipTest("FIXME: can't see how this ever worked?")
        self.function_to_test(**self.test_args)
        getpass.getpass.assert_called_with()


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_BaseTestCase):
    """ 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,
                }),
            ('stat-mode-0640', {
                'stat_mode': 0o0644,
                'expected_ssh_chmod': False,
                }),
            ]

    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


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().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=make_fake_fqdn(self),
                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_BaseTestCase.files_scenarios,
            upload_BaseTestCase.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 = unittest.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")
        with contextlib.suppress(FakeSystemExit):
            self.function_to_test(**self.test_args)
        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_BaseTestCase.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 scenario.get('expected_ssh_chmod', False))

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

    scenarios = testscenarios.multiply_scenarios(
            files_scenarios,
            upload_BaseTestCase.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 = unittest.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")
        with contextlib.suppress(FakeSystemExit):
            self.function_to_test(**self.test_args)
        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)


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_BaseTestCase.files_scenarios,
            upload_BaseTestCase.incoming_scenarios,
            ssh_channel_upload_TestCase.login_scenarios,
            ssh_channel_upload_TestCase.stat_mode_scenarios)

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

        self.expected_rsync_options = dput.methods.rsync.rsync_options
        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=make_fake_fqdn(self),
                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(self.expected_rsync_options)
        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 = unittest.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")
        with contextlib.suppress(FakeSystemExit):
            self.function_to_test(**self.test_args)
        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 = unittest.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")
        with contextlib.suppress(FakeSystemExit):
            self.function_to_test(**self.test_args)
        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)


# Copyright © 2015–2024 Ben Finney <bignose@debian.org>
#
# 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.

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