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