test/test_dcut.py

   1 # -*- coding: utf-8; -*-
   2 #
   3 # test/test_dcut.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 ‘dcut’ module. """
  11
  12 from __future__ import (absolute_import, unicode_literals)
  13
  14 import doctest
  15 import itertools
  16 import os
  17 import shutil
  18 import subprocess
  19 import sys
  20 import tempfile
  21 import textwrap
  22
  23 import pkg_resources
  24 import testscenarios
  25 import testtools
  26
  27 import dput.dcut
  28 from dput.helper import dputhelper
  29
  30 from .helper import (
  31         ARG_ANY, ARG_MORE,
  32         EXIT_STATUS_FAILURE, EXIT_STATUS_SUCCESS,
  33         FakeSystemExit,
  34         FileDouble,
  35         PasswdEntry,
  36         StringIO,
  37         SubprocessDouble,
  38         mock,
  39         patch_os_environ,
  40         patch_os_getpid,
  41         patch_os_getuid,
  42         patch_os_rmdir,
  43         patch_os_unlink,
  44         patch_pwd_getpwuid,
  45         patch_shutil_rmtree,
  46         patch_subprocess_call,
  47         patch_subprocess_check_call,
  48         patch_subprocess_popen,
  49         patch_sys_argv,
  50         patch_system_interfaces,
  51         patch_tempfile_mkdtemp,
  52         patch_time_time,
  53         setup_file_double_behaviour,
  54         setup_subprocess_double_behaviour,
  55         )
  56 from .test_changesfile import (
  57         make_changes_file_scenarios,
  58         set_changes_file_scenario,
  59         )
  60 from .test_configfile import (
  61         set_config,
  62         )
  63 from . import test_dput_main
  64 from .test_dputhelper import (
  65         patch_getopt,
  66         patch_pkg_resources_get_distribution,
  67         )
  68
  69 \f
  70 dummy_pwent = PasswdEntry(
  71         pw_name="lorem",
  72         pw_passwd="!",
  73         pw_uid=1,
  74         pw_gid=1,
  75         pw_gecos="Lorem Ipsum,spam,eggs,beans",
  76         pw_dir=tempfile.mktemp(),
  77         pw_shell=tempfile.mktemp())
  78
  79
  80 def patch_getoptions(testcase):
  81     """ Patch the `getoptions` function for this test case. """
  82     default_options = {
  83             'debug': False,
  84             'simulate': False,
  85             'config': None,
  86             'host': "foo",
  87             'passive': False,
  88             'changes': None,
  89             'filetoupload': None,
  90             'filetocreate': None,
  91             }
  92
  93     if not hasattr(testcase, 'getoptions_opts'):
  94         testcase.getoptions_opts = {}
  95     if not hasattr(testcase, 'getoptions_args'):
  96         testcase.getoptions_args = []
  97
  98     def fake_getoptions():
  99         options = dict(default_options)
 100         options.update(testcase.getoptions_opts)
 101         arguments = list(testcase.getoptions_args)
 102         result = (options, arguments)
 103         return result
 104
 105     func_patcher = mock.patch.object(
 106             dput.dcut, "getoptions", autospec=True,
 107             side_effect=fake_getoptions)
 108     func_patcher.start()
 109     testcase.addCleanup(func_patcher.stop)
 110
 111
 112 def get_upload_method_func(testcase):
 113     """ Get the specified upload method. """
 114     host = testcase.test_host
 115     method_name = testcase.runtime_config_parser.get(host, 'method')
 116     method_func = testcase.upload_methods[method_name]
 117     return method_func
 118
 119 \f
 120 class make_usage_message_TestCase(testtools.TestCase):
 121     """ Test cases for `make_usage_message` function. """
 122
 123     def setUp(self):
 124         """ Set up test fixtures. """
 125         super(make_usage_message_TestCase, self).setUp()
 126
 127         patch_sys_argv(self)
 128
 129     def test_returns_text_with_program_name(self):
 130         """ Should return text with expected program name. """
 131         result = dput.dcut.make_usage_message()
 132         expected_result = textwrap.dedent("""\
 133                 Usage: {progname} ...
 134                 ...
 135                 """).format(progname=self.progname)
 136         self.expectThat(
 137                 result,
 138                 testtools.matchers.DocTestMatches(
 139                     expected_result, flags=doctest.ELLIPSIS))
 140
 141 \f
 142 class getoptions_TestCase(testtools.TestCase):
 143     """ Base for test cases for `getoptions` function. """
 144
 145     default_options = NotImplemented
 146
 147     scenarios = NotImplemented
 148
 149     def setUp(self):
 150         """ Set up test fixtures. """
 151         super(getoptions_TestCase, self).setUp()
 152         patch_system_interfaces(self)
 153
 154         patch_os_environ(self)
 155         patch_os_getuid(self)
 156         patch_pwd_getpwuid(self)
 157         patch_sys_argv(self)
 158
 159         self.patch_etc_mailname()
 160         setup_file_double_behaviour(
 161                 self, [self.mailname_file_double])
 162
 163         self.set_hostname_subprocess_double()
 164         patch_subprocess_popen(self)
 165
 166         self.patch_getopt()
 167         if hasattr(self, 'expected_options'):
 168             self.set_expected_result()
 169
 170         self.patch_distribution()
 171         self.patch_make_usage_message()
 172
 173     def patch_etc_mailname(self):
 174         """ Patch the ‘/etc/mailname’ file. """
 175         path = "/etc/mailname"
 176         if hasattr(self, 'mailname_fake_file'):
 177             double = FileDouble(path, self.mailname_fake_file)
 178         else:
 179             double = FileDouble(path, StringIO())
 180         if hasattr(self, 'mailname_file_open_scenario_name'):
 181             double.set_open_scenario(self.mailname_file_open_scenario_name)
 182         self.mailname_file_double = double
 183
 184     def set_hostname_subprocess_double(self):
 185         """ Set the test double for the ‘hostname’ subprocess. """
 186         path = "/bin/hostname"
 187         argv = (path, "--fqdn")
 188         double = SubprocessDouble(path, argv=argv)
 189         double.register_for_testcase(self)
 190
 191         double.set_subprocess_popen_scenario('success')
 192         double.set_stdout_content(getattr(self, 'hostname_stdout_content', ""))
 193
 194         self.hostname_subprocess_double = double
 195
 196     def patch_getopt(self):
 197         """ Patch the `dputhelper.getopt` function. """
 198         if not hasattr(self, 'getopt_opts'):
 199             self.getopt_opts = []
 200         else:
 201             self.getopt_opts = list(self.getopt_opts)
 202         if not hasattr(self, 'getopt_args'):
 203             self.getopt_args = []
 204         else:
 205             self.getopt_args = list(self.getopt_args)
 206
 207         patch_getopt(self)
 208
 209     def set_expected_result(self):
 210         """ Set the expected result value. """
 211         if not hasattr(self, 'expected_arguments'):
 212             self.expected_arguments = []
 213         expected_options = self.default_options.copy()
 214         expected_options.update(self.expected_options)
 215         self.expected_result = (expected_options, self.expected_arguments)
 216
 217     def patch_distribution(self):
 218         """ Patch the Python distribution for this test case. """
 219         self.fake_distribution = mock.MagicMock(pkg_resources.Distribution)
 220         if hasattr(self, 'dcut_version'):
 221             self.fake_distribution.version = self.dcut_version
 222         patch_pkg_resources_get_distribution(self)
 223
 224     def patch_make_usage_message(self):
 225         """ Patch the `make_usage_message` function. """
 226         if hasattr(self, 'dcut_usage_message'):
 227             text = self.dcut_usage_message
 228         else:
 229             text = self.getUniqueString()
 230         func_patcher = mock.patch.object(
 231                 dput.dcut, "make_usage_message", autospec=True,
 232                 return_value=text)
 233         func_patcher.start()
 234         self.addCleanup(func_patcher.stop)
 235
 236
 237 class getoptions_UploaderTestCase(
 238         testscenarios.WithScenarios,
 239         getoptions_TestCase):
 240     """ Test cases for `getoptions` function, determining uploader. """
 241
 242     environ_scenarios = [
 243             ('environ-none', {
 244                 'os_environ': {},
 245                 }),
 246             ('environ-email-not-delimited', {
 247                 'os_environ': {'EMAIL': "quux@example.org"},
 248                 'expected_environ_uploader': "<quux@example.org>",
 249                 }),
 250             ('environ-email-delimited', {
 251                 'os_environ': {'EMAIL': "<quux@example.org>"},
 252                 'expected_environ_uploader': "<quux@example.org>",
 253                 }),
 254             ('environ-debemail-not-delimited', {
 255                 'os_environ': {'DEBEMAIL': "flup@example.org"},
 256                 'expected_environ_uploader': "<flup@example.org>",
 257                 }),
 258             ('environ-debemail-delimited', {
 259                 'os_environ': {'DEBEMAIL': "<flup@example.org>"},
 260                 'expected_environ_uploader': "<flup@example.org>",
 261                 }),
 262             ('environ-both-email-and-debfullname', {
 263                 'os_environ': {
 264                     'EMAIL': "quux@example.org",
 265                     'DEBFULLNAME': "Lorem Ipsum",
 266                     },
 267                 'expected_environ_uploader': "Lorem Ipsum <quux@example.org>",
 268                 }),
 269             ('environ-both-debemail-and-debfullname', {
 270                 'os_environ': {
 271                     'DEBEMAIL': "flup@example.org",
 272                     'DEBFULLNAME': "Lorem Ipsum",
 273                     },
 274                 'expected_environ_uploader': "Lorem Ipsum <flup@example.org>",
 275                 }),
 276             ('environ-both-email-and-debemail', {
 277                 'os_environ': {
 278                     'EMAIL': "quux@example.org",
 279                     'DEBEMAIL': "flup@example.org",
 280                     },
 281                 'expected_environ_uploader': "<flup@example.org>",
 282                 }),
 283             ('environ-both-email-and-debemail-and-debfullname', {
 284                 'os_environ': {
 285                     'EMAIL': "quux@example.org",
 286                     'DEBEMAIL': "flup@example.org",
 287                     'DEBFULLNAME': "Lorem Ipsum",
 288                     },
 289                 'expected_environ_uploader': "Lorem Ipsum <flup@example.org>",
 290                 }),
 291             ]
 292
 293     system_scenarios = [
 294             ('domain-from-mailname-file', {
 295                 'mailname_fake_file': StringIO("consecteur.example.org"),
 296                 'pwd_getpwuid_return_value': dummy_pwent._replace(
 297                         pw_name="grue",
 298                         pw_gecos="Dolor Sit Amet,spam,beans,eggs"),
 299                 'expected_debug_chatter': textwrap.dedent("""\
 300                         D: Guessing uploader
 301                         """),
 302                 'expected_system_uploader':
 303                     "Dolor Sit Amet <grue@consecteur.example.org>",
 304                 }),
 305             ('domain-from-hostname-command', {
 306                 'mailname_file_open_scenario_name': "read_denied",
 307                 'hostname_stdout_content': "consecteur.example.org\n",
 308                 'pwd_getpwuid_return_value': dummy_pwent._replace(
 309                         pw_name="grue",
 310                         pw_gecos="Dolor Sit Amet,spam,beans,eggs"),
 311                 'expected_debug_chatter': textwrap.dedent("""\
 312                         D: Guessing uploader
 313                         D: Guessing uploader: /etc/mailname was a failure
 314                         """),
 315                 'expected_system_uploader':
 316                     "Dolor Sit Amet <grue@consecteur.example.org>",
 317                 }),
 318             ('domain-failure', {
 319                 'mailname_file_open_scenario_name': "read_denied",
 320                 'hostname_stdout_content': "",
 321                 'pwd_getpwuid_return_value': dummy_pwent._replace(
 322                         pw_name="grue",
 323                         pw_gecos="Dolor Sit Amet,spam,beans,eggs"),
 324                 'expected_debug_chatter': textwrap.dedent("""\
 325                         D: Guessing uploader
 326                         D: Guessing uploader: /etc/mailname was a failure
 327                         D: Couldn't guess uploader
 328                         """),
 329                 }),
 330             ]
 331
 332     scenarios = testscenarios.multiply_scenarios(
 333             environ_scenarios, system_scenarios)
 334
 335     def setUp(self, *args, **kwargs):
 336         """ Set up test fixtures. """
 337         super(getoptions_UploaderTestCase, self).setUp(*args, **kwargs)
 338
 339         self.set_expected_uploader()
 340
 341     def set_expected_uploader(self):
 342         """ Set the expected uploader value for this test case. """
 343         for attrib_name in [
 344                 'expected_command_line_uploader',
 345                 'expected_environ_uploader',
 346                 'expected_system_uploader']:
 347             if hasattr(self, attrib_name):
 348                 self.expected_uploader = getattr(self, attrib_name)
 349                 break
 350
 351     def test_emits_debug_message_for_uploader_discovery(self):
 352         """ Should emit debug message for uploader discovery. """
 353         sys.argv.insert(1, "--debug")
 354         dput.dcut.getoptions()
 355         expected_output_lines = [
 356                 "D: trying to get maintainer email from environment"]
 357         if hasattr(self, 'expected_environ_uploader'):
 358             guess_line_template = "D: Uploader from env: {uploader}"
 359         else:
 360             expected_output_lines.extend(
 361                 self.expected_debug_chatter.split("\n")[:-1])
 362             if hasattr(self, 'expected_system_uploader'):
 363                 guess_line_template = "D: Guessed uploader: {uploader}"
 364         if hasattr(self, 'expected_uploader'):
 365             expected_output_lines.append(guess_line_template.format(
 366                     uploader=self.expected_uploader))
 367         expected_output = "\n".join(expected_output_lines)
 368         self.assertIn(expected_output, sys.stdout.getvalue())
 369
 370
 371 class getoptions_ParseCommandLineTestCase(
 372         testscenarios.WithScenarios,
 373         getoptions_TestCase):
 374     """ Test cases for `getoptions` function, parsing command line. """
 375
 376     dcut_usage_message = "Lorem ipsum, dolor sit amet."
 377
 378     progname = "lorem"
 379     dcut_version = "ipsum"
 380
 381     config_file_path = tempfile.mktemp()
 382     changes_file_path = tempfile.mktemp()
 383     output_file_path = tempfile.mktemp()
 384     upload_file_path = tempfile.mktemp()
 385
 386     default_options = dict()
 387     default_options.update(
 388             (key, None) for key in [
 389                 'config', 'host', 'uploader', 'keyid',
 390                 'filetocreate', 'filetoupload', 'changes'])
 391     default_options.update(
 392             (key, False) for key in ['debug', 'simulate', 'passive'])
 393
 394     option_scenarios = [
 395             ('no-options', {
 396                 'getopt_opts': [],
 397                 }),
 398             ('option-bogus', {
 399                 'getopt_opts': [("--b0gUs", "BOGUS")],
 400                 'expected_stderr_output': (
 401                     "{progname} internal error:"
 402                     " Option --b0gUs, argument BOGUS unknown").format(
 403                         progname=progname),
 404                 'expected_exit_status': EXIT_STATUS_FAILURE,
 405                 }),
 406             ('option-help', {
 407                 'getopt_opts': [("--help", None)],
 408                 'expected_stdout_output': dcut_usage_message,
 409                 'expected_exit_status': EXIT_STATUS_SUCCESS,
 410                 }),
 411             ('option-version', {
 412                 'getopt_opts': [("--version", None)],
 413                 'expected_stdout_output': " ".join(
 414                     [progname, dcut_version]),
 415                 'expected_exit_status': EXIT_STATUS_SUCCESS,
 416                 }),
 417             ('option-filetoupload-and-environ-uploader', {
 418                 'os_environ': {
 419                     'DEBEMAIL': "flup@example.org",
 420                     'DEBFULLNAME': "Lorem Ipsum",
 421                     },
 422                 'getopt_opts': [
 423                     ("--upload", upload_file_path),
 424                     ],
 425                 'expected_options': {
 426                     'uploader': "Lorem Ipsum <flup@example.org>",
 427                     'filetoupload': upload_file_path,
 428                     },
 429                 'expected_arguments': [],
 430                 }),
 431             ('option-changes-and-environ-uploader', {
 432                 'os_environ': {
 433                     'DEBEMAIL': "flup@example.org",
 434                     'DEBFULLNAME': "Lorem Ipsum",
 435                     },
 436                 'getopt_opts': [
 437                     ("--input", changes_file_path),
 438                     ],
 439                 'expected_options': {
 440                     'uploader': "Lorem Ipsum <flup@example.org>",
 441                     'changes': changes_file_path,
 442                     },
 443                 'expected_arguments': [],
 444                 }),
 445             ('option-filetoupload-and-option-maintaineraddress', {
 446                 'getopt_opts': [
 447                     ("--upload", upload_file_path),
 448                     ("--maintaineraddress", "Lorem Ipsum <flup@example.org>"),
 449                     ],
 450                 'expected_options': {
 451                     'uploader': "Lorem Ipsum <flup@example.org>",
 452                     'filetoupload': upload_file_path,
 453                     },
 454                 'expected_arguments': [],
 455                 }),
 456             ('option-changes-and-option-maintaineraddress', {
 457                 'getopt_opts': [
 458                     ("--input", changes_file_path),
 459                     ("--maintaineraddress", "Lorem Ipsum <flup@example.org>"),
 460                     ],
 461                 'expected_options': {
 462                     'uploader': "Lorem Ipsum <flup@example.org>",
 463                     'changes': changes_file_path,
 464                     },
 465                 'expected_arguments': [],
 466                 }),
 467             ('option-filetoupload-with-no-uploader', {
 468                 'getopt_opts': [("--upload", upload_file_path)],
 469                 'expected_stderr_output': "command file cannot be created",
 470                 'expected_exit_status': EXIT_STATUS_FAILURE,
 471                 }),
 472             ('option-changes-with-no-uploader', {
 473                 'getopt_opts': [("--input", changes_file_path)],
 474                 'expected_stderr_output': "command file cannot be created",
 475                 'expected_exit_status': EXIT_STATUS_FAILURE,
 476                 }),
 477             ('option-several', {
 478                 'getopt_opts': [
 479                     ("--debug", None),
 480                     ("--simulate", None),
 481                     ("--config", config_file_path),
 482                     ("--maintaineraddress", "Lorem Ipsum <flup@example.org>"),
 483                     ("--keyid", "DEADBEEF"),
 484                     ("--passive", None),
 485                     ("--output", output_file_path),
 486                     ("--host", "quux.example.com"),
 487                     ],
 488                 'expected_options': {
 489                     'debug': True,
 490                     'simulate': True,
 491                     'config': config_file_path,
 492                     'uploader': "Lorem Ipsum <flup@example.org>",
 493                     'keyid': "DEADBEEF",
 494                     'passive': True,
 495                     'filetocreate': output_file_path,
 496                     'host': "quux.example.com",
 497                     },
 498                 'expected_arguments': [],
 499                 }),
 500             ]
 501
 502     scenarios = option_scenarios
 503
 504     def test_emits_debug_message_for_program_version(self):
 505         """ Should emit debug message for program version. """
 506         sys.argv.insert(1, "--debug")
 507         expected_progname = self.progname
 508         expected_version = self.dcut_version
 509         try:
 510             dput.dcut.getoptions()
 511         except FakeSystemExit:
 512             pass
 513         expected_output = textwrap.dedent("""\
 514                 D: {progname} {version}
 515                 """).format(
 516                     progname=expected_progname,
 517                     version=expected_version)
 518         self.assertIn(expected_output, sys.stdout.getvalue())
 519
 520     def test_calls_getopt_with_expected_args(self):
 521         """ Should call `getopt` with expected arguments. """
 522         try:
 523             dput.dcut.getoptions()
 524         except FakeSystemExit:
 525             pass
 526         dputhelper.getopt.assert_called_with(
 527                 self.sys_argv[1:], mock.ANY, mock.ANY)
 528
 529     def test_emits_debug_message_for_each_option(self):
 530         """ Should emit a debug message for each option processed. """
 531         sys.argv.insert(1, "--debug")
 532         try:
 533             dput.dcut.getoptions()
 534         except FakeSystemExit:
 535             pass
 536         expected_output_lines = [
 537                 "D: processing arg \"{opt}\", option \"{arg}\"".format(
 538                     opt=option, arg=option_argument)
 539                 for (option, option_argument) in self.getopt_args]
 540         expected_output = "\n".join(expected_output_lines)
 541         self.assertIn(expected_output, sys.stdout.getvalue())
 542
 543     def test_emits_expected_message(self):
 544         """ Should emit message with expected content. """
 545         try:
 546             dput.dcut.getoptions()
 547         except FakeSystemExit:
 548             pass
 549         if hasattr(self, 'expected_stdout_output'):
 550             self.assertIn(self.expected_stdout_output, sys.stdout.getvalue())
 551         if hasattr(self, 'expected_stderr_output'):
 552             self.assertIn(self.expected_stderr_output, sys.stderr.getvalue())
 553
 554     def test_calls_sys_exit_with_expected_exit_status(self):
 555         """ Should call `sys.exit` with expected exit status. """
 556         if not hasattr(self, 'expected_exit_status'):
 557             dput.dcut.getoptions()
 558         else:
 559             with testtools.ExpectedException(FakeSystemExit):
 560                 dput.dcut.getoptions()
 561             sys.exit.assert_called_with(self.expected_exit_status)
 562
 563     def test_returns_expected_values(self):
 564         """ Should return expected values. """
 565         if not hasattr(self, 'expected_result'):
 566             self.skipTest("No return result expected")
 567         result = dput.dcut.getoptions()
 568         self.assertEqual(self.expected_result, result)
 569
 570
 571 class getoptions_DetermineHostTestCase(
 572         testscenarios.WithScenarios,
 573         getoptions_TestCase):
 574     """ Test cases for `getoptions` function, determine host name. """
 575
 576     system_scenarios = [
 577             ('domain-from-mailname-file', {
 578                 'mailname_fake_file': StringIO("consecteur.example.org"),
 579                 }),
 580             ]
 581
 582     default_options = getattr(
 583             getoptions_ParseCommandLineTestCase, 'default_options')
 584
 585     command_scenarios = [
 586             ('no-opts no-args', {
 587                 'getopt_opts': [],
 588                 'getopt_args': [],
 589                 'expected_options': {
 590                     'host': None,
 591                     },
 592                 }),
 593             ('no-opts command-first-arg', {
 594                 'getopt_opts': [],
 595                 'getopt_args': ["cancel"],
 596                 'expected_options': {
 597                     'host': None,
 598                     },
 599                 'expected_arguments': ["cancel"],
 600                 }),
 601             ('no-opts host-first-arg', {
 602                 'getopt_opts': [],
 603                 'getopt_args': ["quux.example.com", "cancel"],
 604                 'expected_options': {
 605                     'host': "quux.example.com",
 606                     },
 607                 'expected_arguments': ["cancel"],
 608                 'expected_debug_output': textwrap.dedent("""\
 609                     D: first argument "quux.example.com" treated as host
 610                     """),
 611                 }),
 612             ('option-host host-first-arg', {
 613                 'getopt_opts': [("--host", "quux.example.com")],
 614                 'getopt_args': ["decoy.example.net", "cancel"],
 615                 'expected_options': {
 616                     'host': "quux.example.com",
 617                     },
 618                 'expected_arguments': ["decoy.example.net", "cancel"],
 619                 }),
 620             ]
 621
 622     scenarios = testscenarios.multiply_scenarios(
 623             system_scenarios, command_scenarios)
 624
 625     def test_emits_expected_debug_message(self):
 626         """ Should emit the expected debug message. """
 627         if not hasattr(self, 'expected_debug_output'):
 628             self.expected_debug_output = ""
 629         self.getopt_opts = list(
 630                 self.getopt_opts + [("--debug", None)])
 631         dput.dcut.getoptions()
 632         self.assertIn(self.expected_debug_output, sys.stdout.getvalue())
 633
 634     def test_returns_expected_values(self):
 635         """ Should return expected values. """
 636         if not hasattr(self, 'expected_result'):
 637             self.skipTest("No return result expected")
 638         (options, arguments) = dput.dcut.getoptions()
 639         self.assertEqual(self.expected_options['host'], options['host'])
 640         self.assertEqual(self.expected_arguments, arguments)
 641
 642 \f
 643 class parse_queuecommands_TestCase(testtools.TestCase):
 644     """ Base for test cases for `parse_queuecommands` function. """
 645
 646     scenarios = NotImplemented
 647
 648     def setUp(self):
 649         """ Set up test fixtures. """
 650         super(parse_queuecommands_TestCase, self).setUp()
 651         patch_system_interfaces(self)
 652
 653         self.set_test_args()
 654
 655     def set_test_args(self):
 656         """ Set the arguments for the test call to the function. """
 657         default_options = {
 658                 'debug': False,
 659                 }
 660         self.test_args = dict(
 661                 arguments=getattr(self, 'arguments', []),
 662                 options=getattr(self, 'options', default_options),
 663                 config=object(),
 664                 )
 665
 666
 667 class parse_queuecommands_SuccessTestCase(
 668         testscenarios.WithScenarios,
 669         parse_queuecommands_TestCase):
 670     """ Success test cases for `parse_queuecommands` function. """
 671
 672     scenarios = [
 673             ('one-command-rm', {
 674                 'arguments': ["rm", "lorem.deb"],
 675                 'expected_commands': [
 676                     "rm --searchdirs lorem.deb",
 677                     ],
 678                 }),
 679             ('one-command-rm nosearchdirs', {
 680                 'arguments': ["rm", "--nosearchdirs", "lorem.deb"],
 681                 'expected_commands': [
 682                     "rm lorem.deb",
 683                     ],
 684                 }),
 685             ('one-command-cancel', {
 686                 'arguments': ["cancel", "lorem.deb"],
 687                 'expected_commands': [
 688                     "cancel lorem.deb",
 689                     ],
 690                 }),
 691             ('one-command-cancel nosearchdirs', {
 692                 'arguments': ["cancel", "--nosearchdirs", "lorem.deb"],
 693                 'expected_commands': [
 694                     "cancel --nosearchdirs lorem.deb",
 695                     ],
 696                 }),
 697             ('one-command-reschedule', {
 698                 'arguments': ["reschedule", "lorem.deb"],
 699                 'expected_commands': [
 700                     "reschedule lorem.deb",
 701                     ],
 702                 }),
 703             ('one-command-reschedule nosearchdirs', {
 704                 'arguments': ["reschedule", "--nosearchdirs", "lorem.deb"],
 705                 'expected_commands': [
 706                     "reschedule --nosearchdirs lorem.deb",
 707                     ],
 708                 }),
 709             ('three-commands comma-separated', {
 710                 'arguments': [
 711                     "rm", "foo", ",",
 712                     "cancel", "bar", ",",
 713                     "reschedule", "baz"],
 714                 'expected_commands': [
 715                     "rm --searchdirs foo ",
 716                     "cancel bar ",
 717                     "reschedule baz",
 718                     ],
 719                 }),
 720             ('three-commands semicolon-separated', {
 721                 'arguments': [
 722                     "rm", "foo", ";",
 723                     "cancel", "bar", ";",
 724                     "reschedule", "baz"],
 725                 'expected_commands': [
 726                     "rm --searchdirs foo ",
 727                     "cancel bar ",
 728                     "reschedule baz",
 729                     ],
 730                 }),
 731             ]
 732
 733     def test_emits_debug_message_for_each_command(self):
 734         """ Should emit a debug message for each command. """
 735         self.test_args['options'] = dict(self.test_args['options'])
 736         self.test_args['options']['debug'] = True
 737         dput.dcut.parse_queuecommands(**self.test_args)
 738         expected_output = "\n".join(
 739                 "D: Successfully parsed command \"{command}\"".format(
 740                     command=command)
 741                 for command in self.expected_commands)
 742         self.assertIn(expected_output, sys.stdout.getvalue())
 743
 744     def test_returns_expected_commands(self):
 745         """ Should return expected commands value. """
 746         result = dput.dcut.parse_queuecommands(**self.test_args)
 747         self.assertEqual(self.expected_commands, result)
 748
 749
 750 class parse_queuecommands_ErrorTestCase(
 751         testscenarios.WithScenarios,
 752         parse_queuecommands_TestCase):
 753     """ Error test cases for `parse_queuecommands` function. """
 754
 755     scenarios = [
 756             ('no-arguments', {
 757                 'arguments': [],
 758                 'expected_debug_output': textwrap.dedent("""\
 759                         Error: no arguments given, see dcut -h
 760                         """),
 761                 'expected_exit_status': EXIT_STATUS_FAILURE,
 762                 }),
 763             ('first-command-bogus', {
 764                 'arguments': ["b0gUs", "spam", "eggs"],
 765                 'expected_debug_output': textwrap.dedent("""\
 766                         Error: Could not parse commands at "b0gUs"
 767                         """),
 768                 'expected_exit_status': EXIT_STATUS_FAILURE,
 769                 }),
 770             ('third-command-bogus', {
 771                 'arguments': ["rm", "foo", ",", "cancel", "bar", ",", "b0gUs"],
 772                 'expected_debug_output': textwrap.dedent("""\
 773                         Error: Could not parse commands at "b0gUs"
 774                         """),
 775                 'expected_exit_status': EXIT_STATUS_FAILURE,
 776                 }),
 777             ]
 778
 779     def test_emits_expected_error_message(self):
 780         """ Should emit expected error message. """
 781         try:
 782             dput.dcut.parse_queuecommands(**self.test_args)
 783         except FakeSystemExit:
 784             pass
 785         self.assertIn(self.expected_debug_output, sys.stderr.getvalue())
 786
 787     def test_calls_sys_exit_with_exit_status(self):
 788         """ Should call `sys.exit` with expected exit status. """
 789         with testtools.ExpectedException(FakeSystemExit):
 790             dput.dcut.parse_queuecommands(**self.test_args)
 791         sys.exit.assert_called_with(self.expected_exit_status)
 792
 793 \f
 794 class create_commands_TestCase(
 795         testtools.TestCase):
 796     """ Test cases for `create_commands` function. """
 797
 798     def setUp(self):
 799         """ Set up test fixtures. """
 800         super(create_commands_TestCase, self).setUp()
 801         patch_system_interfaces(self)
 802
 803         self.changes_file_scenarios = make_changes_file_scenarios()
 804         set_changes_file_scenario(self, 'no-format')
 805         setup_file_double_behaviour(self)
 806
 807         self.set_expected_commands()
 808
 809         self.set_options()
 810
 811         test_dput_main.patch_parse_changes(self)
 812         dput.dput.parse_changes.return_value = self.changes_file_scenario[
 813                 'expected_result']
 814
 815         self.set_test_args()
 816
 817     def set_options(self):
 818         """ Set the options mapping to pass to the function. """
 819         self.options = {
 820                 'debug': False,
 821                 'changes': self.changes_file_double.path,
 822                 }
 823
 824     def set_test_args(self):
 825         """ Set the arguments for the test call to the function. """
 826         self.test_args = dict(
 827                 options=dict(self.options),
 828                 config=object(),
 829                 parse_changes=dput.dput.parse_changes,
 830                 )
 831
 832     def set_expected_commands(self):
 833         """ Set the expected commands for this test case. """
 834         files_to_remove = [os.path.basename(self.changes_file_double.path)]
 835         files_from_changes = self.changes_file_scenario[
 836                 'expected_result']['files']
 837         for line in files_from_changes.split("\n"):
 838             files_to_remove.append(line.split(" ")[4])
 839         self.expected_commands = [
 840                 "rm --searchdirs {path}".format(path=path)
 841                 for path in files_to_remove]
 842
 843     def test_emits_debug_message_for_changes_file(self):
 844         """ Should emit debug message for changes file. """
 845         self.options['debug'] = True
 846         self.set_test_args()
 847         dput.dcut.create_commands(**self.test_args)
 848         expected_output = textwrap.dedent("""\
 849                 D: Parsing changes file ({path}) for files to remove
 850                 """).format(path=self.changes_file_double.path)
 851         self.assertIn(expected_output, sys.stdout.getvalue())
 852
 853     def test_emits_error_message_when_changes_file_open_error(self):
 854         """ Should emit error message when changes file raises error. """
 855         self.changes_file_double.set_open_scenario('read_denied')
 856         try:
 857             dput.dcut.create_commands(**self.test_args)
 858         except FakeSystemExit:
 859             pass
 860         expected_output = textwrap.dedent("""\
 861                 Can't open changes file: {path}
 862                 """).format(path=self.changes_file_double.path)
 863         self.assertIn(expected_output, sys.stdout.getvalue())
 864
 865     def test_calls_sys_exit_when_changes_file_open_error(self):
 866         """ Should call `sys.exit` when changes file raises error. """
 867         self.changes_file_double.set_open_scenario('read_denied')
 868         with testtools.ExpectedException(FakeSystemExit):
 869             dput.dcut.create_commands(**self.test_args)
 870         sys.exit.assert_called_with(EXIT_STATUS_FAILURE)
 871
 872     def test_returns_expected_result(self):
 873         """ Should return expected result. """
 874         result = dput.dcut.create_commands(**self.test_args)
 875         self.assertEqual(self.expected_commands, result)
 876
 877 \f
 878 class write_commands_TestCase(
 879         testscenarios.WithScenarios,
 880         testtools.TestCase):
 881     """ Test cases for `write_commands` function. """
 882
 883     default_options = {
 884             'filetocreate': None,
 885             }
 886
 887     path_scenarios = [
 888             ('default-path', {}),
 889             ('filetocreate', {
 890                 'option_filetocreate': str("ipsum.commands"),
 891                 'expected_result': "ipsum.commands",
 892                 }),
 893             ('no-tempdir', {
 894                 'tempdir': None,
 895                 }),
 896             ]
 897
 898     commands_scenarios = [
 899             ('commands-none', {
 900                 'commands': [],
 901                 }),
 902             ('commands-one', {
 903                 'commands': ["foo"],
 904                 }),
 905             ('commands-three', {
 906                 'commands': ["foo", "bar", "baz"],
 907                 }),
 908             ]
 909
 910     keyid_scenarios = [
 911             ('keyid-none', {}),
 912             ('keyid-set', {
 913                 'option_keyid': "DEADBEEF",
 914                 }),
 915             ]
 916
 917     scenarios = testscenarios.multiply_scenarios(
 918             path_scenarios, commands_scenarios, keyid_scenarios)
 919
 920     for (scenario_name, scenario) in scenarios:
 921         default_options = getattr(
 922                 getoptions_ParseCommandLineTestCase, 'default_options')
 923         options = dict(default_options)
 924         options.update({
 925                 'uploader': str("Lorem Ipsum <flup@example.org>"),
 926                 })
 927         scenario['uploader_filename_part'] = str(
 928                 "Lorem_Ipsum__flup_example_org_")
 929         if 'option_filetocreate' in scenario:
 930             options['filetocreate'] = scenario['option_filetocreate']
 931         if 'option_keyid' in scenario:
 932             options['keyid'] = scenario['option_keyid']
 933         scenario['options'] = options
 934         if 'tempdir' not in scenario:
 935             scenario['tempdir'] = tempfile.mktemp()
 936     del scenario_name, scenario
 937     del default_options, options
 938
 939     def setUp(self):
 940         """ Set up test fixtures. """
 941         super(write_commands_TestCase, self).setUp()
 942         patch_system_interfaces(self)
 943
 944         patch_os_getpid(self)
 945         os.getpid.return_value = self.getUniqueInteger()
 946
 947         self.time_return_value = self.getUniqueInteger()
 948         patch_time_time(self, itertools.repeat(self.time_return_value))
 949
 950         patch_sys_argv(self)
 951         self.set_commands_file_double()
 952         setup_file_double_behaviour(self)
 953         self.set_expected_result()
 954
 955         self.set_commands()
 956
 957         self.set_test_args()
 958
 959         patch_subprocess_popen(self)
 960         patch_subprocess_check_call(self)
 961         self.set_debsign_subprocess_double()
 962         setup_subprocess_double_behaviour(self)
 963
 964     def set_options(self):
 965         """ Set the options mapping to pass to the function. """
 966
 967     def set_test_args(self):
 968         """ Set the arguments for the test call to the function. """
 969         self.test_args = dict(
 970                 commands=list(self.commands),
 971                 options=dict(self.options),
 972                 config=object(),
 973                 tempdir=self.tempdir,
 974                 )
 975
 976     def make_commands_filename(self):
 977         """ Make the filename for the commands output file. """
 978         expected_progname = self.progname
 979         filename = "{progname}.{uploadpart}.{time:d}.{pid:d}.commands".format(
 980                 progname=expected_progname,
 981                 uploadpart=self.uploader_filename_part,
 982                 time=self.time_return_value,
 983                 pid=os.getpid.return_value)
 984         return filename
 985
 986     def set_commands_file_double(self):
 987         """ Set the commands file double for this test case. """
 988         if self.options['filetocreate']:
 989             path = self.options['filetocreate']
 990         else:
 991             output_filename = self.make_commands_filename()
 992             if self.tempdir:
 993                 path = os.path.join(self.tempdir, output_filename)
 994             else:
 995                 path = output_filename
 996         double = FileDouble(path)
 997         double.register_for_testcase(self)
 998         self.commands_file_double = double
 999
