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 sys
  15 import os
  16 import shutil
  17 import subprocess
  18 import tempfile
  19 import itertools
  20 import textwrap
  21 import doctest
  22
  23 import testtools
  24 import testscenarios
  25 import pkg_resources
  26
  27 import dput.dcut
  28 from dput.helper import dputhelper
  29
  30 from .helper import (
  31         StringIO,
  32         mock,
  33         FakeSystemExit,
  34         EXIT_STATUS_SUCCESS, EXIT_STATUS_FAILURE,
  35         patch_sys_argv,
  36         patch_system_interfaces,
  37         patch_time_time,
  38         patch_os_environ,
  39         patch_os_getpid,
  40         patch_os_getuid,
  41         PasswdEntry,
  42         patch_pwd_getpwuid,
  43         patch_os_unlink,
  44         patch_os_rmdir,
  45         patch_shutil_rmtree,
  46         patch_tempfile_mkdtemp,
  47         FileDouble,
  48         setup_file_double_behaviour,
  49         ARG_ANY, ARG_MORE,
  50         SubprocessDouble,
  51         patch_subprocess_popen,
  52         patch_subprocess_call,
  53         patch_subprocess_check_call,
  54         setup_subprocess_double_behaviour,
  55         )
  56 from .test_configfile import (
  57         set_config,
  58         )
  59 from .test_dputhelper import (
  60         patch_pkg_resources_get_distribution,
  61         patch_getopt,
  62         )
  63 from .test_changesfile import (
  64         make_changes_file_scenarios,
  65         set_changes_file_scenario,
  66         )
  67 from . import test_dput_main
  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 :