test/test_methods.py

   1 # -*- coding: utf-8; -*-
   2 #
   3 # test/test_methods.py
   4 # Part of ‘dput’, a Debian package upload toolkit.
   5 #
   6 # This is free software, and you are welcome to redistribute it under
   7 # certain conditions; see the end of this file for copyright
   8 # information, grant of license, and disclaimer of warranty.
   9
  10 """ Unit tests for upload method behaviour. """
  11
  12 from __future__ import (absolute_import, unicode_literals)
  13
  14 import collections
  15 import doctest
  16 import ftplib
  17 import getpass
  18 import importlib
  19 import io
  20 import os
  21 import os.path
  22 import pkgutil
  23 import stat
  24 import subprocess
  25 import sys
  26 import tempfile
  27 import textwrap
  28
  29 if sys.version_info >= (3, 3):
  30     import urllib.parse as urlparse
  31 elif sys.version_info >= (3, 0):
  32     raise RuntimeError("Python 3 earlier than 3.3 is not supported.")
  33 else:
  34     import urlparse
  35
  36 import httpretty
  37 import testscenarios
  38 import testtools
  39
  40 import dput.dput
  41 import dput.helper.dputhelper
  42 import dput.methods
  43 import dput.methods.ftp
  44 import dput.methods.http
  45 import dput.methods.https
  46 import dput.methods.local
  47 import dput.methods.rsync
  48 import dput.methods.scp
  49
  50 from .helper import (
  51         ARG_ANY,
  52         EXIT_STATUS_FAILURE,
  53         FakeSystemExit,
  54         FileDouble,
  55         SubprocessDouble,
  56         mock,
  57         patch_os_lstat,
  58         patch_os_stat,
  59         patch_subprocess_check_call,
  60         patch_system_interfaces,
  61         setup_file_double_behaviour,
  62         )
  63 from .test_dputhelper import (
  64         patch_filewithprogress,
  65         )
  66
  67 \f
  68 class import_upload_functions_TestCase(
  69         testscenarios.WithScenarios,
  70         testtools.TestCase):
  71     """ Test cases for `import_upload_functions` function. """
  72
  73     scenarios = [
  74             ('empty', {
  75                 'module_names': [],
  76                 }),
  77             ('one', {
  78                 'module_names': ["foo"],
  79                 }),
  80             ('three', {
  81                 'module_names': ["foo", "bar", "baz"],
  82                 }),
  83             ]
  84
  85     for (scenario_name, scenario) in scenarios:
  86         modules_by_name = collections.OrderedDict()
  87         iter_modules_result = []
  88         for module_name in scenario['module_names']:
  89             module = mock.Mock()
  90             module.__name__ = module_name
  91             module.upload = mock.Mock()
  92             modules_by_name[module_name] = module
  93             module_entry = (module, module_name, False)
  94             iter_modules_result.append(module_entry)
  95         scenario['modules_by_name'] = modules_by_name
  96         scenario['iter_modules_result'] = iter_modules_result
  97     del scenario_name, scenario
  98
  99     def setUp(self):
 100         """ Set up test fixtures. """
 101         super(import_upload_functions_TestCase, self).setUp()
 102         patch_system_interfaces(self)
 103
 104         self.patch_import_functions()
 105
 106     def patch_import_functions(self):
 107         """ Patch import functions used by the function. """
 108         self.patch_pkgutil_iter_modules()
 109         self.patch_importlib_import_module()
 110
 111     def patch_pkgutil_iter_modules(self):
 112         """ Patch `pkgutil.iter_modules` function for this test case. """
 113         func_patcher = mock.patch.object(
 114                 pkgutil, "iter_modules", autospec=True)
 115         func_patcher.start()
 116         self.addCleanup(func_patcher.stop)
 117         pkgutil.iter_modules.return_value = self.iter_modules_result
 118
 119     def patch_importlib_import_module(self):
 120         """ Patch `importlib.import_module` function for this test case. """
 121         func_patcher = mock.patch.object(
 122                 importlib, "import_module", autospec=True)
 123         func_patcher.start()
 124         self.addCleanup(func_patcher.stop)
 125
 126         def fake_import_module(full_name):
 127             module_name = full_name.split(".")[-1]
 128             module = self.modules_by_name[module_name]
 129             return module
 130
 131         importlib.import_module.side_effect = fake_import_module
 132
 133     @mock.patch.object(dput.dput, 'debug', new=True)
 134     def test_emits_debug_message_for_modules_found(self):
 135         """ Should emit a debug message for the modules found. """
 136         expected_message = "D: modules_found: {names!r}".format(
 137                 names=self.module_names)
 138         dput.dput.import_upload_functions()
 139         self.assertIn(expected_message, sys.stdout.getvalue())
 140
 141     @mock.patch.object(dput.dput, 'debug', new=True)
 142     def test_emits_debug_message_for_each_import(self):
 143         """ Should emit a debug message for each module imported. """
 144         dput.dput.import_upload_functions()
 145         for (module_name, module) in self.modules_by_name.items():
 146             expected_message = "D: Module: {name} ({module!r})".format(
 147                     name=module_name, module=module)
 148             self.assertIn(expected_message, sys.stdout.getvalue())
 149
 150     @mock.patch.object(dput.dput, 'debug', new=True)
 151     def test_emits_debug_message_for_each_upload_method(self):
 152         """ Should emit a debug message for each upload method. """
 153         dput.dput.import_upload_functions()
 154         for module_name in self.module_names:
 155             expected_message = "D: Method name: {name}".format(
 156                     name=module_name)
 157             self.assertIn(expected_message, sys.stdout.getvalue())
 158
 159     def test_returns_expected_function_mapping(self):
 160         """ Should return expected mapping of upload functions. """
 161         result = dput.dput.import_upload_functions()
 162         expected_result = {
 163                 name: self.modules_by_name[name].upload
 164                 for name in self.module_names}
 165         self.assertEqual(expected_result, result)
 166
 167 \f
 168 def patch_getpass_getpass(testcase):
 169     """ Patch the `getpass.getpass` function for the test case. """
 170     func_patcher = mock.patch.object(getpass, "getpass", autospec=True)
 171     func_patcher.start()
 172     testcase.addCleanup(func_patcher.stop)
 173
 174 \f
 175 class upload_TestCase(
 176         testscenarios.WithScenarios,
 177         testtools.TestCase):
 178     """ Base for test cases for method modules `upload` functions. """
 179
 180     files_scenarios = [
 181             ('file-list-empty', {
 182                 'paths_to_upload': [],
 183                 }),
 184             ('file-list-one', {
 185                 'paths_to_upload': [tempfile.mktemp()],
 186                 }),
 187             ('file-list-three', {
 188                 'paths_to_upload': [tempfile.mktemp() for __ in range(3)],
 189                 }),
 190             ]
 191
 192     check_call_scenarios = [
 193             ('check-call-success', {
 194                 'check_call_scenario_name': 'success',
 195                 }),
 196             ('check-call-failure', {
 197                 'check_call_scenario_name': 'failure',
 198                 'check_call_error': subprocess.CalledProcessError,
 199                 }),
 200             ]
 201
 202     check_call_success_scenarios = [
 203             (name, params) for (name, params) in check_call_scenarios
 204             if 'check_call_error' not in params]
 205
 206     incoming_scenarios = [
 207             ('incoming-simple', {
 208                 'incoming_path': tempfile.mktemp(),
 209                 }),
 210             ('incoming-no-leading-slash', {
 211                 'incoming_path': tempfile.mktemp().lstrip("/"),
 212                 }),
 213             ('incoming-has-trailing-slash', {
 214                 'incoming_path': tempfile.mktemp() + "/",
 215                 }),
 216             ]
 217
 218     def setUp(self):
 219         """ Set up test fixtures. """
 220         super(upload_TestCase, self).setUp()
 221         patch_system_interfaces(self)
 222
 223         patch_subprocess_check_call(self)
 224
 225         patch_os_stat(self)
 226         self.set_file_doubles()
 227         setup_file_double_behaviour(self)
 228
 229         patch_filewithprogress(self)
 230
 231         self.set_test_args()
 232
 233     def set_file_doubles(self):
 234         """ Set the file doubles for this test case. """
 235         for path in self.paths_to_upload:
 236             fake_file = getattr(self, 'fake_file', None)
 237             double = FileDouble(path, fake_file)
 238             double.set_os_stat_scenario(
 239                     getattr(self, 'os_stat_scenario_name', "okay"))
 240             double.register_for_testcase(self)
 241
 242             func_patcher = mock.patch.object(
 243                     double.fake_file, "close", autospec=True)
 244             func_patcher.start()
 245             self.addCleanup(func_patcher.stop)
 246
 247     def set_test_args(self):
 248         """ Set the arguments for the test call to the function. """
 249         raise NotImplementedError
 250
 251     def get_command_args_from_latest_check_call(self):
 252         """ Get command line arguments from latest `subprocess.check_call`. """
 253         latest_call = subprocess.check_call.call_args
 254         (args, kwargs) = latest_call
 255         command_args = args[0]
 256         return command_args
 257
 258 \f
 259 class local_upload_TestCase(upload_TestCase):
 260     """ Test cases for `methods.local.upload` function. """
 261
 262     scenarios = testscenarios.multiply_scenarios(
 263             upload_TestCase.incoming_scenarios,
 264             upload_TestCase.files_scenarios,
 265             upload_TestCase.check_call_success_scenarios)
 266
 267     command_file_path = os.path.join("/usr/bin", "install")
 268
 269     def setUp(self):
 270         """ Set up test fixtures. """
 271         super(local_upload_TestCase, self).setUp()
 272
 273         self.set_subprocess_double()
 274
 275     def set_test_args(self):
 276         """ Set the arguments for the test call to the function. """
 277         self.test_args = dict(
 278                 fqdn=object(),
 279                 login=object(),
 280                 incoming=self.incoming_path,
 281                 files_to_upload=self.paths_to_upload,
 282                 debug=None,
 283                 compress=object(),
 284                 progress=object(),
 285                 )
 286
 287     def set_subprocess_double(self):
 288         """ Set the test double for the subprocess. """
 289         argv = [self.command_file_path, "-m", ARG_ANY, ARG_ANY]
 290         double = SubprocessDouble(self.command_file_path, argv=argv)
 291         double.register_for_testcase(self)
 292         double.set_subprocess_check_call_scenario(
 293                 self.check_call_scenario_name)
 294         self.subprocess_double = double
 295
 296     def test_calls_check_call_with_install_command(self):
 297         """ Should call `subprocess.check_call` to invoke ‘install’. """
 298         dput.methods.local.upload(**self.test_args)
 299         command_args = self.get_command_args_from_latest_check_call()
 300         expected_command = self.command_file_path
 301         self.assertEqual(expected_command, command_args[0])
 302
 303     def test_calls_check_call_with_mode_option_in_command(self):
 304         """ Should call `subprocess.check_call`, invoke command with mode. """
 305         dput.methods.local.upload(**self.test_args)
 306         command_args = self.get_command_args_from_latest_check_call()
 307         expected_mode = (
 308                 stat.S_IRUSR | stat.S_IWUSR
 309                 | stat.S_IRGRP
 310                 | stat.S_IROTH)
 311         expected_mode_text = "{mode:04o}".format(mode=expected_mode)[-3:]
 312         expected_option_args = ["-m", expected_mode_text]
 313         self.assertEqual(expected_option_args, command_args[1:3])
 314
 315     def test_calls_check_call_with_file_paths_in_command(self):
 316         """ Should call `subprocess.check_call` with file paths in command. """
 317         dput.methods.local.upload(**self.test_args)
 318         command_args = self.get_command_args_from_latest_check_call()
 319         self.assertEqual(self.paths_to_upload, command_args[3:-1])
 320
 321     def test_calls_check_call_with_incoming_path_as_final_arg(self):
 322         """ Should call `subprocess.check_call` with incoming path. """
 323         dput.methods.local.upload(**self.test_args)
 324         command_args = self.get_command_args_from_latest_check_call()
 325         self.assertEqual(self.incoming_path, command_args[-1])
 326
 327     def test_emits_debug_message_for_upload_command(self):
 328         """ Should emit a debug message for the upload command. """
 329         self.test_args['debug'] = True
 330         dput.methods.local.upload(**self.test_args)
 331         expected_message = textwrap.dedent("""\
 332                 D: Uploading with cp to {dir_path}
 333                 D: ...
 334                 """).format(dir_path=self.incoming_path)
 335         self.assertThat(
 336                 sys.stdout.getvalue(),
 337                 testtools.matchers.DocTestMatches(
 338                     expected_message, flags=doctest.ELLIPSIS))
 339
 340     def test_calls_sys_exit_if_check_call_returns_nonzero(self):
 341         """ Should call `sys.exit` if `subprocess.check_call` fails. """
 342         self.subprocess_double.set_subprocess_check_call_scenario('failure')
 343         with testtools.ExpectedException(FakeSystemExit):
 344             dput.methods.local.upload(**self.test_args)
 345         expected_output = textwrap.dedent("""\
 346                 Error while uploading.
 347                 """)
 348         self.assertIn(expected_output, sys.stdout.getvalue())
 349         sys.exit.assert_called_with(EXIT_STATUS_FAILURE)
 350
 351 \f
 352 class ftp_upload_TestCase(upload_TestCase):
 353     """ Test cases for `methods.ftp.upload` function. """
 354
 355     scenarios = NotImplemented
 356
 357     def setUp(self):
 358         """ Set up test fixtures. """
 359         super(ftp_upload_TestCase, self).setUp()
 360
 361         self.set_ftp_client()
 362         self.patch_ftplib_ftp()
 363
 364         patch_getpass_getpass(self)
 365         self.fake_password = self.getUniqueString()
 366         getpass.getpass.return_value = self.fake_password
 367         if not hasattr(self, 'expected_password'):
 368             self.expected_password = self.fake_password
 369
 370     def set_test_args(self):
 371         """ Set the arguments for the test call to the function. """
 372         if not hasattr(self, 'progress_type'):
 373             self.progress_type = 0
 374         self.test_args = dict(
 375                 fqdn=self.getUniqueString(),
 376                 login=self.login,
 377                 incoming=self.incoming_path,
 378                 files_to_upload=self.paths_to_upload,
 379                 debug=False,
 380                 ftp_mode=self.ftp_mode,
 381                 progress=self.progress_type,
 382                 port=object(),
 383                 )
 384
 385     def set_ftp_client(self):
 386         """ Set the FTP client double. """
 387         self.ftp_client = mock.MagicMock(name="FTP")
 388
 389     def patch_ftplib_ftp(self):
 390         """ Patch `ftplib.FTP` class for this test case. """
 391         patcher = mock.patch.object(ftplib, "FTP", autospec=True)
 392         patcher.start()
 393         self.addCleanup(patcher.stop)
 394
 395         ftplib.FTP.return_value = self.ftp_client
 396
 397
 398 class ftp_upload_NormalFilesTestCase(ftp_upload_TestCase):
 399     """ Test cases for `methods.ftp.upload` function, upload normal files. """
 400
 401     login_scenarios = [
 402             ('anonymous', {
 403                 'login': "anonymous",
 404                 'expected_password': "dput@packages.debian.org",
 405                 }),
 406             ('username', {
 407                 'login': "lorem",
 408                 }),
 409             ]
 410
 411     ftp_client_scenarios = [
 412             ('default', {
 413                 'ftp_mode': False,
 414                 }),
 415             ('passive-mode', {
 416                 'ftp_mode': True,
 417                 }),
 418             ]
 419
 420     scenarios = testscenarios.multiply_scenarios(
 421             upload_TestCase.incoming_scenarios,
 422             upload_TestCase.files_scenarios,
 423             login_scenarios, ftp_client_scenarios)
 424
 425     def test_emits_debug_message_for_connect(self):
 426         """ Should emit debug message for successful connect. """
 427         self.test_args['debug'] = True
 428         dput.methods.ftp.upload(**self.test_args)
 429         expected_fqdn = self.test_args['fqdn']
 430         expected_output = textwrap.dedent("""\
 431                 D: FTP-Connection to host: {fqdn}
 432                 """).format(fqdn=expected_fqdn)
 433         self.assertIn(expected_output, sys.stdout.getvalue())
 434
 435     def test_calls_ftp_connect_with_expected_args(self):
 436         """ Should call `FTP.connect` with expected args. """
 437         dput.methods.ftp.upload(**self.test_args)
 438         expected_args = (
 439                 self.test_args['fqdn'],
 440                 self.test_args['port'],
 441                 )
 442         self.ftp_client.connect.assert_called_with(*expected_args)
 443
 444     def test_emits_error_message_when_ftp_connect_error(self):
 445         """ Should emit error message when `FTP.connect` raises error. """
 446         self.ftp_client.connect.side_effect = ftplib.error_temp
 447         try:
 448             dput.methods.ftp.upload(**self.test_args)
 449         except FakeSystemExit:
 450             pass
 451         expected_output = "Connection failed, aborting"
 452         self.assertIn(expected_output, sys.stdout.getvalue())
 453
 454     def test_calls_sys_exit_when_ftp_connect_permission_error(self):
 455         """ Should call `sys.exit` when `FTP.connect` raises error. """
 456         self.ftp_client.connect.side_effect = ftplib.error_temp
 457         with testtools.ExpectedException(FakeSystemExit):
 458             dput.methods.ftp.upload(**self.test_args)
 459         sys.exit.assert_called_with(EXIT_STATUS_FAILURE)
 460
 461     def test_calls_ftp_login_with_expected_args(self):
 462         """ Should call `FTP.login` with expected args. """
 463         dput.methods.ftp.upload(**self.test_args)
 464         expected_args = (
 465                 self.test_args['login'],
 466                 self.expected_password,
 467                 )
 468         self.ftp_client.login.assert_called_with(*expected_args)
 469
 470     def test_emits_error_message_when_ftp_login_permission_error(self):
 471         """ Should emit error message when `FTP.login` permission error. """
 472         self.ftp_client.login.side_effect = ftplib.error_perm
 473         try:
 474             dput.methods.ftp.upload(**self.test_args)
 475         except FakeSystemExit:
 476             pass
 477         expected_output = "Wrong Password"
 478         self.assertIn(expected_output, sys.stdout.getvalue())
 479
 480     def test_calls_sys_exit_when_ftp_login_permission_error(self):
 481         """ Should call `sys.exit` when `FTP.login` permission error. """
 482         self.ftp_client.login.side_effect = ftplib.error_perm
 483         with testtools.ExpectedException(FakeSystemExit):
 484             dput.methods.ftp.upload(**self.test_args)
 485         sys.exit.assert_called_with(EXIT_STATUS_FAILURE)
 486
 487     def test_emits_error_message_when_ftp_login_eof_error(self):
 488         """ Should emit error message when `FTP.login` EOF error. """
 489         self.ftp_client.login.side_effect = EOFError
 490         try:
 491             dput.methods.ftp.upload(**self.test_args)
 492         except FakeSystemExit:
 493             pass
 494         expected_output = "Server closed the connection"
 495         self.assertIn(expected_output, sys.stdout.getvalue())
 496
 497     def test_calls_sys_exit_when_ftp_login_eof_error(self):
 498         """ Should call `sys.exit` when `FTP.login` EOF error. """
 499         self.ftp_client.login.side_effect = EOFError
 500         with testtools.ExpectedException(FakeSystemExit):
 501             dput.methods.ftp.upload(**self.test_args)
 502         sys.exit.assert_called_with(EXIT_STATUS_FAILURE)
 503
 504     def test_calls_ftp_set_pasv_with_expected_args(self):
 505         """ Should call `FTP.set_pasv` with expected args. """
 506         dput.methods.ftp.upload(**self.test_args)
 507         expected_mode = bool(self.test_args['ftp_mode'])
 508         expected_args = (expected_mode,)
 509         self.ftp_client.set_pasv.assert_called_with(*expected_args)
 510
 511     def test_calls_ftp_cwd_with_expected_args(self):
 512         """ Should call `FTP.cwd` with expected args. """
 513         dput.methods.ftp.upload(**self.test_args)
 514         expected_path = self.incoming_path
 515         expected_args = (expected_path,)
 516         self.ftp_client.cwd.assert_called_with(*expected_args)
 517
 518     def test_emits_debug_message_for_cwd(self):
 519         """ Should emit debug message for successful `FTP.cwd`. """
 520         self.test_args['debug'] = True
 521         dput.methods.ftp.upload(**self.test_args)
 522         expected_output = textwrap.dedent("""\
 523                 D: Directory to upload to: {path}
 524                 """).format(path=self.incoming_path)
 525         self.assertIn(expected_output, sys.stdout.getvalue())
 526
 527     def test_emits_error_message_when_destination_directory_not_found(self):
 528         """ Should emit error message when destination directory not found. """
 529         error = ftplib.error_perm("550 Not Found")
 530         self.ftp_client.cwd.side_effect = error
 531         try:
 532             dput.methods.ftp.upload(**self.test_args)
 533         except FakeSystemExit:
 534             pass
 535         expected_output = "Directory to upload to does not exist."
 536         self.assertIn(expected_output, sys.stdout.getvalue())
 537
 538     def test_calls_sys_exit_when_ftp_cwd_permission_error(self):
 539         """ Should call `sys.exit` when `FTP.cwd` permission error. """
 540         error = ftplib.error_perm("550 Not Found")
 541         self.ftp_client.cwd.side_effect = error
 542         with testtools.ExpectedException(FakeSystemExit):
 543             dput.methods.ftp.upload(**self.test_args)
 544         sys.exit.assert_called_with(EXIT_STATUS_FAILURE)
 545
 546     def test_propagates_exception_when_ftp_cwd_permission_error(self):
 547         """ Should call `sys.exit` when `FTP.cwd` permission error. """
 548         error = ftplib.error_perm("500 Bad Stuff Happened")
 549         self.ftp_client.cwd.side_effect = error
 550         with testtools.ExpectedException(error.__class__):
 551             dput.methods.ftp.upload(**self.test_args)
 552
 553     def test_propagates_exception_when_ftp_cwd_eof_error(self):
 554         """ Should call `sys.exit` when `FTP.cwd` EOF error. """
 555         error = EOFError()
 556         self.ftp_client.cwd.side_effect = error
 557         with testtools.ExpectedException(error.__class__):
 558             dput.methods.ftp.upload(**self.test_args)
 559
 560     def test_emits_debug_message_for_each_file(self):
 561         """ Should emit debug message for each file to upload. """
 562         self.test_args['debug'] = True
 563         dput.methods.ftp.upload(**self.test_args)
 564         expected_output = "".join(textwrap.dedent("""\
 565                 D: Uploading File: {path}
 566                   Uploading {filename}: done.
 567                 """).format(path=path, filename=os.path.basename(path))
 568                 for path in self.paths_to_upload)
 569         self.assertIn(expected_output, sys.stdout.getvalue())
 570
 571     def test_calls_ftp_storbinary_for_each_file(self):
 572         """ Should call `FTP.storbinary` for each file to upload. """
 573         dput.methods.ftp.upload(**self.test_args)
 574         registry = FileDouble.get_registry_for_testcase(self)
 575         expected_blocksize = 1024
 576         expected_calls = [
 577                 mock.call(
 578                     "STOR {filename}".format(filename=os.path.basename(path)),
 579                     registry[path].fake_file, expected_blocksize)
 580                 for path in self.paths_to_upload]
 581         self.ftp_client.storbinary.assert_has_calls(
 582                 expected_calls, any_order=True)
 583
 584     def test_calls_close_for_each_file(self):
 585         """ Should call `file.close` for each file to upload. """
 586         dput.methods.ftp.upload(**self.test_args)
 587         registry = FileDouble.get_registry_for_testcase(self)
 588         for path in self.paths_to_upload:
 589             fake_file = registry[path].fake_file
 590             fake_file.close.assert_called_with()
 591
 592
 593 class ftp_upload_ErrorTestCase(ftp_upload_TestCase):
 594     """ Test cases for `methods.ftp.upload` function, error conditions. """
 595
 596     login_scenarios = [
 597             ('anonymous', {
 598                 'login': "anonymous",
 599                 'expected_password': "dput@packages.debian.org",
 600                 }),
 601             ]
 602
 603     ftp_client_scenarios = [
 604             ('default', {
 605                 'ftp_mode': False,
 606                 }),
 607             ]
 608
 609     progress_scenarios = [
 610             ('progress-type-0', {
 611                 'progress_type': 0,
 612                 }),
 613             ('progress-type-1', {
 614                 'progress_type': 1,
 615                 }),
 616             ('progress-type-2', {
 617                 'progress_type': 2,
 618                 }),
 619             ]
 620
 621     files_scenarios = list(
 622             (scenario_name, scenario) for (scenario_name, scenario)
 623             in upload_TestCase.files_scenarios
 624             if scenario['paths_to_upload'])
 625
 626     scenarios = testscenarios.multiply_scenarios(
 627             upload_TestCase.incoming_scenarios,
 628             files_scenarios,
 629             login_scenarios, ftp_client_scenarios, progress_scenarios)
 630
 631     def test_emits_warning_when_remote_file_exists(self):
 632         """ Should emit a warning message when remote file exists. """
 633         error = ftplib.error_perm("553 Exists")
 634         self.ftp_client.storbinary.side_effect = error
 635         dput.methods.ftp.upload(**self.test_args)
 636         for path in self.paths_to_upload:
 637             expected_output = textwrap.dedent("""\
 638                     Leaving existing {path} on the server and continuing
 639                     """).format(path=os.path.basename(path))
 640             self.expectThat(
 641                     sys.stdout.getvalue(),
 642                     testtools.matchers.Contains(expected_output))
 643
 644     def test_omits_sys_exit_when_remote_file_exists(self):
 645         """ Should omit call to `sys.exit` when remote file exists. """
 646         error = ftplib.error_perm("553 Exists")
 647         self.ftp_client.storbinary.side_effect = error
 648         dput.methods.ftp.upload(**self.test_args)
 649         self.assertFalse(sys.exit.called)
 650
 651     def test_emits_error_message_when_storbinary_failure(self):
 652         """ Should emit an error message when `FTP.storbinary` failure. """
 653         error = ftplib.error_perm("504 Weird Stuff Happened")
 654         self.ftp_client.storbinary.side_effect = error
 655         try:
 656             dput.methods.ftp.upload(**self.test_args)
 657         except FakeSystemExit:
 658             pass
 659         for path in self.paths_to_upload[:1]:
 660             expected_output = (
 661                     "Note: This error might indicate a problem with"
 662                     " your passive_ftp setting.\n")
 663             self.expectThat(
 664                     sys.stdout.getvalue(),
 665                     testtools.matchers.Contains(expected_output))
 666
 667     def test_calls_sys_exit_when_storbinary_failure(self):
 668         """ Should call `sys.exit` when `FTP.storbinary` failure. """
 669         error = ftplib.error_perm("504 Weird Stuff Happened")
 670         self.ftp_client.storbinary.side_effect = error
 671         with testtools.ExpectedException(FakeSystemExit):
 672             dput.methods.ftp.upload(**self.test_args)
 673         sys.exit.assert_called_with(EXIT_STATUS_FAILURE)
 674
 675     def test_emits_debug_message_when_open_failure(self):
 676         """ Should emit a debug message when `builtins.open` failure. """
 677         self.test_args['debug'] = True
 678         registry = FileDouble.get_registry_for_testcase(self)
 679         for path in self.paths_to_upload:
 680             double = registry[path]
 681             double.set_open_scenario('nonexist')
 682         try:
 683             dput.methods.ftp.upload(**self.test_args)
 684         except EnvironmentError:
 685             pass
 686         expected_output = (
 687                 "D: Should exit silently now, but"
 688                 " will throw exception for debug.")
 689         self.assertIn(expected_output, sys.stdout.getvalue())
 690
 691     def test_propagates_error_from_storbinary_for_debug(self):
 692         """ Should propagate error from `FTP.storbinary` when debug. """
 693         self.test_args['debug'] = True
 694         error = ftplib.error_perm("504 Weird Stuff Happened")
 695         self.ftp_client.storbinary.side_effect = error
 696         with testtools.ExpectedException(error.__class__):
 697             dput.methods.ftp.upload(**self.test_args)
 698
 699     def test_propagates_error_from_quit_for_debug(self):
 700         """ Should propagate error from `FTP.quit` when debug. """
 701         self.test_args['debug'] = True
 702         error = ftplib.error_perm("504 Weird Stuff Happened")
 703         self.ftp_client.quit.side_effect = error
 704         with testtools.ExpectedException(error.__class__):
 705             dput.methods.ftp.upload(**self.test_args)
 706
 707
 708 def make_expected_filewithprogress_attributes_by_path(testcase, attrs):
 709     """ Make a mapping from path to expected FileWithProgress attribs. """
 710     expected_attributes_by_path = {}
 711     registry = FileDouble.get_registry_for_testcase(testcase)
 712     for path in testcase.paths_to_upload:
 713         file_double = registry[path]
 714         expected_attributes = {
 715                 'f': file_double.fake_file,
 716                 'size': file_double.stat_result.st_size,
 717                 }
 718         expected_attributes.update(attrs)
 719         expected_attributes_by_path[path] = expected_attributes
 720
 721     return expected_attributes_by_path
 722
 723
 724 class ftp_upload_ProgressTestCase(ftp_upload_TestCase):
 725     """ Test cases for `methods.ftp.upload` function, with progress meter. """
 726
 727     login_scenarios = [
 728             ('anonymous', {
 729                 'login': "anonymous",
 730                 'expected_password': "dput@packages.debian.org",
 731                 }),
 732             ]
 733
 734     ftp_client_scenarios = [
 735             ('default', {
 736                 'ftp_mode': False,
 737                 }),
 738             ]
 739
 740     progress_scenarios = [
 741             ('progress-type-1', {
 742                 'progress_type': 1,
 743                 }),
 744             ('progress-type-2', {
 745                 'progress_type': 2,
 746                 }),
 747             ]
 748
 749     scenarios = testscenarios.multiply_scenarios(
 750             upload_TestCase.incoming_scenarios,
 751             upload_TestCase.files_scenarios,
 752             login_scenarios, ftp_client_scenarios, progress_scenarios)
 753
 754     def test_calls_storbinary_with_filewithprogress(self):
 755         """ Should use a `FileWithProgress` to call `FTP.storbinary`. """
 756         dput.methods.ftp.upload(**self.test_args)
 757         expected_calls = [
 758                 mock.call(
 759                     mock.ANY, self.fake_filewithprogress,
 760                     mock.ANY)
 761                 for path in self.paths_to_upload]
 762         self.ftp_client.storbinary.assert_has_calls(
 763                 expected_calls, any_order=True)
 764
 765     def test_filewithprogress_has_expected_attributes(self):
 766         """ Should have expected attributes on the `FileWithProgress`. """
 767         expected_attributes_by_path = (
 768                 make_expected_filewithprogress_attributes_by_path(
 769                     self, {'ptype': self.progress_type}))
 770         dput.methods.ftp.upload(**self.test_args)
 771         for call in self.ftp_client.storbinary.mock_calls:
 772             (__, call_args, call_kwargs) = call
 773             (__, stor_file, __) = call_args
 774             path = stor_file.f.name
 775             expected_attributes = expected_attributes_by_path[path]
 776             stor_file_attributes = {
 777                     name: getattr(stor_file, name)
 778                     for name in expected_attributes}
 779             self.expectThat(
 780                     expected_attributes,
 781                     testtools.matchers.Equals(stor_file_attributes))
 782
 783     def test_filewithprogress_has_sentinel_size_when_stat_failure(self):
 784         """ Should have sentinel `size` value when `os.stat` failure. """
 785         expected_attributes_by_path = (
 786                 make_expected_filewithprogress_attributes_by_path(
 787                     self, {'size': -1}))
 788         registry = FileDouble.get_registry_for_testcase(self)
 789         for path in self.paths_to_upload:
 790             double = registry[path]
 791             double.set_os_stat_scenario('notfound_error')
 792         dput.methods.ftp.upload(**self.test_args)
 793         for call in self.ftp_client.storbinary.mock_calls:
 794             (__, call_args, call_kwargs) = call
 795             (__, stor_file, __) = call_args
 796             path = stor_file.f.name
 797             expected_attributes = expected_attributes_by_path[path]
 798             stor_file_attributes = {
 799                     name: getattr(stor_file, name)
 800                     for name in expected_attributes}
 801             self.expectThat(
 802                     expected_attributes,
 803                     testtools.matchers.Equals(stor_file_attributes))
 804
 805     def test_emits_debug_message_when_stat_failure(self):
 806         """ Should have sentinel `size` value when `os.stat` failure. """
 807         self.test_args['debug'] = True
 808         registry = FileDouble.get_registry_for_testcase(self)
 809         for path in self.paths_to_upload:
 810             double = registry[path]
 811             double.set_os_stat_scenario('notfound_error')
 812         dput.methods.ftp.upload(**self.test_args)
 813         for path in self.paths_to_upload:
 814             expected_output = textwrap.dedent("""\
 815                     D: Determining size of file '{path}' failed
 816                     """).format(path=path)
 817             self.expectThat(
 818                     sys.stdout.getvalue(),
 819                     testtools.matchers.Contains(expected_output))
 820
 821 \f
 822 class http_upload_TestCase(upload_TestCase):
 823     """ Base for test cases for `methods.http.upload` function. """
 824
 825     scenarios = NotImplemented
 826
 827     protocol_scenarios = [
 828             ('http', {
 829                 'function_to_test': dput.methods.http.upload,
 830                 'protocol': "http",
 831                 'protocol_version': "HTTP/1.0",
 832                 }),
 833             ('https', {
 834                 'function_to_test': dput.methods.https.upload,
 835                 'protocol': "https",
 836                 'protocol_version': "HTTP/1.0",
 837                 }),
 838             ]
 839
 840     login_scenarios = [
 841             ('username', {
 842                 'login': "lorem",
 843                 }),
 844             ]
 845
 846     def setUp(self):
 847         """ Set up test fixtures. """
 848         super(http_upload_TestCase, self).setUp()
 849
 850         httpretty.enable()
 851         self.addCleanup(httpretty.disable)
 852
 853         self.set_response_header_fields()
 854         self.patch_put_requests()
 855
 856         patch_getpass_getpass(self)
 857         self.fake_password = self.getUniqueString()
 858         getpass.getpass.return_value = self.fake_password
 859         if not hasattr(self, 'expected_password'):
 860             self.expected_password = self.fake_password
 861
 862     def set_test_args(self):
 863         """ Set the arguments for the test call to the function. """
 864         if not hasattr(self, 'progress_type'):
 865             self.progress_type = 0
 866         self.test_args = dict(
 867                 fqdn=self.getUniqueString(),
 868                 login=self.login,
 869                 incoming=self.incoming_path,
 870                 files_to_upload=self.paths_to_upload,
 871                 debug=False,
 872                 dummy=object(),
 873                 progress=self.progress_type,
 874                 )
 875         if self.function_to_test is dput.methods.http.upload:
 876             self.test_args['protocol'] = self.protocol
 877
 878     def make_upload_uri(self, file_name):
 879         """ Make the URI for a file for upload. """
 880         uri = urlparse.urlunsplit([
 881                 self.protocol, self.test_args['fqdn'],
 882                 os.path.join(os.path.sep, self.incoming_path, file_name),
 883                 None, None])
 884         return uri
 885
 886     def set_response_header_fields(self):
 887         """ Set the header fields for the HTTP response. """
 888         if not hasattr(self, 'response_header_fields'):
 889             self.response_header_fields = {}
 890
 891     def patch_put_requests(self):
 892         """ Patch the HTTP PUT requests. """
 893         self.path_by_request_uri = {}
 894         for path in self.paths_to_upload:
 895             upload_uri = self.make_upload_uri(os.path.basename(path))
 896             self.path_by_request_uri[upload_uri] = path
 897             response_body = ""
 898             httpretty.register_uri(
 899                     httpretty.PUT, upload_uri,
 900                     status=self.status_code, body=response_body)
 901
 902
 903 class http_upload_SuccessTestCase(http_upload_TestCase):
 904     """ Success test cases for `methods.http.upload` function. """
 905
 906     response_scenarios = [
 907             ('okay', {
 908                 'status_code': 200,
 909                 'status_reason': "Okay",
 910                 }),
 911             ('chatter', {
 912                 'status_code': 203,
 913                 'status_reason': "Non-Authoritative Information",
 914                 }),
 915             ]
 916
 917     auth_scenarios = [
 918             ('auth-accepted', {
 919                 'auth_response_status_code': 200,
 920                 'auth_response_status_reason': "Okay",
 921                 }),
 922             ]
 923
 924     size_scenarios = [
 925             ('size-empty', {
 926                 'fake_file': io.BytesIO(),
 927                 }),
 928             ('size-1k', {
 929                 'fake_file': io.BytesIO(
 930                     b"Lorem ipsum, dolor sit amet.___\n" * 32),
 931                 }),
 932             ('size-100k', {
 933                 'fake_file': io.BytesIO(
 934                     b"Lorem ipsum, dolor sit amet.___\n" * 3200),
 935                 }),
 936             ]
 937
 938     incoming_scenarios = list(upload_TestCase.incoming_scenarios)
 939     for (scenario_name, scenario) in incoming_scenarios:
 940         scenario['expected_url_path_prefix'] = os.path.join(
 941                 os.path.sep, scenario['incoming_path'])
 942     del scenario_name, scenario
 943
 944     scenarios = testscenarios.multiply_scenarios(
 945             upload_TestCase.files_scenarios,
 946             size_scenarios,
 947             upload_TestCase.incoming_scenarios,
 948             http_upload_TestCase.protocol_scenarios,
 949             http_upload_TestCase.login_scenarios,
 950             response_scenarios, auth_scenarios)
 951
 952     def test_emits_debug_message_for_upload(self):
 953         """ Should emit debug message for upload. """
 954         self.test_args['debug'] = True
 955         self.function_to_test(**self.test_args)
 956         for path in self.paths_to_upload:
 957             expected_uri = self.make_upload_uri(os.path.basename(path))
 958             expected_output = textwrap.dedent("""\
 959                     D: HTTP-PUT to URL: {uri}
 960                     """).format(uri=expected_uri)
 961             self.expectThat(
 962                     sys.stdout.getvalue(),
 963                     testtools.matchers.Contains(expected_output))
 964
 965     def test_request_has_expected_fields(self):
 966         """ Should send request with expected fields in header. """
 967         if not self.paths_to_upload:
 968             self.skipTest("No files to upload")
 969         self.function_to_test(**self.test_args)
 970         registry = FileDouble.get_registry_for_testcase(self)
 971         path = self.paths_to_upload[-1]
 972         double = registry[path]
 973         request = httpretty.last_request()
 974         expected_fields = {
 975                 'User-Agent': "dput",
 976                 'Connection': "close",
 977                 'Content-Length': "{size:d}".format(
 978                     size=len(double.fake_file.getvalue())),
 979                 }
 980         for (name, value) in expected_fields.items():
 981             self.expectThat(
 982                     request.headers.get(name),
 983                     testtools.matchers.Equals(value))
 984
 985
 986 class http_upload_ProgressTestCase(http_upload_TestCase):
 987     """ Test cases for `methods.http.upload` function, with progress meter. """
 988
 989     files_scenarios = list(
 990             (scenario_name, scenario) for (scenario_name, scenario)
 991             in upload_TestCase.files_scenarios
 992             if scenario['paths_to_upload'])
 993
 994     response_scenarios = [
 995             ('okay', {
 996                 'status_code': 200,
 997                 'status_reason': "Okay",
 998                 }),
 999             ]