1000     def set_expected_result(self):
1001         """ Set the `expected_result` for this test case. """
1002         self.expected_result = self.commands_file_double.path
1003
1004     def set_commands(self):
1005         """ Set the commands to use for this test case. """
1006         if not hasattr(self, 'commands'):
1007             self.commands = []
1008
1009     def make_expected_content(self):
1010         """ Make the expected content for the output file. """
1011         uploader_value = self.options['uploader']
1012         if self.commands:
1013             commands_value = "\n".join(
1014                     " {command}".format(command=command)
1015                     for command in self.commands)
1016         else:
1017             commands_value = " "
1018         commands_value += "\n"
1019         text = textwrap.dedent("""\
1020                 Uploader: {uploader}
1021                 Commands:
1022                 {commands}
1023                 """).format(uploader=uploader_value, commands=commands_value)
1024         return text
1025
1026     def set_debsign_subprocess_double(self):
1027         """ Set the ‘debsign’ subprocess double for this test case. """
1028         path = "/usr/bin/debsign"
1029         argv = [os.path.basename(path), ARG_MORE]
1030         double = SubprocessDouble(path, argv)
1031         double.register_for_testcase(self)
1032         self.debsign_subprocess_double = double
1033
1034     def make_expected_debsign_argv(self):
1035         """ Make the expected command-line arguments for ‘debsign’. """
1036         argv = [
1037                 str("debsign"),
1038                 str("-m{uploader}").format(uploader=self.options['uploader']),
1039                 ]
1040         if self.options['keyid']:
1041             argv.append(
1042                     "-k{keyid}".format(keyid=self.options['keyid']))
1043         argv.append(self.commands_file_double.path)
1044
1045         return argv
1046
1047     def test_returns_expected_file_path(self):
1048         """ Should return expected file path. """
1049         result = dput.dcut.write_commands(**self.test_args)
1050         self.assertEqual(self.expected_result, result)
1051
1052     def test_output_file_has_expected_content(self):
1053         """ Should have expected content in output file. """
1054         with mock.patch.object(
1055                 self.commands_file_double.fake_file, "close", autospec=True):
1056             dput.dcut.write_commands(**self.test_args)
1057         expected_value = self.make_expected_content()
1058         self.assertEqual(
1059                 expected_value, self.commands_file_double.fake_file.getvalue())
1060
1061     def test_emits_debug_message_for_debsign(self):
1062         """ Should emit debug message for ‘debsign’ command. """
1063         self.options['debug'] = True
1064         self.test_args['options'] = self.options
1065         dput.dcut.write_commands(**self.test_args)
1066         debsign_argv = self.make_expected_debsign_argv()
1067         expected_output = textwrap.dedent("""\
1068                 D: calling debsign: {argv}
1069                 """).format(argv=debsign_argv)
1070         self.assertIn(expected_output, sys.stdout.getvalue())
1071
1072     def test_calls_subprocess_check_call_with_expected_args(self):
1073         """ Should call `subprocess.check_call` with expected args. """
1074         debsign_argv = self.make_expected_debsign_argv()
1075         expected_args = [debsign_argv]
1076         dput.dcut.write_commands(**self.test_args)
1077         subprocess.check_call.assert_called_with(*expected_args)
1078
1079     def test_emits_error_message_when_debsign_failure(self):
1080         """ Should emit error message when ‘debsign’ command failure. """
1081         self.debsign_subprocess_double.set_subprocess_check_call_scenario(
1082                 'failure')
1083         try:
1084             dput.dcut.write_commands(**self.test_args)
1085         except FakeSystemExit:
1086             pass
1087         expected_output = textwrap.dedent("""\
1088                 Error: debsign failed.
1089                 """)
1090         self.assertIn(expected_output, sys.stderr.getvalue())
1091
1092     def test_calls_sys_exit_when_debsign_failure(self):
1093         """ Should call `sys.exit` when ‘debsign’ command failure. """
1094         self.debsign_subprocess_double.set_subprocess_check_call_scenario(
1095                 'failure')
1096         with testtools.ExpectedException(FakeSystemExit):
1097             dput.dcut.write_commands(**self.test_args)
1098         sys.exit.assert_called_with(EXIT_STATUS_FAILURE)
1099
1100 \f
1101 class upload_TestCase(test_dput_main.main_TestCase):
1102     """ Base for test cases for `upload_stolen_from_dput_main` function. """
1103
1104     function_to_test = staticmethod(dput.dcut.upload_stolen_from_dput_main)
1105
1106     def setUp(self):
1107         """ Set up test fixtures. """
1108         super(upload_TestCase, self).setUp()
1109
1110         self.set_cat_subprocess_double()
1111         patch_subprocess_call(self)
1112         patch_tempfile_mkdtemp(self)
1113         patch_os_rmdir(self)
1114
1115         patch_getoptions(self)
1116
1117     def set_cat_subprocess_double(self):
1118         """ Set the ‘cat’ subprocess double for this test case. """
1119         path = "/bin/cat"
1120         argv = [os.path.basename(path), ARG_ANY]
1121         double = SubprocessDouble(path, argv)
1122         double.register_for_testcase(self)
1123         double.set_subprocess_call_scenario('success')
1124         self.cat_subprocess_double = double
1125
1126     def set_test_args(self):
1127         """ Set the arguments for the test call to the function. """
1128         self.test_args = dict(
1129                 host=self.test_host,
1130                 upload_methods=self.upload_methods,
1131                 config=self.runtime_config_parser,
1132                 debug=False,
1133                 simulate=False,
1134                 files_to_upload=self.files_to_upload,
1135                 ftp_passive_mode=False,
1136                 )
1137
1138         if hasattr(self, 'test_args_extra'):
1139             self.test_args.update(self.test_args_extra)
1140
1141     def get_upload_method_func(self):
1142         """ Get the specified upload method. """
1143         method_name = self.runtime_config_parser.get(self.test_host, 'method')
1144         method_func = self.upload_methods[method_name]
1145         return method_func
1146
1147
1148 class upload_DebugMessageTestCase(upload_TestCase):
1149     """ Test cases for `upload_stolen_from_dput_main` debug messages. """
1150
1151     def test_emits_debug_message_for_discovered_methods(self):
1152         """ Should emit debug message for discovered upload methods. """
1153         self.test_args['debug'] = True
1154         self.function_to_test(**self.test_args)
1155         expected_output = textwrap.dedent("""\
1156                 D: Default Method: {default_method}
1157                 D: Host Method: {host_method}
1158                 """).format(
1159                     default_method=self.runtime_config_parser.get(
1160                         'DEFAULT', 'method'),
1161                     host_method=self.runtime_config_parser.get(
1162                         self.test_host, 'method'))
1163         self.assertIn(expected_output, sys.stdout.getvalue())
1164
1165
1166 class upload_UnknownUploadMethodTestCase(
1167         testscenarios.WithScenarios,
1168         upload_TestCase):
1169     """ Test cases for `upload_stolen_from_dput_main`, unknown method. """
1170
1171     scenarios = [
1172             ('bogus-default-method', {
1173                 'config_extras': {
1174                     'default': {
1175                         'method': "b0gUs",
1176                         },
1177                     },
1178                 'expected_output': "Unknown upload method: b0gUs",
1179                 'expected_exit_status': EXIT_STATUS_FAILURE,
1180                 }),
1181             ('bogus-host-method', {
1182                 'config_extras': {
1183                     'host': {
1184                         'method': "b0gUs",
1185                         },
1186                     },
1187                 'expected_output': "Unknown upload method: b0gUs",
1188                 'expected_exit_status': EXIT_STATUS_FAILURE,
1189                 }),
1190             ]
1191
1192     def test_emits_error_message_when_unknown_method(self):
1193         """ Should emit error message when unknown upload method. """
1194         try:
1195             self.function_to_test(**self.test_args)
1196         except FakeSystemExit:
1197             pass
1198         self.assertIn(self.expected_output, sys.stderr.getvalue())
1199
1200     def test_calls_sys_exit_when_unknown_method(self):
1201         """ Should call `sys.exit` when unknown upload method. """
1202         with testtools.ExpectedException(FakeSystemExit):
1203             self.function_to_test(**self.test_args)
1204         sys.exit.assert_called_with(self.expected_exit_status)
1205
1206
1207 class upload_DiscoverLoginTestCase(
1208         testscenarios.WithScenarios,
1209         upload_TestCase):
1210     """ Test cases for `upload_stolen_from_dput_main` discovery of login. """
1211
1212     fallback_login_scenarios = [
1213             ('login-from-environ', {
1214                 'os_environ': {
1215                     'USER': "login-from-environ",
1216                     },
1217                 'expected_fallback_login': "login-from-environ",
1218                 'expected_system_uid_debug_message': "",
1219                 }),
1220             ('login-from-pwd', {
1221                 'os_getuid_return_value': 42,
1222                 'pwd_getpwuid_return_value': PasswdEntry(
1223                         *(["login-from-pwd"] + [object()] * 6)),
1224                 'expected_fallback_login': "login-from-pwd",
1225                 'expected_system_uid_debug_message': "D: User-ID: 42",
1226                 }),
1227             ]
1228
1229     config_login_scenarios = [
1230             ('config-default-login', {
1231                 'config_extras': {
1232                     'default': {
1233                         'login': "login-from-config-default",
1234                         },
1235                     },
1236                 'expected_login': "login-from-config-default",
1237                 'expected_output_template':
1238                     "D: Login to use: {login}",
1239                 }),
1240             ('config-host-login', {
1241                 'config_extras': {
1242                     'host': {
1243                         'login': "login-from-config-host",
1244                         },
1245                     },
1246                 'expected_login': "login-from-config-host",
1247                 'expected_output_template':
1248                     "D: Login to use: {login}",
1249                 }),
1250             ('config-default-login sentinel', {
1251                 'config_extras': {
1252                     'default': {
1253                         'login': "username",
1254                         },
1255                     },
1256                 'expected_output_template': (
1257                     "D: Neither host {host} nor default login used."
1258                     " Using {login}"),
1259                 }),
1260             ('config-host-login sentinel', {
1261                 'config_extras': {
1262                     'host': {
1263                         'login': "username",
1264                         },
1265                     },
1266                 'expected_output_template': (
1267                     "D: Neither host {host} nor default login used."
1268                     " Using {login}"),
1269                 }),
1270             ]
1271
1272     scenarios = testscenarios.multiply_scenarios(
1273             fallback_login_scenarios, config_login_scenarios)
1274     for (scenario_name, scenario) in scenarios:
1275         if 'expected_login' not in scenario:
1276             scenario['expected_login'] = scenario['expected_fallback_login']
1277     del scenario_name, scenario
1278
1279     def test_emits_debug_message_for_system_uid(self):
1280         """ Should emit a debug message for the system UID. """
1281         if self.expected_login != self.expected_fallback_login:
1282             self.skipTest("No fallback in this scenario")
1283         self.test_args['debug'] = True
1284         self.function_to_test(**self.test_args)
1285         expected_output = self.expected_system_uid_debug_message.format(
1286                 uid=self.os_getuid_return_value)
1287         self.assertIn(expected_output, sys.stdout.getvalue())
1288
1289     def test_emits_debug_message_for_discovered_login(self):
1290         """ Should emit a debug message for the discovered login. """
1291         self.test_args['debug'] = True
1292         self.function_to_test(**self.test_args)
1293         expected_output = self.expected_output_template.format(
1294                 login=self.expected_login, host=self.test_host)
1295         self.assertIn(expected_output, sys.stdout.getvalue())
1296
1297     def test_calls_upload_method_with_expected_login(self):
1298         """ Should call upload method function with expected login arg. """
1299         upload_method_func = get_upload_method_func(self)
1300         self.function_to_test(**self.test_args)
1301         upload_method_func.assert_called_with(
1302                 mock.ANY, self.expected_login,
1303                 mock.ANY, mock.ANY, mock.ANY, mock.ANY)
1304
1305
1306 class upload_SimulateTestCase(
1307         testscenarios.WithScenarios,
1308         upload_TestCase):
1309     """ Test cases for `upload_stolen_from_dput_main`, ‘simulate’ option. """
1310
1311     scenarios = [
1312             ('simulate', {
1313                 'config_default_login': "login-from-config-default",
1314                 'test_args_extra': {
1315                     'simulate': True,
1316                     },
1317                 }),
1318             ('simulate three-files', {
1319                 'config_default_login': "login-from-config-default",
1320                 'test_args_extra': {
1321                     'simulate': True,
1322                     },
1323                 'files_to_upload': [tempfile.mktemp() for __ in range(3)],
1324                 }),
1325             ]
1326
1327     def test_omits_upload_method(self):
1328         """ Should omit call to upload method function. """
1329         upload_method_func = get_upload_method_func(self)
1330         self.function_to_test(**self.test_args)
1331         self.assertFalse(upload_method_func.called)
1332
1333     def test_emits_message_for_each_file_to_upload(self):
1334         """ Should emit a message for each file to upload. """
1335         self.function_to_test(**self.test_args)
1336         method = self.runtime_config_parser.get(self.test_host, 'method')
1337         fqdn = self.runtime_config_parser.get(self.test_host, 'fqdn')
1338         incoming = self.runtime_config_parser.get(self.test_host, 'incoming')
1339         expected_output = "\n".join(
1340                 "Uploading with {method}: {path} to {fqdn}:{incoming}".format(
1341                     method=method, path=path,
1342                     fqdn=fqdn, incoming=incoming)
1343                 for path in self.files_to_upload)
1344         self.assertIn(expected_output, sys.stderr.getvalue())
1345
1346     def test_calls_cat_for_each_file_to_upload(self):
1347         """ Should call ‘cat’ for each file to upload. """
1348         self.function_to_test(**self.test_args)
1349         for path in self.files_to_upload:
1350             expected_call = mock.call(
1351                     "cat {path}".format(path=path),
1352                     shell=True)
1353             self.expectThat(
1354                     subprocess.call.mock_calls,
1355                     testtools.matchers.Contains(expected_call))
1356
1357
1358 class upload_UploadMethodTestCase(
1359         testscenarios.WithScenarios,
1360         upload_TestCase):
1361     """ Test cases for `upload_stolen_from_dput_main`, invoking method. """
1362
1363     method_scenarios = [
1364             ('method-local', {
1365                 'config_method': "local",
1366                 'config_progress_indicator': 23,
1367                 'expected_args': (
1368                     "localhost", mock.ANY, mock.ANY, mock.ANY, mock.ANY,
1369                     0),
1370                 }),
1371             ('method-ftp', {
1372                 'config_method': "ftp",
1373                 'config_fqdn': "foo.example.com",
1374                 'config_passive_ftp': False,
1375                 'config_progress_indicator': 23,
1376                 'expected_args': (
1377                     "foo.example.com", mock.ANY, mock.ANY, mock.ANY, mock.ANY,
1378                     False),
1379                 'expected_stdout_output': "",
1380                 }),
1381             ('method-ftp port-custom', {
1382                 'config_method': "ftp",
1383                 'config_fqdn': "foo.example.com:42",
1384                 'config_passive_ftp': False,
1385                 'config_progress_indicator': 23,
1386                 'expected_args': (
1387                     "foo.example.com:42",
1388                     mock.ANY, mock.ANY, mock.ANY, mock.ANY,
1389                     False),
1390                 'expected_stdout_output': "",
1391                 }),
1392             ('method-ftp config-passive-mode', {
1393                 'config_method': "ftp",
1394                 'config_fqdn': "foo.example.com",
1395                 'config_passive_ftp': True,
1396                 'config_progress_indicator': 23,
1397                 'expected_args': (
1398                     "foo.example.com", mock.ANY, mock.ANY, mock.ANY, mock.ANY,
1399                     True),
1400                 'expected_stdout_output': "",
1401                 }),
1402             ('method-ftp config-passive-mode arg-ftp-active-mode', {
1403                 'config_method': "ftp",
1404                 'config_fqdn': "foo.example.com",
1405                 'config_passive_ftp': True,
1406                 'config_progress_indicator': 23,
1407                 'test_args_extra': {
1408                     'ftp_passive_mode': False,
1409                     },
1410                 'expected_args': (
1411                     "foo.example.com", mock.ANY, mock.ANY, mock.ANY, mock.ANY,
1412                     True),
1413                 'expected_stdout_output': "D: Using active ftp",
1414                 }),
1415             ('method-ftp arg-ftp-passive-mode', {
1416                 'config_method': "ftp",
1417                 'config_fqdn': "foo.example.com",
1418                 'config_progress_indicator': 23,
1419                 'test_args_extra': {
1420                     'ftp_passive_mode': True,
1421                     },
1422                 'expected_args': (
1423                     "foo.example.com", mock.ANY, mock.ANY, mock.ANY, mock.ANY,
1424                     True),
1425                 'expected_stdout_output': "D: Using passive ftp",
1426                 }),
1427             ('method-ftp config-passive-mode arg-ftp-passive-mode', {
1428                 'config_method': "ftp",
1429                 'config_fqdn': "foo.example.com",
1430                 'config_passive_ftp': True,
1431                 'config_progress_indicator': 23,
1432                 'test_args_extra': {
1433                     'ftp_passive_mode': True,
1434                     },
1435                 'expected_args': (
1436                     "foo.example.com", mock.ANY, mock.ANY, mock.ANY, mock.ANY,
1437                     True),
1438                 'expected_stdout_output': "D: Using passive ftp",
1439                 }),
1440             ('method-scp', {
1441                 'config_method': "scp",
1442                 'config_fqdn': "foo.example.com",
1443                 'expected_args': (
1444                     "foo.example.com", mock.ANY, mock.ANY, mock.ANY, mock.ANY,
1445                     False, []),
1446                 'expected_stdout_output': "",
1447                 }),
1448             ('method-scp scp-compress', {
1449                 'config_method': "scp",
1450                 'config_fqdn': "foo.example.com",
1451                 'config_extras': {
1452                     'host': {
1453                         'scp_compress': "True",
1454                         'ssh_config_options': "spam eggs beans",
1455                         },
1456                     },
1457                 'expected_args': (
1458                     "foo.example.com", mock.ANY, mock.ANY, mock.ANY, mock.ANY,
1459                     True, ["spam eggs beans"]),
1460                 'expected_stdout_output': "D: Setting compression for scp",
1461                 }),
1462             ]
1463
1464     login_scenarios = [
1465             ('default-login', {
1466                 'config_default_login': "login-from-config-default",
1467                 }),
1468             ]
1469
1470     commands_scenarios = [
1471             ('commands-from-changes', {
1472                 'getoptions_args': ["foo", "bar", "baz"],
1473                 'getoptions_opts': {
1474                     'filetocreate': None,
1475                     'filetoupload': tempfile.mktemp() + "commands",
1476                     },
1477                 }),
1478             ('commands-from-changes', {
1479                 'getoptions_args': ["foo", "bar", "baz"],
1480                 'getoptions_opts': {
1481                     'filetocreate': None,
1482                     'filetoupload': None,
1483                     'changes': tempfile.mktemp(),
1484                     },
1485                 }),
1486             ('commands-from-arguments', {
1487                 'getoptions_args': ["foo", "bar", "baz"],
1488                 'getoptions_opts': {
1489                     'filetocreate': None,
1490                     'filetoupload': None,
1491                     'changes': None,
1492                     },
1493                 }),
1494             ]
1495
1496     files_scenarios = [
1497             ('no-files', {
1498                 'files_to_remove': [],
1499                 }),
1500             ('three-files', {
1501                 'files_to_remove': [
1502                         tempfile.mktemp() for __ in range(3)],
1503                 }),
1504             ]
1505
1506     scenarios = testscenarios.multiply_scenarios(
1507             method_scenarios, login_scenarios,
1508             commands_scenarios, files_scenarios)
1509
1510     def test_emits_expected_debug_message(self):
1511         """ Should emit expected debug message. """
1512         self.test_args['debug'] = True
1513         self.function_to_test(**self.test_args)
1514         if hasattr(self, 'expected_stdout_output'):
1515             self.assertIn(self.expected_stdout_output, sys.stdout.getvalue())
1516
1517     def test_calls_upload_method_with_expected_args(self):
1518         """ Should call upload method function with expected args. """
1519         upload_method_func = get_upload_method_func(self)
1520         self.function_to_test(**self.test_args)
1521         upload_method_func.assert_called_with(*self.expected_args)
1522
1523 \f
1524 class dcut_TestCase(testtools.TestCase):
1525     """ Base for test cases for `dput` function. """
1526
1527     def setUp(self):
1528         """ Set up test fixtures. """
1529         super(dcut_TestCase, self).setUp()
1530         patch_system_interfaces(self)
1531
1532         patch_tempfile_mkdtemp(self)
1533         patch_os_unlink(self)
1534         patch_os_rmdir(self)
1535         patch_shutil_rmtree(self)
1536
1537         set_config(
1538                 self,
1539                 getattr(self, 'config_scenario_name', 'exist-simple'))
1540         test_dput_main.patch_runtime_config_options(self)
1541
1542         self.set_test_args()
1543
1544         patch_getoptions(self)
1545         test_dput_main.patch_parse_changes(self)
1546         test_dput_main.patch_read_configs(self)
1547         test_dput_main.set_upload_methods(self)
1548         test_dput_main.patch_import_upload_functions(self)
1549
1550         self.patch_parse_queuecommands()
1551         self.patch_create_commands()
1552         self.patch_write_commands()
1553         self.patch_upload_stolen_from_dput_main()
1554
1555     def set_test_args(self):
1556         """ Set the arguments for the test call to the function. """
1557         self.test_args = dict()
1558
1559     def patch_parse_queuecommands(self):
1560         """ Patch the `parse_queuecommands` function for this test case. """
1561         func_patcher = mock.patch.object(
1562                 dput.dcut, "parse_queuecommands", autospec=True)
1563         func_patcher.start()
1564         self.addCleanup(func_patcher.stop)
1565
1566     def patch_create_commands(self):
1567         """ Patch the `create_commands` function for this test case. """
1568         func_patcher = mock.patch.object(
1569                 dput.dcut, "create_commands", autospec=True)
1570         func_patcher.start()
1571         self.addCleanup(func_patcher.stop)
1572
1573     def patch_write_commands(self):
1574         """ Patch the `write_commands` function for this test case. """
1575         func_patcher = mock.patch.object(
1576                 dput.dcut, "write_commands", autospec=True)
1577         func_patcher.start()
1578         self.addCleanup(func_patcher.stop)
1579
1580     def patch_upload_stolen_from_dput_main(self):
1581         """ Patch `upload_stolen_from_dput_main` for this test case. """
1582         func_patcher = mock.patch.object(
1583                 dput.dcut, "upload_stolen_from_dput_main", autospec=True)
1584         func_patcher.start()
1585         self.addCleanup(func_patcher.stop)
1586
1587
1588 class dcut_DebugMessageTestCase(dcut_TestCase):
1589     """ Test cases for `dcut` debug messages. """
1590
1591     def test_emits_debug_message_for_read_configs(self):
1592         """ Should emit debug message for `read_configs` call. """
1593         self.getoptions_opts['debug'] = True
1594         dput.dcut.dcut(**self.test_args)
1595         expected_output = textwrap.dedent("""\
1596                 D: calling dput.read_configs
1597                 """)
1598         self.assertIn(expected_output, sys.stdout.getvalue())
1599
1600
1601 class dcut_ConfigFileTestCase(
1602         testscenarios.WithScenarios,
1603         dcut_TestCase):
1604     """ Test cases for `main` specification of configuration file. """
1605
1606     scenarios = [
1607             ('default', {
1608                 'expected_args': (None, mock.ANY),
1609                 }),
1610             ('config-from-command-line', {
1611                 'getoptions_opts': {
1612                     'config': "lorem.conf",
1613                     },
1614                 'expected_args': ("lorem.conf", mock.ANY),
1615                 }),
1616             ]
1617
1618     def test_calls_read_configs_with_expected_args(self):
1619         """ Should call `read_configs` with expected arguments. """
1620         dput.dcut.dcut(**self.test_args)
1621         dput.dput.read_configs.assert_called_with(*self.expected_args)
1622
1623
1624 class dcut_OptionsErrorTestCase(
1625         testscenarios.WithScenarios,
1626         dcut_TestCase):
1627     """ Test cases for `dcut` function, startup options cause error. """
1628
1629     scenarios = [
1630             ('no-host-discovered', {
1631                 'config_default_default_host_main': None,
1632                 'getoptions_opts': {
1633                     'host': None,
1634                     },
1635                 'expected_output': (
1636                     "Error: No host specified"
1637                     " and no default found in config"),
1638                 'expected_exit_status': EXIT_STATUS_FAILURE,
1639                 }),
1640             ('host-not-in-config', {
1641                 'config_scenario_name': "exist-minimal",
1642                 'expected_output': "No host foo found in config",
1643                 'expected_exit_status': EXIT_STATUS_FAILURE,
1644                 }),
1645             ('no-allow-dcut', {
1646                 'config_allow_dcut': False,
1647                 'expected_output': (
1648                     "Error: dcut is not supported for this upload queue."),
1649                 'expected_exit_status': EXIT_STATUS_FAILURE,
1650                 }),
1651             ('filetoupload arguments', {
1652                 'getoptions_opts': {
1653                     'filetoupload': tempfile.mktemp() + ".commands",
1654                     },
1655                 'getoptions_args': ["lorem", "ipsum", "dolor", "sit", "amet"],
1656                 'expected_output': (
1657                     "Error: cannot take commands"
1658                     " when uploading existing file"),
1659                 'expected_exit_status': EXIT_STATUS_FAILURE,
1660                 }),
1661             ]
1662
1663     def test_emits_expected_error_message(self):
1664         """ Should emit expected error message. """
1665         try:
1666             dput.dcut.dcut(**self.test_args)
1667         except FakeSystemExit:
1668             pass
1669         self.assertIn(self.expected_output, sys.stdout.getvalue())
1670
1671     def test_calls_sys_exit_with_failure_exit_status(self):
1672         """ Should call `sys.exit` with failure exit status. """
1673         with testtools.ExpectedException(FakeSystemExit):
1674             dput.dcut.dcut(**self.test_args)
1675         sys.exit.assert_called_with(self.expected_exit_status)
1676
1677
1678 class dcut_NamedHostTestCase(
1679         testscenarios.WithScenarios,
1680         dcut_TestCase):
1681     """ Test cases for `dcut` function, named host processing. """
1682
1683     scenarios = [
1684             ('host-from-command-line', {
1685                 'config_scenario_name': "exist-simple-host-three",
1686                 'config_default_default_host_main': "quux",
1687                 'getoptions_opts': {
1688                     'host': "bar",
1689                     },
1690                 'expected_host': "bar",
1691                 'expected_debug_output': "",
1692                 }),
1693             ('host-from-config-default', {
1694                 'config_scenario_name': "exist-simple-host-three",
1695                 'config_default_default_host_main': "bar",
1696                 'getoptions_opts': {
1697                     'host': None,
1698                     },
1699                 'expected_host': "bar",
1700                 'expected_debug_output': textwrap.dedent("""\
1701                     D: Using host "bar" (default_host_main)
1702                     """),
1703                 }),
1704             ('host-from-hardcoded-default', {
1705                 'config_scenario_name': "exist-default-distribution-only",
1706                 'config_default_default_host_main': "",
1707                 'getoptions_opts': {
1708                     'host': None,
1709                     },
1710                 'expected_host': "ftp-master",
1711                 'expected_debug_output': textwrap.dedent("""\
1712                     D: Using host "" (default_host_main)
1713                     D: Using host "ftp-master" (hardcoded)
1714                     """),
1715                 }),
1716             ]
1717
1718     def test_emits_debug_message_for_discovered_host(self):
1719         """ Should emit debug message for discovered host values. """
1720         self.getoptions_opts['debug'] = True
1721         dput.dcut.dcut(**self.test_args)
1722         self.assertIn(self.expected_debug_output, sys.stdout.getvalue())
1723
1724     def test_calls_write_commands_with_expected_host_option(self):
1725         """ Should call `write_commands` with expected `host` option. """
1726         dput.dcut.dcut(**self.test_args)
1727         self.assertEqual(1, len(dput.dcut.write_commands.mock_calls))
1728         (__, call_args, call_kwargs) = dput.dcut.write_commands.mock_calls[0]
1729         (__, options, __, __) = call_args
1730         self.assertEqual(self.expected_host, options['host'])
1731
1732
1733 class dcut_FileToUploadBadNameTestCase(
1734         testscenarios.WithScenarios,
1735         dcut_TestCase):
1736     """ Test cases for `dcut` function, file to upload with bad name. """
1737
1738     scenarios = [
1739             ('filetoupload', {
1740                 'getoptions_args': [],
1741                 'getoptions_opts': {
1742                     'filetoupload': tempfile.mktemp(),
1743                     },
1744                 'expected_output': (
1745                     "Error: I'm insisting on the .commands extension"),
1746                 }),
1747             ]
1748
1749     def test_emits_error_message_for_bad_filename(self):
1750         """ Should emit error message for bad filename. """
1751         dput.dcut.dcut(**self.test_args)
1752         self.assertIn(self.expected_output, sys.stdout.getvalue())
1753
1754
1755 class dcut_ParseChangesTestCase(
1756         testscenarios.WithScenarios,
1757         dcut_TestCase):
1758     """ Test cases for `dcut` function, parse upload control file. """
1759
1760     scenarios = [
1761             ('changes-file no-filetoupload', {
1762                 'getoptions_opts': {
1763                     'filetoupload': None,
1764                     'changes': tempfile.mktemp(),
1765                     },
1766                 }),
1767             ('changes-file no-filetoupload no-filetocreate', {
1768                 'getoptions_opts': {
1769                     'filetoupload': None,
1770                     'filetocreate': None,
1771                     'changes': tempfile.mktemp(),
1772                     },
1773                 }),
1774             ]
1775
1776     def test_calls_create_commands_with_expected_args(self):
1777         """ Should call `create_commands` with expected args. """
1778         dput.dcut.dcut(**self.test_args)
1779         (expected_options, __) = dput.dcut.getoptions()
1780         expected_config = self.runtime_config_parser
1781         expected_parse_changes = dput.dput.parse_changes
1782         dput.dcut.create_commands.assert_called_with(
1783                 expected_options, expected_config, expected_parse_changes)
1784
1785     def test_calls_write_commands_with_expected_args(self):
1786         """ Should call `write_commands` with expected args. """
1787         expected_commands = object()
1788         dput.dcut.create_commands.return_value = expected_commands
1789         dput.dcut.dcut(**self.test_args)
1790         (expected_options, __) = dput.dcut.getoptions()
1791         expected_config = self.runtime_config_parser
1792         expected_tempdir = self.tempfile_mkdtemp_file_double.path
1793         dput.dcut.write_commands.assert_called_with(
1794                 expected_commands, expected_options, expected_config,
1795                 expected_tempdir)
1796
1797
1798 class dcut_ParseQueueCommandsTestCase(
1799         testscenarios.WithScenarios,
1800         dcut_TestCase):
1801     """ Test cases for `dcut` function, parse commands from arguments. """
1802
1803     scenarios = [
1804             ('no-changes-file no-filetoupload', {
1805                 'getoptions_opts': {
1806                     'filetoupload': None,
1807                     'changes': None,
1808                     },
1809                 }),
1810             ]
1811
1812     def test_calls_parse_queuecommands_with_expected_args(self):
1813         """ Should call `parse_queuecommands` with expected args. """
1814         dput.dcut.dcut(**self.test_args)
1815         (expected_options, expected_arguments) = dput.dcut.getoptions()
1816         expected_config = self.runtime_config_parser
1817         dput.dcut.parse_queuecommands.assert_called_with(
1818                 expected_arguments, expected_options, expected_config)
1819
1820     def test_calls_write_commands_with_expected_args(self):
1821         """ Should call `write_commands` with expected args. """
1822         expected_commands = object()
1823         dput.dcut.parse_queuecommands.return_value = expected_commands
1824         dput.dcut.dcut(**self.test_args)
1825         (expected_options, __) = dput.dcut.getoptions()
1826         expected_config = self.runtime_config_parser
1827         expected_tempdir = self.tempfile_mkdtemp_file_double.path
1828         dput.dcut.write_commands.assert_called_with(
1829                 expected_commands, expected_options, expected_config,
1830                 expected_tempdir)
1831
1832
1833 class dcut_CleanupTestCase(
1834         testscenarios.WithScenarios,
1835         dcut_TestCase):
1836     """ Test cases for `dcut` function, cleanup from exception. """
1837
1838     commands_scenarios = [
1839             ('commands-from-arguments', {
1840                 'getoptions_args': ["foo", "bar", "baz"],
1841                 'getoptions_opts': {
1842                     'filetocreate': None,
1843                     'filetoupload': None,
1844                     'changes': None,
1845                     },
1846                 }),
1847             ]
1848
1849     files_scenarios = upload_UploadMethodTestCase.files_scenarios
1850
1851     scenarios = testscenarios.multiply_scenarios(
1852             commands_scenarios, files_scenarios)
1853
1854     def setUp(self):
1855         """ Set up test fixtures. """
1856         super(dcut_CleanupTestCase, self).setUp()
1857
1858         upload_method_func = get_upload_method_func(self)
1859         self.upload_error = RuntimeError("Bad stuff happened")
1860         upload_method_func.side_effect = self.upload_error
1861
1862     def test_removes_temporary_directory_when_upload_raises_exception(self):
1863         """ Should remove directory `tempdir` when exception raised. """
1864         try:
1865             dput.dcut.dcut(**self.test_args)
1866         except self.upload_error.__class__:
1867             pass
1868         shutil.rmtree.assert_called_with(
1869                 self.tempfile_mkdtemp_file_double.path)
1870
1871 \f
1872 # Copyright © 2015–2016 Ben Finney <bignose@debian.org>
1873 #
1874 # This is free software: you may copy, modify, and/or distribute this work
1875 # under the terms of the GNU General Public License as published by the
1876 # Free Software Foundation; version 3 of that license or any later version.
1877 # No warranty expressed or implied. See the file ‘LICENSE.GPL-3’ for details.
1878
1879 \f
1880 # Local variables:
1881 # coding: utf-8
1882 # mode: python
1883 # End:
1884 # vim: fileencoding=utf-8 filetype=python :