1000
1001     progress_scenarios = [
1002             ('progress-type-1', {
1003                 'progress_type': 1,
1004                 }),
1005             ('progress-type-2', {
1006                 'progress_type': 2,
1007                 }),
1008             ]
1009
1010     scenarios = testscenarios.multiply_scenarios(
1011             files_scenarios,
1012             progress_scenarios,
1013             upload_TestCase.incoming_scenarios,
1014             http_upload_TestCase.protocol_scenarios,
1015             http_upload_TestCase.login_scenarios,
1016             http_upload_SuccessTestCase.auth_scenarios,
1017             response_scenarios)
1018
1019     def test_filewithprogress_has_expected_attributes(self):
1020         """ Should have expected attributes on the `FileWithProgress`. """
1021         expected_attributes_by_path = (
1022                 make_expected_filewithprogress_attributes_by_path(
1023                     self, {'ptype': self.progress_type}))
1024         self.function_to_test(**self.test_args)
1025         path = self.paths_to_upload[-1]
1026         expected_attributes = expected_attributes_by_path[path]
1027         fake_file_attributes = {
1028                 name: getattr(self.fake_filewithprogress, name)
1029                 for name in expected_attributes}
1030         self.expectThat(
1031                 expected_attributes,
1032                 testtools.matchers.Equals(fake_file_attributes))
1033
1034
1035 class http_upload_UnknownProtocolTestCase(http_upload_TestCase):
1036     """ Test cases for `methods.http.upload` function, unknown protocol. """
1037
1038     files_scenarios = list(
1039             (scenario_name, scenario) for (scenario_name, scenario)
1040             in upload_TestCase.files_scenarios
1041             if scenario['paths_to_upload'])
1042
1043     protocol_scenarios = [
1044             ('protocol-bogus', {
1045                 'function_to_test': dput.methods.http.upload,
1046                 'protocol': "b0gUs",
1047                 'protocol_version': "b0gUs",
1048                 'expected_exit_status': EXIT_STATUS_FAILURE,
1049                 }),
1050             ]
1051
1052     response_scenarios = [
1053             (scenario_name, scenario) for (scenario_name, scenario)
1054             in http_upload_SuccessTestCase.response_scenarios
1055             if scenario['status_code'] == 200]
1056
1057     scenarios = testscenarios.multiply_scenarios(
1058             files_scenarios,
1059             upload_TestCase.incoming_scenarios,
1060             protocol_scenarios,
1061             http_upload_TestCase.login_scenarios,
1062             response_scenarios,
1063             http_upload_SuccessTestCase.auth_scenarios)
1064
1065     def test_emits_error_message_when_unknown_protocol(self):
1066         """ Should emit error message when unknown protocol. """
1067         try:
1068             self.function_to_test(**self.test_args)
1069         except FakeSystemExit:
1070             pass
1071         expected_output = "Wrong protocol for upload "
1072         self.assertIn(expected_output, sys.stderr.getvalue())
1073
1074     def test_calls_sys_exit_when_unknown_protocol(self):
1075         """ Should call `sys.exit` when unknown protocol. """
1076         with testtools.ExpectedException(FakeSystemExit):
1077             self.function_to_test(**self.test_args)
1078         sys.exit.assert_called_with(self.expected_exit_status)
1079
1080
1081 class http_upload_FileStatFailureTestCase(http_upload_TestCase):
1082     """ Test cases for `methods.http.upload` function, `os.stat` failure. """
1083
1084     files_scenarios = list(
1085             (scenario_name, scenario) for (scenario_name, scenario)
1086             in upload_TestCase.files_scenarios
1087             if scenario['paths_to_upload'])
1088
1089     os_stat_scenarios = [
1090             ('os-stat-notfound', {
1091                 'os_stat_scenario_name': "notfound_error",
1092                 'expected_exit_status': EXIT_STATUS_FAILURE,
1093                 }),
1094             ('os-stat-denied', {
1095                 'os_stat_scenario_name': "denied_error",
1096                 'expected_exit_status': EXIT_STATUS_FAILURE,
1097                 }),
1098             ]
1099
1100     response_scenarios = list(
1101             (scenario_name, scenario) for (scenario_name, scenario)
1102             in http_upload_SuccessTestCase.response_scenarios
1103             if scenario['status_code'] == 200)
1104
1105     scenarios = testscenarios.multiply_scenarios(
1106             files_scenarios,
1107             os_stat_scenarios,
1108             upload_TestCase.incoming_scenarios,
1109             http_upload_TestCase.protocol_scenarios,
1110             http_upload_TestCase.login_scenarios,
1111             response_scenarios,
1112             http_upload_SuccessTestCase.auth_scenarios)
1113
1114     def test_emits_error_message(self):
1115         """ Should emit error message when `os.stat` failure. """
1116         try:
1117             self.function_to_test(**self.test_args)
1118         except FakeSystemExit:
1119             pass
1120         expected_output = textwrap.dedent("""\
1121                 Determining size of file '{path}' failed
1122                 """).format(path=self.paths_to_upload[0])
1123         self.assertIn(expected_output, sys.stderr.getvalue())
1124
1125     def test_calls_sys_exit_with_expected_exit_status(self):
1126         """ Should call `sys.exit` with expected exit status. """
1127         with testtools.ExpectedException(FakeSystemExit):
1128             self.function_to_test(**self.test_args)
1129         sys.exit.assert_called_with(self.expected_exit_status)
1130
1131
1132 class http_upload_ResponseErrorTestCase(http_upload_TestCase):
1133     """ Error test cases for `methods.http.upload` function. """
1134
1135     files_scenarios = list(
1136             (scenario_name, scenario) for (scenario_name, scenario)
1137             in upload_TestCase.files_scenarios
1138             if scenario['paths_to_upload'])
1139
1140     response_scenarios = [
1141             ('server-error', {
1142                 'status_code': 500,
1143                 'status_reason': "Internal Server Error",
1144                 'auth_response_status_code': 200,
1145                 'auth_response_status_reason': "Okay",
1146                 'expected_exit_status': EXIT_STATUS_FAILURE,
1147                 }),
1148             ]
1149
1150     scenarios = testscenarios.multiply_scenarios(
1151             files_scenarios,
1152             upload_TestCase.incoming_scenarios,
1153             http_upload_TestCase.protocol_scenarios,
1154             http_upload_TestCase.login_scenarios,
1155             response_scenarios)
1156
1157     def test_emits_error_message_when_response_status_error(self):
1158         """ Should emit debug message when response status is error. """
1159         try:
1160             self.function_to_test(**self.test_args)
1161         except FakeSystemExit:
1162             pass
1163         expected_output = textwrap.dedent("""\
1164                 Upload failed: {status} {reason}
1165                 """).format(status=self.status_code, reason=self.status_reason)
1166         self.assertIn(expected_output, sys.stdout.getvalue())
1167
1168     def test_calls_sys_exit_when_response_status_error(self):
1169         """ Should call `sys.exit` when response status is error. """
1170         with testtools.ExpectedException(FakeSystemExit):
1171             self.function_to_test(**self.test_args)
1172         sys.exit.assert_called_with(self.expected_exit_status)
1173
1174 \f
1175 def make_host_spec(username, host):
1176     """ Make an SSH host specification. """
1177     host_spec = host
1178     if username != "*":
1179         host_spec = "{user}@{fqdn}".format(user=username, fqdn=host)
1180     return host_spec
1181
1182
1183 def make_remote_spec(username, host, dir_path):
1184     """ Make an SCP remote specification. """
1185     host_spec = make_host_spec(username, host)
1186     remote_spec = "{host}:{dir}".format(host=host_spec, dir=dir_path)
1187     return remote_spec
1188
1189
1190 class ssh_channel_upload_TestCase(upload_TestCase):
1191     """ Base for test cases for upload over SSH channel. """
1192
1193     function_to_test = NotImplemented
1194
1195     scenarios = NotImplemented
1196
1197     login_scenarios = [
1198             ('login-username', {
1199                 'login': "lorem",
1200                 }),
1201             ('login-wildcard', {
1202                 'login': "*",
1203                 }),
1204             ]
1205
1206     stat_mode_scenarios = [
1207             ('stat-mode-default', {}),
1208             ('stat-mode-0620', {
1209                 'stat_mode': 0o0620,
1210                 'expected_ssh_chmod': True,
1211                 }),
1212             ]
1213
1214     def set_upload_file_modes(self):
1215         """ Set filesystem modes for upload files. """
1216         registry = FileDouble.get_registry_for_testcase(self)
1217         if hasattr(self, 'stat_mode'):
1218             for path in self.paths_to_upload:
1219                 file_double = registry[path]
1220                 file_double.stat_result = file_double.stat_result._replace(
1221                         st_mode=self.stat_mode)
1222
1223     def set_ssh_chmod_subprocess_double(self):
1224         """ Set the ‘ssh … chmod’ test double for the subprocess. """
1225         command_file_path = "/usr/bin/ssh"
1226         argv = [os.path.basename(command_file_path)]
1227         argv.extend(self.expected_ssh_options)
1228         argv.append(make_host_spec(
1229                 username=self.login, host=self.test_args['fqdn']))
1230         argv.extend(["chmod", "0644"])
1231         argv.extend(
1232                 os.path.join(self.incoming_path, os.path.basename(path))
1233                 for path in self.paths_to_upload)
1234         double = SubprocessDouble(command_file_path, argv=argv)
1235         double.register_for_testcase(self)
1236         check_call_scenario_name = getattr(
1237                 self, 'ssh_chmod_check_call_scenario_name', "success")
1238         double.set_subprocess_check_call_scenario(check_call_scenario_name)
1239         self.ssh_chmod_subprocess_double = double
1240
1241 \f
1242 class scp_upload_TestCase(ssh_channel_upload_TestCase):
1243     """ Test cases for `methods.scp.upload` function. """
1244
1245     function_to_test = staticmethod(dput.methods.scp.upload)
1246
1247     scenarios = NotImplemented
1248
1249     ssh_config_scenarios = [
1250             ('ssh-opts-none', {
1251                 'ssh_config_options': [],
1252                 'expected_ssh_options': [],
1253                 }),
1254             ('ssh-opts-one', {
1255                 'ssh_config_options': ["foo"],
1256                 'expected_ssh_options': ["-o", "foo"],
1257                 }),
1258             ('ssh-opts-three', {
1259                 'ssh_config_options': ["foo", "bar", "baz"],
1260                 'expected_ssh_options': [
1261                     "-o", "foo", "-o", "bar", "-o", "baz"],
1262                 }),
1263             ]
1264
1265     def setUp(self):
1266         """ Set up test fixtures. """
1267         super(scp_upload_TestCase, self).setUp()
1268
1269         patch_os_lstat(self)
1270         self.set_upload_file_modes()
1271
1272         self.set_scp_subprocess_double()
1273         self.set_ssh_chmod_subprocess_double()
1274
1275     def set_test_args(self):
1276         """ Set the arguments for the test call to the function. """
1277         self.test_args = dict(
1278                 fqdn=self.getUniqueString(),
1279                 login=self.login,
1280                 incoming=self.incoming_path,
1281                 files_to_upload=self.paths_to_upload,
1282                 debug=None,
1283                 compress=self.compress,
1284                 ssh_config_options=self.ssh_config_options,
1285                 progress=object(),
1286                 )
1287
1288     def set_scp_subprocess_double(self):
1289         """ Set the ‘scp’ test double for the subprocess. """
1290         command_file_path = "/usr/bin/scp"
1291         argv = [os.path.basename(command_file_path), "-p"]
1292         argv.extend(self.scp_compress_options)
1293         argv.extend(self.expected_ssh_options)
1294         argv.extend(self.paths_to_upload)
1295         argv.append(make_remote_spec(
1296                 username=self.login, host=self.test_args['fqdn'],
1297                 dir_path=self.incoming_path))
1298         double = SubprocessDouble(command_file_path, argv=argv)
1299         double.register_for_testcase(self)
1300         check_call_scenario_name = getattr(
1301                 self, 'scp_subprocess_check_call_scenario_name', "success")
1302         double.set_subprocess_check_call_scenario(check_call_scenario_name)
1303         self.scp_subprocess_double = double
1304
1305
1306 class scp_upload_ScpTestCase(scp_upload_TestCase):
1307     """ Test cases for `methods.scp.upload` function, with ‘scp’ command. """
1308
1309     compress_scenarios = [
1310             ('compress-false', {
1311                 'compress': False,
1312                 'scp_compress_options': [],
1313                 }),
1314             ('compress-true', {
1315                 'compress': True,
1316                 'scp_compress_options': ["-C"],
1317                 }),
1318             ]
1319
1320     scenarios = testscenarios.multiply_scenarios(
1321             upload_TestCase.files_scenarios,
1322             upload_TestCase.incoming_scenarios,
1323             ssh_channel_upload_TestCase.login_scenarios,
1324             ssh_channel_upload_TestCase.stat_mode_scenarios,
1325             compress_scenarios,
1326             scp_upload_TestCase.ssh_config_scenarios)
1327
1328     def test_emits_debug_message_for_upload(self):
1329         """ Should emit debug message for files upload. """
1330         self.test_args['debug'] = True
1331         self.function_to_test(**self.test_args)
1332         expected_output = textwrap.dedent("""\
1333                 D: Uploading with scp to {host}:{incoming}
1334                 """).format(
1335                     host=make_host_spec(
1336                         username=self.login, host=self.test_args['fqdn']),
1337                     incoming=self.incoming_path)
1338         self.assertIn(expected_output, sys.stdout.getvalue())
1339
1340     def test_calls_check_call_with_expected_scp_command(self):
1341         """ Should call `subprocess.check_call` with ‘scp’ command. """
1342         self.function_to_test(**self.test_args)
1343         expected_call = mock.call(self.scp_subprocess_double.argv)
1344         self.assertIn(expected_call, subprocess.check_call.mock_calls)
1345
1346     def test_emits_error_message_when_scp_failure(self):
1347         """ Should emit error message when ‘scp’ command fails. """
1348         double = self.scp_subprocess_double
1349         double.set_subprocess_check_call_scenario("failure")
1350         try:
1351             self.function_to_test(**self.test_args)
1352         except FakeSystemExit:
1353             pass
1354         expected_output = "Error while uploading."
1355         self.assertIn(expected_output, sys.stdout.getvalue())
1356
1357     def test_calls_sys_exit_when_scp_failure(self):
1358         """ Should call `sys.exit` when ‘scp’ command fails. """
1359         double = self.scp_subprocess_double
1360         double.set_subprocess_check_call_scenario("failure")
1361         with testtools.ExpectedException(FakeSystemExit):
1362             self.function_to_test(**self.test_args)
1363         sys.exit.assert_called_with(EXIT_STATUS_FAILURE)
1364
1365
1366 class scp_upload_ChmodTestCase(scp_upload_TestCase):
1367     """ Test cases for `methods.scp.upload` function, with ‘ssh … chmod’. """
1368
1369     files_scenarios = list(
1370             (scenario_name, scenario) for (scenario_name, scenario)
1371             in upload_TestCase.files_scenarios
1372             if scenario['paths_to_upload'])
1373
1374     stat_mode_scenarios = list(
1375             (scenario_name, scenario) for (scenario_name, scenario)
1376             in ssh_channel_upload_TestCase.stat_mode_scenarios
1377             if 'expected_ssh_chmod' in scenario)
1378
1379     compress_scenarios = [
1380             ('compress-false', {
1381                 'compress': False,
1382                 'scp_compress_options': [],
1383                 }),
1384             ]
1385
1386     scenarios = testscenarios.multiply_scenarios(
1387             files_scenarios,
1388             upload_TestCase.incoming_scenarios,
1389             ssh_channel_upload_TestCase.login_scenarios,
1390             stat_mode_scenarios,
1391             compress_scenarios,
1392             scp_upload_TestCase.ssh_config_scenarios)
1393
1394     def test_emits_debug_message_for_fixing_permissions(self):
1395         """ Should emit debug message for fixing file permissions . """
1396         self.test_args['debug'] = True
1397         self.function_to_test(**self.test_args)
1398         expected_output = "D: Fixing some permissions"
1399         self.assertIn(expected_output, sys.stdout.getvalue())
1400
1401     def test_calls_check_call_with_expected_ssh_chmod_command(self):
1402         """ Should call `subprocess.check_call` with ‘ssh … chmod’ command. """
1403         self.function_to_test(**self.test_args)
1404         expected_call = mock.call(self.ssh_chmod_subprocess_double.argv)
1405         self.assertIn(expected_call, subprocess.check_call.mock_calls)
1406
1407     def test_emits_error_message_when_ssh_chmod_failure(self):
1408         """ Should emit error message when ‘ssh … chmod’ command fails. """
1409         double = self.ssh_chmod_subprocess_double
1410         double.set_subprocess_check_call_scenario("failure")
1411         try:
1412             self.function_to_test(**self.test_args)
1413         except FakeSystemExit:
1414             pass
1415         expected_output = "Error while fixing permissions."
1416         self.assertIn(expected_output, sys.stdout.getvalue())
1417
1418     def test_calls_sys_exit_when_ssh_chmod_failure(self):
1419         """ Should call `sys.exit` when ‘ssh … chmod’ command fails. """
1420         double = self.ssh_chmod_subprocess_double
1421         double.set_subprocess_check_call_scenario("failure")
1422         with testtools.ExpectedException(FakeSystemExit):
1423             self.function_to_test(**self.test_args)
1424         sys.exit.assert_called_with(EXIT_STATUS_FAILURE)
1425
1426 \f
1427 class rsync_upload_TestCase(ssh_channel_upload_TestCase):
1428     """ Test cases for `methods.rsync.upload` function. """
1429
1430     function_to_test = staticmethod(dput.methods.rsync.upload)
1431
1432     scenarios = testscenarios.multiply_scenarios(
1433             upload_TestCase.files_scenarios,
1434             upload_TestCase.incoming_scenarios,
1435             ssh_channel_upload_TestCase.login_scenarios,
1436             ssh_channel_upload_TestCase.stat_mode_scenarios)
1437
1438     def setUp(self):
1439         """ Set up test fixtures. """
1440         super(rsync_upload_TestCase, self).setUp()
1441
1442         self.set_rsync_subprocess_double()
1443
1444         self.expected_ssh_options = []
1445         self.set_ssh_chmod_subprocess_double()
1446
1447     def set_test_args(self):
1448         """ Set the arguments for the test call to the function. """
1449         self.test_args = dict(
1450                 fqdn=self.getUniqueString(),
1451                 login=self.login,
1452                 incoming=self.incoming_path,
1453                 files_to_upload=self.paths_to_upload,
1454                 debug=False,
1455                 dummy=object(),
1456                 progress=object(),
1457                 )
1458
1459     def set_rsync_subprocess_double(self):
1460         """ Set the ‘rsync’ test double for the subprocess. """
1461         command_file_path = "/usr/bin/rsync"
1462         argv = [os.path.basename(command_file_path)]
1463         argv.extend(self.paths_to_upload)
1464         argv.extend([
1465                 "--copy-links", "--progress", "--partial",
1466                 "-zave", "ssh -x"])
1467         argv.append(make_remote_spec(
1468                 username=self.login, host=self.test_args['fqdn'],
1469                 dir_path=self.incoming_path))
1470         double = SubprocessDouble(command_file_path, argv=argv)
1471         double.register_for_testcase(self)
1472         check_call_scenario_name = getattr(
1473                 self, 'rsync_check_call_scenario_name', "success")
1474         double.set_subprocess_check_call_scenario(check_call_scenario_name)
1475         self.rsync_subprocess_double = double
1476
1477     def test_emits_debug_message_for_upload(self):
1478         """ Should emit debug message for files upload. """
1479         self.test_args['debug'] = True
1480         self.function_to_test(**self.test_args)
1481         expected_output = textwrap.dedent("""\
1482                 D: Uploading with rsync to {host}:{incoming}
1483                 """).format(
1484                     host=make_host_spec(
1485                         username=self.login, host=self.test_args['fqdn']),
1486                     incoming=self.incoming_path)
1487         self.assertIn(expected_output, sys.stdout.getvalue())
1488
1489     def test_calls_check_call_with_expected_rsync_command(self):
1490         """ Should call `subprocess.check_call` with ‘rsync’ command. """
1491         self.function_to_test(**self.test_args)
1492         expected_call = mock.call(self.rsync_subprocess_double.argv)
1493         self.assertIn(expected_call, subprocess.check_call.mock_calls)
1494
1495     def test_emits_error_message_when_rsync_failure(self):
1496         """ Should emit error message when ‘rsync’ command fails. """
1497         double = self.rsync_subprocess_double
1498         double.set_subprocess_check_call_scenario("failure")
1499         try:
1500             self.function_to_test(**self.test_args)
1501         except FakeSystemExit:
1502             pass
1503         expected_output = "Error while uploading."
1504         self.assertIn(expected_output, sys.stdout.getvalue())
1505
1506     def test_calls_sys_exit_when_rsync_failure(self):
1507         """ Should call `sys.exit` when ‘rsync’ command fails. """
1508         double = self.rsync_subprocess_double
1509         double.set_subprocess_check_call_scenario("failure")
1510         with testtools.ExpectedException(FakeSystemExit):
1511             self.function_to_test(**self.test_args)
1512         sys.exit.assert_called_with(EXIT_STATUS_FAILURE)
1513
1514     def test_emits_debug_message_for_fixing_permissions(self):
1515         """ Should emit debug message for fixing file permissions . """
1516         self.test_args['debug'] = True
1517         self.function_to_test(**self.test_args)
1518         expected_output = textwrap.dedent("""\
1519                 D: Fixing file permissions with {host}
1520                 """).format(
1521                     host=make_host_spec(
1522                         username=self.login, host=self.test_args['fqdn']))
1523         self.assertIn(expected_output, sys.stdout.getvalue())
1524
1525     def test_calls_check_call_with_expected_ssh_chmod_command(self):
1526         """ Should call `subprocess.check_call` with ‘ssh … chmod’ command. """
1527         self.function_to_test(**self.test_args)
1528         expected_call = mock.call(list(self.ssh_chmod_subprocess_double.argv))
1529         self.assertIn(expected_call, subprocess.check_call.mock_calls)
1530
1531     def test_emits_error_message_when_ssh_chmod_failure(self):
1532         """ Should emit error message when ‘ssh … chmod’ command fails. """
1533         double = self.ssh_chmod_subprocess_double
1534         double.set_subprocess_check_call_scenario("failure")
1535         try:
1536             self.function_to_test(**self.test_args)
1537         except FakeSystemExit:
1538             pass
1539         expected_output = "Error while fixing permission."
1540         self.assertIn(expected_output, sys.stdout.getvalue())
1541
1542     def test_calls_sys_exit_when_ssh_chmod_failure(self):
1543         """ Should call `sys.exit` when ‘ssh … chmod’ command fails. """
1544         double = self.ssh_chmod_subprocess_double
1545         double.set_subprocess_check_call_scenario("failure")
1546         with testtools.ExpectedException(FakeSystemExit):
1547             self.function_to_test(**self.test_args)
1548         sys.exit.assert_called_with(EXIT_STATUS_FAILURE)
1549
1550 \f
1551 # Copyright © 2015–2016 Ben Finney <bignose@debian.org>
1552 #
1553 # This is free software: you may copy, modify, and/or distribute this work
1554 # under the terms of the GNU General Public License as published by the
1555 # Free Software Foundation; version 3 of that license or any later version.
1556 # No warranty expressed or implied. See the file ‘LICENSE.GPL-3’ for details.
1557
1558 \f
1559 # Local variables:
1560 # coding: utf-8
1561 # mode: python
1562 # End:
1563 # vim: fileencoding=utf-8 filetype=python :