test/test_changesfile.py

   1 # -*- coding: utf-8; -*-
   2 #
   3 # test/test_changesfile.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 Debian upload control (‘*.changes’) files. """
  11
  12 from __future__ import (absolute_import, unicode_literals)
  13
  14 import collections
  15 import doctest
  16 import email.message
  17 import os
  18 import os.path
  19 import sys
  20 import tempfile
  21 import textwrap
  22
  23 import testscenarios
  24 import testtools
  25
  26 import dput.dput
  27 from dput.helper import dputhelper
  28
  29 from .helper import (
  30         EXIT_STATUS_FAILURE,
  31         FakeSystemExit,
  32         FileDouble,
  33         StringIO,
  34         get_file_doubles_from_fake_file_scenarios,
  35         make_fake_file_scenarios,
  36         make_unique_slug,
  37         mock,
  38         patch_os_stat,
  39         patch_system_interfaces,
  40         setup_file_double_behaviour,
  41         )
  42 from .test_configfile import (
  43         patch_runtime_config_options,
  44         set_config,
  45         )
  46
  47 __metaclass__ = type
  48
  49 \f
  50 class _FieldsMapping:
  51     """ A mapping to stand in for the `dict` of an `email.message.Message`. """
  52
  53     def __init__(self, *args, **kwargs):
  54         try:
  55             self._message = kwargs.pop('_message')
  56         except KeyError:
  57             raise TypeError("no ‘_message’ specified for this mapping")
  58         super(_FieldsMapping, self).__init__(*args, **kwargs)
  59
  60     def __len__(self, *args, **kwargs):
  61         return self._message.__len__(*args, **kwargs)
  62
  63     def __contains__(self, *args, **kwargs):
  64         return self._message.__contains__(*args, **kwargs)
  65
  66     def __getitem__(self, *args, **kwargs):
  67         return self._message.__getitem__(*args, **kwargs)
  68
  69     def __setitem__(self, *args, **kwargs):
  70         self._message.__setitem__(*args, **kwargs)
  71
  72     def __delitem__(self, *args, **kwargs):
  73         self._message.__delitem__(*args, **kwargs)
  74
  75     # Deprecated method for Python 2 compatibility.
  76     def has_key(self, *args, **kwargs):
  77         return self._message.has_key(*args, **kwargs)    # noqa: W601
  78
  79     def keys(self, *args, **kwargs):
  80         return self._message.keys(*args, **kwargs)
  81
  82     def values(self, *args, **kwargs):
  83         return self._message.values(*args, **kwargs)
  84
  85     def items(self, *args, **kwargs):
  86         return self._message.items(*args, **kwargs)
  87
  88     def get(self, *args, **kwargs):
  89         return self._message.get(*args, **kwargs)
  90
  91
  92 class FakeMessage(email.message.Message, object):
  93     """ A fake RFC 2822 message that mocks the obsolete `rfc822.Message`. """
  94
  95     def __init__(self, *args, **kwargs):
  96         super(FakeMessage, self).__init__(*args, **kwargs)
  97         self.dict = _FieldsMapping(_message=self)
  98
  99
 100 def make_fake_message(fields):
 101     """ Make a fake message instance. """
 102     message = FakeMessage()
 103     for (name, value) in fields.items():
 104         message.add_header(name, value)
 105     return message
 106
 107
 108 def make_files_field_value(params_by_name):
 109     """ Make a value for “Files” field of a changes document. """
 110     result = "\n".join(
 111             " ".join(params)
 112             for (file_name, params) in params_by_name.items())
 113     return result
 114
 115
 116 def make_upload_files_params(checksums_by_file_name, sizes_by_file_name):
 117     """ Make a mapping of upload parameters for files. """
 118     params_by_name = {
 119             file_name: [
 120                 checksums_by_file_name[file_name],
 121                 str(sizes_by_file_name[file_name]),
 122                 "foo", "bar", file_name]
 123             for file_name in checksums_by_file_name}
 124     return params_by_name
 125
 126
 127 def make_changes_document(fields, upload_params_by_name=None):
 128     """ Make a changes document from field values.
 129
 130         :param fields: Sequence of (name, value) tuples for fields.
 131         :param upload_params_by_name: Mapping from filename to upload
 132             parameters for each file.
 133         :return: The changes document as an RFC 822 formatted text.
 134
 135         """
 136     document_fields = fields.copy()
 137     if upload_params_by_name is not None:
 138         files_field_text = make_files_field_value(upload_params_by_name)
 139         document_fields.update({'files': files_field_text})
 140     document = make_fake_message(document_fields)
 141
 142     return document
 143
 144
 145 def make_changes_file_scenarios():
 146     """ Make fake Debian upload control (‘*.changes’) scenarios. """
 147     file_path = make_changes_file_path()
 148
 149     fake_file_empty = StringIO()
 150     fake_file_no_format = StringIO(textwrap.dedent("""\
 151             FOO
 152             BAR
 153             Files:
 154                 Lorem ipsum dolor sit amet
 155             """))
 156     fake_file_with_signature = StringIO(textwrap.dedent("""\
 157             -----BEGIN PGP SIGNED MESSAGE-----
 158             Hash: SHA1
 159
 160             FOO
 161             BAR
 162             Files:
 163                 Lorem ipsum dolor sit amet
 164
 165             -----BEGIN PGP SIGNATURE-----
 166             Version: 0.0
 167             Comment: Proin ac massa at orci sagittis fermentum.
 168
 169             gibberishgibberishgibberishgibberishgibberishgibberish
 170             gibberishgibberishgibberishgibberishgibberishgibberish
 171             gibberishgibberishgibberishgibberishgibberishgibberish
 172             -----END PGP SIGNATURE-----
 173             """))
 174     fake_file_with_format = StringIO(textwrap.dedent("""\
 175             Format: FOO
 176             Files:
 177                 Lorem ipsum dolor sit amet
 178             """))
 179     fake_file_invalid = StringIO(textwrap.dedent("""\
 180             Format: FOO
 181             Files:
 182                 FOO BAR
 183             """))
 184
 185     scenarios = [
 186             ('no-format', {
 187                 'file_double': FileDouble(
 188                     path=file_path,
 189                     fake_file=fake_file_no_format),
 190                 'expected_result': make_changes_document({
 191                     'files': "Lorem ipsum dolor sit amet",
 192                     }),
 193                 }),
 194             ('with-pgp-signature', {
 195                 'file_double': FileDouble(
 196                     path=file_path,
 197                     fake_file=fake_file_with_signature),
 198                 'expected_result': make_changes_document({
 199                     'files': "Lorem ipsum dolor sit amet",
 200                     }),
 201                 }),
 202             ('with-format', {
 203                 'file_double': FileDouble(
 204                     path=file_path,
 205                     fake_file=fake_file_with_format),
 206                 'expected_result': make_changes_document({
 207                     'files': "Lorem ipsum dolor sit amet",
 208                     }),
 209                 }),
 210             ('error empty', {
 211                 'file_double': FileDouble(
 212                     path=file_path, fake_file=fake_file_empty),
 213                 'expected_error': KeyError,
 214                 }),
 215             ('error invalid', {
 216                 'file_double': FileDouble(
 217                     path=file_path,
 218                     fake_file=fake_file_invalid),
 219                 'expected_error': FakeSystemExit,
 220                 }),
 221             ]
 222
 223     for (scenario_name, scenario) in scenarios:
 224         scenario['changes_file_scenario_name'] = scenario_name
 225
 226     return scenarios
 227
 228 \f
 229 def set_fake_upload_file_paths(testcase):
 230     """ Set the fake upload file paths. """
 231     testcase.fake_upload_file_paths = [
 232             os.path.join(
 233                 os.path.dirname(testcase.changes_file_double.path),
 234                 os.path.basename(tempfile.mktemp()))
 235             for __ in range(10)]
 236
 237     required_suffixes = [".dsc", ".tar.xz"]
 238     suffixes = required_suffixes + getattr(
 239             testcase, 'additional_file_suffixes', [])
 240     file_path_base = testcase.fake_upload_file_paths.pop()
 241     for suffix in suffixes:
 242         file_path = file_path_base + suffix
 243         testcase.fake_upload_file_paths.insert(0, file_path)
 244
 245
 246 def set_file_checksums(testcase):
 247     """ Set the fake file checksums for the test case. """
 248     testcase.fake_checksum_by_file = {
 249             os.path.basename(file_path): make_unique_slug(testcase)
 250             for file_path in testcase.fake_upload_file_paths}
 251
 252
 253 def set_file_sizes(testcase):
 254     """ Set the fake file sizes for the test case. """
 255     testcase.fake_size_by_file = {
 256             os.path.basename(file_path): testcase.getUniqueInteger()
 257             for file_path in testcase.fake_upload_file_paths}
 258
 259
 260 def set_file_doubles(testcase):
 261     """ Set the file doubles for the test case. """
 262     for file_path in testcase.fake_upload_file_paths:
 263         file_double = FileDouble(file_path)
 264         file_double.set_os_stat_scenario('okay')
 265         file_double.stat_result = file_double.stat_result._replace(
 266                 st_size=testcase.fake_size_by_file[
 267                     os.path.basename(file_path)],
 268                 )
 269         file_double.register_for_testcase(testcase)
 270
 271
 272 def setup_upload_file_fixtures(testcase):
 273     """ Set fixtures for fake files to upload for the test case. """
 274     set_fake_upload_file_paths(testcase)
 275     set_file_checksums(testcase)
 276     set_file_sizes(testcase)
 277     set_file_doubles(testcase)
 278
 279 \f
 280 def make_changes_file_path(file_dir_path=None):
 281     """ Make a filesystem path for the changes file. """
 282     if file_dir_path is None:
 283         file_dir_path = tempfile.mktemp()
 284     file_name = os.path.basename(
 285             "{base}.changes".format(base=tempfile.mktemp()))
 286     file_path = os.path.join(file_dir_path, file_name)
 287
 288     return file_path
 289
 290
 291 def setup_changes_file_fixtures(testcase):
 292     """ Set up fixtures for changes file doubles. """
 293     file_path = make_changes_file_path()
 294
 295     scenarios = make_fake_file_scenarios(file_path)
 296     testcase.changes_file_scenarios = scenarios
 297
 298     file_doubles = get_file_doubles_from_fake_file_scenarios(
 299             scenarios.values())
 300     setup_file_double_behaviour(testcase, file_doubles)
 301
 302
 303 def set_changes_file_scenario(testcase, name):
 304     """ Set the changes file scenario for this test case. """
 305     scenario = dict(testcase.changes_file_scenarios)[name]
 306     testcase.changes_file_scenario = scenario
 307     testcase.changes_file_double = scenario['file_double']
 308     testcase.changes_file_double.register_for_testcase(testcase)
 309
 310 \f
 311 class parse_changes_TestCase(
 312         testscenarios.WithScenarios,
 313         testtools.TestCase):
 314     """ Base for test cases for `parse_changes` function. """
 315
 316     scenarios = NotImplemented
 317
 318     def setUp(self):
 319         """ Set up test fixtures. """
 320         super(parse_changes_TestCase, self).setUp()
 321         patch_system_interfaces(self)
 322
 323         self.test_infile = StringIO()
 324
 325
 326 class parse_changes_SuccessTestCase(parse_changes_TestCase):
 327     """ Success test cases for `parse_changes` function. """
 328
 329     scenarios = list(
 330             (name, scenario)
 331             for (name, scenario) in make_changes_file_scenarios()
 332             if not name.startswith('error'))
 333
 334     def test_gives_expected_result_for_infile(self):
 335         """ Should give the expected result for specified input file. """
 336         result = dput.dput.parse_changes(self.file_double.fake_file)
 337         normalised_result_set = set(
 338                 (key.lower(), value.strip())
 339                 for (key, value) in result.items())
 340         self.assertEqual(
 341                 set(self.expected_result.items()), normalised_result_set)
 342
 343
 344 class parse_changes_ErrorTestCase(parse_changes_TestCase):
 345     """ Error test cases for `parse_changes` function. """
 346
 347     scenarios = list(
 348             (name, scenario)
 349             for (name, scenario) in make_changes_file_scenarios()
 350             if name.startswith('error'))
 351
 352     def test_raises_expected_exception_for_infile(self):
 353         """ Should raise the expected exception for specified input file. """
 354         with testtools.ExpectedException(self.expected_error):
 355             dput.dput.parse_changes(self.file_double.fake_file)
 356
 357 \f
 358 class check_upload_variant_TestCase(
 359         testscenarios.WithScenarios,
 360         testtools.TestCase):
 361     """ Test cases for `check_upload_variant` function. """
 362
 363     scenarios = [
 364             ('simple', {
 365                 'fields': {
 366                     'architecture': "foo bar baz",
 367                     },
 368                 'expected_result': True,
 369                 }),
 370             ('arch-missing', {
 371                 'fields': {
 372                     'spam': "Lorem ipsum dolor sit amet",
 373                     },
 374                 'expected_result': False,
 375                 }),
 376             ('source-only', {
 377                 'fields': {
 378                     'architecture': "source",
 379                     },
 380                 'expected_result': False,
 381                 }),
 382             ('source-and-others', {
 383                 'fields': {
 384                     'architecture': "foo source bar",
 385                     },
 386                 'expected_result': False,
 387                 }),
 388             ]
 389
 390     def setUp(self):
 391         """ Set up test fixtures. """
 392         super(check_upload_variant_TestCase, self).setUp()
 393         patch_system_interfaces(self)
 394
 395         self.set_changes_document(self.fields)
 396         self.set_test_args()
 397
 398     def set_changes_document(self, fields):
 399         """ Set the package changes document based on specified fields. """
 400         self.test_changes_document = make_changes_document(fields)
 401
 402     def set_test_args(self):
 403         """ Set the arguments for the test call to the function. """
 404         self.test_args = {
 405                 'changes': self.test_changes_document,
 406                 'debug': False,
 407                 }
 408
 409     def test_returns_expected_result_for_changes_document(self):
 410         """ Should return expected result for specified changes document. """
 411         result = dput.dput.check_upload_variant(**self.test_args)
 412         self.assertEqual(self.expected_result, result)
 413
 414     def test_emits_debug_message_showing_architecture(self):
 415         """ Should emit a debug message for the specified architecture. """
 416         if 'architecture' not in self.fields:
 417             self.skipTest("Architecture field not in this scenario")
 418         self.test_args['debug'] = True
 419         dput.dput.check_upload_variant(**self.test_args)
 420         expected_output = textwrap.dedent("""\
 421                 D: Architecture: {arch}
 422                 """).format(arch=self.fields['architecture'])
 423         self.assertIn(expected_output, sys.stdout.getvalue())
 424
 425     def test_emits_debug_message_for_binary_upload(self):
 426         """ Should emit a debug message for the specified architecture. """
 427         triggers_binaryonly = bool(self.expected_result)
 428         if not triggers_binaryonly:
 429             self.skipTest("Scenario does not trigger binary-only upload")
 430         self.test_args['debug'] = True
 431         dput.dput.check_upload_variant(**self.test_args)
 432         expected_output = textwrap.dedent("""\
 433                 D: Doing a binary upload only.
 434                 """)
 435         self.assertIn(expected_output, sys.stdout.getvalue())
 436
 437 \f
 438 SourceCheckResult = collections.namedtuple(
 439         'SourceCheckResult', ['include_orig', 'include_tar'])
 440
 441
 442 class source_check_TestCase(
 443         testscenarios.WithScenarios,
 444         testtools.TestCase):
 445     """ Test cases for `source_check` function. """
 446
 447     default_expected_result = SourceCheckResult(
 448             include_orig=False, include_tar=False)
 449
 450     scenarios = [
 451             ('no-version', {
 452                 'expected_result': default_expected_result,
 453                 }),
 454             ('no-epoch native-version', {
 455                 'upstream_version': "1.2",
 456                 'expected_result': SourceCheckResult(
 457                     include_orig=False, include_tar=True),
 458                 }),
 459             ('epoch native-version', {
 460                 'epoch': "3",
 461                 'upstream_version': "1.2",
 462                 'expected_result': SourceCheckResult(
 463                     include_orig=False, include_tar=True),
 464                 }),
 465             ('no-epoch debian-release', {
 466                 'upstream_version': "1.2",
 467                 'release': "5",
 468                 'expected_result': SourceCheckResult(
 469                     include_orig=False, include_tar=True),
 470                 }),
 471             ('epoch debian-release', {
 472                 'epoch': "3",
 473                 'upstream_version': "1.2",
 474                 'release': "5",
 475                 'expected_result': SourceCheckResult(
 476                     include_orig=False, include_tar=True),
 477                 }),
 478             ('no-epoch new-upstream-version', {
 479                 'upstream_version': "1.2",
 480                 'release': "1",
 481                 'expected_result': SourceCheckResult(
 482                     include_orig=True, include_tar=False),
 483                 }),
 484             ('epoch new_upstream-version', {
 485                 'epoch': "3",
 486                 'upstream_version': "1.2",
 487                 'release': "1",
 488                 'expected_result': SourceCheckResult(
 489                     include_orig=True, include_tar=False),
 490                 }),
 491             ('no-epoch nmu', {
 492                 'upstream_version': "1.2",
 493                 'release': "4.5",
 494                 'expected_result': SourceCheckResult(
 495                     include_orig=False, include_tar=True),
 496                 }),
 497             ('epoch nmu', {
 498                 'epoch': "3",
 499                 'upstream_version': "1.2",
 500                 'release': "4.5",
 501                 'expected_result': SourceCheckResult(
 502                     include_orig=False, include_tar=True),
 503                 }),
 504             ('no-epoch nmu before-first-release', {
 505                 'upstream_version': "1.2",
 506                 'release': "0.1",
 507                 'expected_result': SourceCheckResult(
 508                     include_orig=True, include_tar=False),
 509                 }),
 510             ('epoch nmu before-first-release', {
 511                 'epoch': "3",
 512                 'upstream_version': "1.2",
 513                 'release': "0.1",
 514                 'expected_result': SourceCheckResult(
 515                     include_orig=True, include_tar=False),
 516                 }),
 517             ('no-epoch nmu after-first-release', {
 518                 'upstream_version': "1.2",
 519                 'release': "1.1",
 520                 'expected_result': SourceCheckResult(
 521                     include_orig=True, include_tar=False),
 522                 }),
 523             ('epoch nmu after-first-release', {
 524                 'epoch': "3",
 525                 'upstream_version': "1.2",
 526                 'release': "1.1",
 527                 'expected_result': SourceCheckResult(
 528                     include_orig=True, include_tar=False),
 529                 }),
 530             ]
 531
 532     for (scenario_name, scenario) in scenarios:
 533         fields = {}
 534         if 'upstream_version' in scenario:
 535             version_string = scenario['upstream_version']
 536             if 'epoch' in scenario:
 537                 version_string = "{epoch}:{version}".format(
 538                         epoch=scenario['epoch'], version=version_string)
 539             if 'release' in scenario:
 540                 version_string = "{version}-{release}".format(
 541                         version=version_string, release=scenario['release'])
 542             fields.update({'version': version_string})
 543             scenario['version'] = version_string
 544         scenario['changes_document'] = make_changes_document(fields)
 545     del scenario_name, scenario
 546     del fields, version_string
 547
 548     def setUp(self):
 549         """ Set up test fixtures. """
 550         super(source_check_TestCase, self).setUp()
 551         patch_system_interfaces(self)
 552
 553         self.test_args = {
 554                 'changes': self.changes_document,
 555                 'debug': False,
 556                 }
 557
 558     def test_returns_expected_result_for_changes_document(self):
 559         """ Should return expected result for specified changes document. """
 560         result = dput.dput.source_check(**self.test_args)
 561         self.assertEqual(self.expected_result, result)
 562
 563     def test_emits_version_string_debug_message_only_if_version(self):
 564         """ Should emit message for version only if has version. """
 565         self.test_args['debug'] = True
 566         version = getattr(self, 'version', None)
 567         message_lead = "D: Package Version:"
 568         expected_output = textwrap.dedent("""\
 569                 {lead} {version}
 570                 """).format(
 571                     lead=message_lead, version=version)
 572         dput.dput.source_check(**self.test_args)
 573         if hasattr(self, 'version'):
 574             self.assertIn(expected_output, sys.stdout.getvalue())
 575         else:
 576             self.assertNotIn(message_lead, sys.stdout.getvalue())
 577
 578     def test_emits_epoch_debug_message_only_if_epoch(self):
 579         """ Should emit message for epoch only if has an epoch. """
 580         self.test_args['debug'] = True
 581         dput.dput.source_check(**self.test_args)
 582         expected_output = textwrap.dedent("""\
 583                 D: Epoch found
 584                 """)
 585         dput.dput.source_check(**self.test_args)
 586         if (hasattr(self, 'epoch') and hasattr(self, 'release')):
 587             self.assertIn(expected_output, sys.stdout.getvalue())
 588         else:
 589             self.assertNotIn(expected_output, sys.stdout.getvalue())
 590
 591     def test_emits_upstream_version_debug_message_only_if_nonnative(self):
 592         """ Should emit message for upstream version only if non-native. """
 593         self.test_args['debug'] = True
 594         upstream_version = getattr(self, 'upstream_version', None)
 595         message_lead = "D: Upstream Version:"
 596         expected_output = textwrap.dedent("""\
 597                 {lead} {version}
 598                 """).format(
 599                     lead=message_lead, version=upstream_version)
 600         dput.dput.source_check(**self.test_args)
 601         if hasattr(self, 'release'):
 602             self.assertIn(expected_output, sys.stdout.getvalue())
 603         else:
 604             self.assertNotIn(message_lead, sys.stdout.getvalue())
 605
 606     def test_emits_debian_release_debug_message_only_if_nonnative(self):
 607         """ Should emit message for Debian release only if non-native. """
 608         self.test_args['debug'] = True
 609         debian_release = getattr(self, 'release', None)
 610         message_lead = "D: Debian Version:"
 611         expected_output = textwrap.dedent("""\
 612                 {lead} {version}
 613                 """).format(
 614                     lead=message_lead, version=debian_release)
 615         dput.dput.source_check(**self.test_args)
 616         if hasattr(self, 'release'):
 617             self.assertIn(expected_output, sys.stdout.getvalue())
 618         else:
 619             self.assertNotIn(message_lead, sys.stdout.getvalue())
 620
 621 \f
 622 class verify_files_TestCase(
 623         testscenarios.WithScenarios,
 624         testtools.TestCase):
 625     """ Test cases for `verify_files` function. """
 626
 627     default_args = dict(
 628             host="foo",
 629             check_only=None,
 630             check_version=None,
 631             unsigned_upload=None,
 632             debug=None,
 633             )
 634
 635     scenarios = [
 636             ('default', {}),
 637             ('binary-only', {
 638                 'check_upload_variant_return_value': False,
 639                 }),
 640             ('include foo.tar.gz', {
 641                 'additional_file_suffixes': [".tar.gz"],
 642                 'source_check_result': SourceCheckResult(
 643                     include_orig=False, include_tar=True),
 644                 }),
 645             ('include foo.orig.tar.gz', {
 646                 'additional_file_suffixes': [".orig.tar.gz"],
 647                 'source_check_result': SourceCheckResult(
 648                     include_orig=True, include_tar=False),
 649                 }),
 650             ('unexpected foo.tar.gz', {
 651                 'additional_file_suffixes': [".tar.gz"],
 652                 'expected_rejection_message': (
 653                     "Package includes a .tar.gz file although"),
 654                 }),
 655             ('unexpected foo.orig.tar.gz', {
 656                 'additional_file_suffixes': [".orig.tar.gz"],
 657                 'expected_rejection_message': (
 658                     "Package includes an .orig.tar.gz file although"),
 659                 }),
 660             ('no distribution', {
 661                 'test_distribution': None,
 662                 }),
 663             ]
 664
 665     def setUp(self):
 666         """ Set up test fixtures. """
 667         super(verify_files_TestCase, self).setUp()
 668         patch_system_interfaces(self)
 669
 670         self.file_double_by_path = {}
 671         set_config(self, 'exist-simple')
 672         patch_runtime_config_options(self)
 673
 674         self.set_test_args()
 675
 676         setup_changes_file_fixtures(self)
 677         set_changes_file_scenario(self, 'exist-minimal')
 678         self.test_args.update(dict(
 679                 path=os.path.dirname(self.changes_file_double.path),
 680                 filename=os.path.basename(self.changes_file_double.path),
 681                 ))
 682
 683         patch_os_stat(self)
 684
 685         setup_upload_file_fixtures(self)
 686         self.set_expected_files_to_upload()
 687
 688         self.patch_checksum_test()
 689         self.patch_parse_changes()
 690         self.patch_check_upload_variant()
 691         self.set_expected_binary_upload()
 692         self.set_expected_source_control_file_path()
 693         self.patch_version_check()
 694         self.patch_verify_signature()
 695         self.patch_source_check()
 696
 697     def set_expected_files_to_upload(self):
 698         """ Set the expected `files_to_upload` result for this test case. """
 699         self.expected_files_to_upload = set(
 700                 path for path in self.fake_upload_file_paths)
 701         self.expected_files_to_upload.add(self.changes_file_double.path)
 702
 703     def patch_checksum_test(self):
 704         """ Patch `checksum_test` function for this test case. """
 705         func_patcher = mock.patch.object(
 706                 dput.dput, "checksum_test", autospec=True)
 707         mock_func = func_patcher.start()
 708         self.addCleanup(func_patcher.stop)
 709
 710         def get_checksum_for_file(path, hash_name):
 711             return self.fake_checksum_by_file[os.path.basename(path)]
 712         mock_func.side_effect = get_checksum_for_file
 713
 714     def set_changes_document(self):
 715         """ Set the changes document for this test case. """
 716         self.changes_document = make_changes_document(
 717                 fields={},
 718                 upload_params_by_name=self.upload_params_by_name)
 719         self.test_distribution = getattr(self, 'test_distribution', "lorem")
 720         if self.test_distribution is not None:
 721             self.changes_document.add_header(
 722                     'distribution', self.test_distribution)
 723             self.runtime_config_parser.set(
 724                     self.test_args['host'], 'allowed_distributions',
 725                     self.test_distribution)
 726
 727         dput.dput.parse_changes.return_value = self.changes_document
 728
 729     def set_upload_params(self):
 730         """ Set the upload parameters for this test case. """
 731         self.upload_params_by_name = make_upload_files_params(
 732                 self.fake_checksum_by_file,
 733                 self.fake_size_by_file)
 734
 735     def patch_parse_changes(self):
 736         """ Patch `parse_changes` function for this test case. """
 737         func_patcher = mock.patch.object(
 738                 dput.dput, "parse_changes", autospec=True)
 739         func_patcher.start()
 740         self.addCleanup(func_patcher.stop)
 741
 742         self.set_upload_params()
 743         self.set_changes_document()
 744
 745     def patch_check_upload_variant(self):
 746         """ Patch `check_upload_variant` function for this test case. """
 747         if not hasattr(self, 'check_upload_variant_return_value'):
 748             self.check_upload_variant_return_value = True
 749
 750         func_patcher = mock.patch.object(
 751                 dput.dput, "check_upload_variant", autospec=True,
 752                 return_value=self.check_upload_variant_return_value)
 753         func_patcher.start()
 754         self.addCleanup(func_patcher.stop)
 755
 756     def patch_version_check(self):
 757         """ Patch `version_check` function for this test case. """
 758         func_patcher = mock.patch.object(
 759                 dput.dput, "version_check", autospec=True)
 760         func_patcher.start()
 761         self.addCleanup(func_patcher.stop)
 762
 763     def patch_verify_signature(self):
 764         """ Patch `verify_signature` function for this test case. """
 765         func_patcher = mock.patch.object(
 766                 dput.dput, "verify_signature", autospec=True)
 767         func_patcher.start()
 768         self.addCleanup(func_patcher.stop)
 769
 770     def patch_source_check(self):
 771         """ Patch `source_check` function for this test case. """
 772         func_patcher = mock.patch.object(
 773                 dput.dput, "source_check", autospec=True)
 774         mock_func = func_patcher.start()
 775         self.addCleanup(func_patcher.stop)
 776
 777         source_check_result = getattr(
 778                 self, 'source_check_result', SourceCheckResult(
 779                     include_orig=False, include_tar=False))
 780         mock_func.return_value = source_check_result
 781
 782     def set_test_args(self):
 783         """ Set test args for this test case. """
 784         extra_args = getattr(self, 'extra_args', {})
 785         self.test_args = self.default_args.copy()
 786         self.test_args['config'] = self.runtime_config_parser
 787         self.test_args.update(extra_args)
 788
 789     def set_expected_binary_upload(self):
 790         """ Set expected value for `binary_upload` flag. """
 791         self.expected_binary_upload = self.check_upload_variant_return_value
 792
 793     def set_expected_source_control_file_path(self):
 794         """ Set expected value for source control file path. """
 795         file_name = next(
 796                 os.path.basename(file_path)
 797                 for file_path in self.fake_upload_file_paths
 798                 if file_path.endswith(".dsc"))
 799         if not self.expected_binary_upload:
 800             self.expected_source_control_file_path = os.path.join(
 801                     os.path.dirname(self.changes_file_double.path), file_name)
 802         else:
 803             self.expected_source_control_file_path = ""
 804
 805     def test_emits_changes_file_path_debug_message(self):
 806         """ Should emit debug message for changes file path. """
 807         self.test_args['debug'] = True
 808         dput.dput.verify_files(**self.test_args)
 809         expected_output = textwrap.dedent("""\
 810                 D: Validating contents of changes file {path}
 811                 """).format(path=self.changes_file_double.path)
 812         self.assertIn(expected_output, sys.stdout.getvalue())
 813
 814     def test_calls_sys_exit_if_input_read_denied(self):
 815         """ Should call `sys.exit` if input file read access is denied. """
 816         set_changes_file_scenario(self, 'error-read-denied')
 817         with testtools.ExpectedException(FakeSystemExit):
 818             dput.dput.verify_files(**self.test_args)
 819         expected_output = textwrap.dedent("""\
 820                 Can't open {path}
 821                 """).format(path=self.changes_file_double.path)
 822         self.assertIn(expected_output, sys.stdout.getvalue())
 823         sys.exit.assert_called_with(EXIT_STATUS_FAILURE)
 824
 825     def test_calls_parse_changes_with_changes_files(self):
 826         """ Should call `parse_changes` with changes file. """
 827         dput.dput.verify_files(**self.test_args)
 828         dput.dput.parse_changes.assert_called_with(
 829                 self.changes_file_double.fake_file)
 830
 831     def test_calls_check_upload_variant_with_changes_document(self):
 832         """ Should call `check_upload_variant` with changes document. """
 833         dput.dput.verify_files(**self.test_args)
 834         dput.dput.check_upload_variant.assert_called_with(
 835                 self.changes_document, mock.ANY)
 836
 837     def test_emits_upload_dsc_file_debug_message(self):
 838         """ Should emit debug message for ‘*.dsc’ file. """
 839         if getattr(self, 'check_upload_variant_return_value', True):
 840             self.skipTest("Binary package upload for this scenario")
 841         self.test_args['debug'] = True
 842         dput.dput.verify_files(**self.test_args)
 843         dsc_file_path = next(
 844                 os.path.basename(file_path)
 845                 for file_path in self.fake_upload_file_paths
 846                 if file_path.endswith(".dsc"))
 847         expected_output = textwrap.dedent("""\
 848                 D: dsc-File: {path}
 849                 """).format(path=dsc_file_path)
 850         self.assertThat(
 851                 sys.stdout.getvalue(),
 852                 testtools.matchers.Contains(expected_output))
 853
 854     def test_calls_sys_exit_when_source_upload_omits_dsc_file(self):
 855         """ Should call `sys.exit` when source upload omits ‘*.dsc’ file. """
 856         if getattr(self, 'check_upload_variant_return_value', True):
 857             self.skipTest("Binary package upload for this scenario")
 858         self.fake_checksum_by_file = dict(
 859                 (file_path, checksum)
 860                 for (file_path, checksum)
 861                 in self.fake_checksum_by_file.items()
 862                 if not file_path.endswith(".dsc"))
 863         self.set_upload_params()
 864         self.set_changes_document()
 865         with testtools.ExpectedException(FakeSystemExit):
 866             dput.dput.verify_files(**self.test_args)
 867         expected_output = textwrap.dedent("""\
 868                 Error: no dsc file found in sourceful upload
 869                 """)
 870         self.assertThat(
 871                 sys.stderr.getvalue(),
 872                 testtools.matchers.Contains(expected_output))
 873         sys.exit.assert_called_with(EXIT_STATUS_FAILURE)
 874
 875     def test_calls_version_check_when_specified_in_config(self):
 876         """ Should call `version_check` when specified in config. """
 877         self.runtime_config_parser.set(
 878                 self.test_args['host'], 'check_version', "true")
 879         dput.dput.verify_files(**self.test_args)
 880         dput.dput.version_check.assert_called_with(
 881                 os.path.dirname(self.changes_file_double.path),
 882                 self.changes_document,
 883                 self.test_args['debug'])
 884
 885     def test_calls_version_check_when_specified_in_args(self):
 886         """ Should call `version_check` when specified in arguments. """
 887         self.test_args['check_version'] = True
 888         dput.dput.verify_files(**self.test_args)
 889         dput.dput.version_check.assert_called_with(
 890                 os.path.dirname(self.changes_file_double.path),
 891                 self.changes_document,
 892                 self.test_args['debug'])
 893
 894     def test_calls_sys_exit_when_host_section_not_in_config(self):
 895         """ Should call `sys.exit` when specified host not in config. """
 896         self.runtime_config_parser.remove_section(self.test_args['host'])
 897         with testtools.ExpectedException(FakeSystemExit):
 898             dput.dput.verify_files(**self.test_args)
 899         expected_output = textwrap.dedent("""\
 900                 Error in config file:
 901                 No section: ...
 902                 """)
 903         self.assertThat(
 904                 sys.stderr.getvalue(),
 905                 testtools.matchers.DocTestMatches(
 906                     expected_output, doctest.ELLIPSIS))
 907         sys.exit.assert_called_with(EXIT_STATUS_FAILURE)
 908
 909     def test_calls_verify_signature_with_expected_args(self):
 910         """ Should call `verify_signature` with expected args. """
 911         dput.dput.verify_files(**self.test_args)
 912         dput.dput.verify_signature.assert_called_with(
 913                 self.test_args['host'],
 914                 self.changes_file_double.path,
 915                 self.expected_source_control_file_path,
 916                 self.runtime_config_parser,
 917                 self.test_args['check_only'],
 918                 self.test_args['unsigned_upload'], mock.ANY,
 919                 self.test_args['debug'])
 920
 921     def test_calls_source_check_with_changes_document(self):
 922         """ Should call `source_check` with changes document. """
 923         dput.dput.verify_files(**self.test_args)
 924         dput.dput.source_check.assert_called_with(
 925                 self.changes_document, self.test_args['debug'])
 926
 927     def test_emits_upload_file_path_debug_message(self):
 928         """ Should emit debug message for each upload file path. """
 929         self.test_args['debug'] = True
 930         dput.dput.verify_files(**self.test_args)
 931         for file_path in self.fake_upload_file_paths:
 932             expected_output = textwrap.dedent("""\
 933                     D: File to upload: {path}
 934                     """).format(path=file_path)
 935             self.expectThat(
 936                     sys.stdout.getvalue(),
 937                     testtools.matchers.Contains(expected_output))
 938
 939     def test_calls_checksum_test_with_upload_files(self):
 940         """ Should call `checksum_test` with each upload file path. """
 941         dput.dput.verify_files(**self.test_args)
 942         expected_calls = [
 943                 mock.call(file_path, mock.ANY)
 944                 for file_path in self.fake_upload_file_paths]
 945         dput.dput.checksum_test.assert_has_calls(
 946                 expected_calls, any_order=True)
 947
 948     def set_bogus_file_checksums(self):
 949         """ Set bogus file checksums that will not match. """
 950         self.fake_checksum_by_file = {
 951                 file_name: self.getUniqueString()
 952                 for file_name in self.fake_checksum_by_file}
 953
 954     def test_emits_checksum_okay_debug_message(self):
 955         """ Should emit debug message checksum okay for each file. """
 956         self.test_args['debug'] = True
 957         dput.dput.verify_files(**self.test_args)
 958         for file_path in self.fake_upload_file_paths:
 959             expected_output = textwrap.dedent("""\
 960                     D: Checksum for {path} is fine
 961                     """).format(path=file_path)
 962             self.expectThat(
 963                     sys.stdout.getvalue(),
 964                     testtools.matchers.Contains(expected_output))
 965
 966     def test_emits_checksum_mismatch_debug_message(self):
 967         """ Should emit debug message when a checksum does not match. """
 968         self.test_args['debug'] = True
 969         self.set_bogus_file_checksums()
 970         with testtools.ExpectedException(FakeSystemExit):
 971             dput.dput.verify_files(**self.test_args)
 972         expected_output = textwrap.dedent("""\
 973                 ...
 974                 D: Checksum from .changes: ...
 975                 D: Generated Checksum: ...
 976                 ...
 977                 """)
 978         self.assertThat(
 979                 sys.stdout.getvalue(),
 980                 testtools.matchers.DocTestMatches(
 981                     expected_output, doctest.ELLIPSIS))
 982
 983     def test_calls_sys_exit_when_checksum_mismatch(self):
 984         """ Should call `sys.exit` when a checksum does not match. """
 985         specified_checksum_by_file = self.fake_checksum_by_file
 986         self.set_bogus_file_checksums()
 987         with testtools.ExpectedException(FakeSystemExit):
 988             dput.dput.verify_files(**self.test_args)
 989
 990         expected_output_for_files = [
 991                 textwrap.dedent("""\
 992                     Checksum doesn't match for {file_name}
 993                     """).format(
 994                         file_name=os.path.join(
 995                             os.path.dirname(self.changes_file_double.path),
 996                             file_name),
 997                         specified_hash=specified_hash,
 998                         computed_hash=self.fake_checksum_by_file[file_name])
 999                 for (file_name, specified_hash)
1000                 in specified_checksum_by_file.items()]
1001         self.assertThat(
1002                 sys.stdout.getvalue(),
1003                 testtools.matchers.MatchesAny(*[
1004                     testtools.matchers.Contains(expected_output)
1005                     for expected_output in expected_output_for_files]))
1006         sys.exit.assert_called_with(EXIT_STATUS_FAILURE)
1007
1008     def test_calls_os_stat_with_upload_files(self):
1009         """ Should call `os.stat` with each upload file path. """
1010         dput.dput.verify_files(**self.test_args)
1011         expected_calls = [
1012                 mock.call(file_path)
1013                 for file_path in self.fake_upload_file_paths]
1014         os.stat.assert_has_calls(expected_calls, any_order=True)
1015
1016     def set_bogus_file_sizes(self):
1017         """ Set bogus file sizes that will not match. """
1018         file_double_registry = FileDouble.get_registry_for_testcase(self)
1019         for file_name in self.fake_size_by_file:
1020                 bogus_size = self.getUniqueInteger()
1021                 self.fake_size_by_file[file_name] = bogus_size
1022                 file_path = os.path.join(
1023                         os.path.dirname(self.changes_file_double.path),
1024                         file_name)
1025                 file_double = file_double_registry[file_path]
1026                 file_double.stat_result = file_double.stat_result._replace(
1027                         st_size=bogus_size)
1028
1029     def test_emits_size_mismatch_debug_message(self):
1030         """ Should emit debug message when a size does not match. """
1031         self.test_args['debug'] = True
1032         self.set_bogus_file_sizes()
1033         dput.dput.verify_files(**self.test_args)
1034         expected_output = textwrap.dedent("""\
1035                 ...
1036                 D: size from .changes: ...
1037                 D: calculated size: ...
1038                 ...
1039                 """)
1040         self.assertThat(
1041                 sys.stdout.getvalue(),
1042                 testtools.matchers.DocTestMatches(
1043                     expected_output, doctest.ELLIPSIS))
1044
1045     def test_emits_size_mismatch_message_for_each_file(self):
1046         """ Should emit error message for each file with size mismatch. """
1047         self.set_bogus_file_sizes()
1048         dput.dput.verify_files(**self.test_args)
1049         for file_path in self.fake_upload_file_paths:
1050             expected_output = textwrap.dedent("""\
1051                     size doesn't match for {path}
1052                     """).format(path=file_path)
1053             self.expectThat(
1054                     sys.stdout.getvalue(),
1055                     testtools.matchers.Contains(expected_output))
1056
1057     def test_emits_rejection_warning_when_unexpected_tarball(self):
1058         """ Should emit warning of rejection when unexpected tarball. """
1059         if not hasattr(self, 'expected_rejection_message'):
1060             self.skipTest("No rejection message expected")
1061         dput.dput.verify_files(**self.test_args)
1062         sys.stderr.write("calls: {calls!r}\n".format(
1063                 calls=sys.stdout.write.mock_calls))
1064         self.assertThat(
1065                 sys.stdout.getvalue(),
1066                 testtools.matchers.Contains(self.expected_rejection_message))
1067
1068     def test_raises_error_when_distribution_mismatch(self):
1069         """ Should raise error when distribution mismatch against allowed. """
1070         if not getattr(self, 'test_distribution', None):
1071             self.skipTest("No distribution set for this test case")
1072         self.runtime_config_parser.set(
1073                 self.test_args['host'], 'allowed_distributions',
1074                 "dolor sit amet")
1075         with testtools.ExpectedException(dputhelper.DputUploadFatalException):
1076             dput.dput.verify_files(**self.test_args)
1077
1078     def test_emits_changes_file_upload_debug_message(self):
1079         """ Should emit debug message for upload of changes file. """
1080         self.test_args['debug'] = True
1081         dput.dput.verify_files(**self.test_args)
1082         expected_output = textwrap.dedent("""\
1083                 D: File to upload: {path}
1084                 """).format(path=self.changes_file_double.path)
1085         self.assertIn(expected_output, sys.stdout.getvalue())
1086
1087     def test_returns_expected_files_to_upload_collection(self):
1088         """ Should return expected `files_to_upload` collection value. """
1089         result = dput.dput.verify_files(**self.test_args)
1090         expected_result = self.expected_files_to_upload
1091         self.assertEqual(expected_result, set(result))
1092
1093 \f
1094 class guess_upload_host_TestCase(
1095         testscenarios.WithScenarios,
1096         testtools.TestCase):
1097     """ Test cases for `guess_upload_host` function. """
1098
1099     changes_file_scenarios = [
1100             ('no-distribution', {
1101                 'fake_file': StringIO(textwrap.dedent("""\
1102                     Files:
1103                         Lorem ipsum dolor sit amet
1104                     """)),
1105                 }),
1106             ('distribution-spam', {
1107                 'fake_file': StringIO(textwrap.dedent("""\
1108                     Distribution: spam
1109                     Files:
1110                         Lorem ipsum dolor sit amet
1111                     """)),
1112                 }),
1113             ('distribution-beans', {
1114                 'fake_file': StringIO(textwrap.dedent("""\
1115                     Distribution: beans
1116                     Files:
1117                         Lorem ipsum dolor sit amet
1118                     """)),
1119                 }),
1120             ]
1121
1122     scenarios = [
1123             ('distribution-found-of-one', {
1124                 'changes_file_scenario_name': "distribution-spam",
1125                 'test_distribution': "spam",
1126                 'config_scenario_name': "exist-distribution-one",
1127                 'expected_host': "foo",
1128                 }),
1129             ('distribution-notfound-of-one', {
1130                 'changes_file_scenario_name': "distribution-beans",
1131                 'test_distribution': "beans",
1132                 'config_scenario_name': "exist-distribution-one",
1133                 'expected_host': "ftp-master",
1134                 }),
1135             ('distribution-first-of-three', {
1136                 'changes_file_scenario_name': "distribution-spam",
1137                 'test_distribution': "spam",
1138                 'config_scenario_name': "exist-distribution-three",
1139                 'expected_host': "foo",
1140                 }),
1141             ('distribution-last-of-three', {
1142                 'changes_file_scenario_name': "distribution-beans",
1143                 'test_distribution': "beans",
1144                 'config_scenario_name': "exist-distribution-three",
1145                 'expected_host': "foo",
1146                 }),
1147             ('no-configured-distribution', {
1148                 'changes_file_scenario_name': "distribution-beans",
1149                 'config_scenario_name': "exist-distribution-none",
1150                 'expected_host': "ftp-master",
1151                 }),
1152             ('no-distribution', {
1153                 'changes_file_scenario_name': "no-distribution",
1154                 'config_scenario_name': "exist-simple",
1155                 'expected_host': "ftp-master",
1156                 }),
1157             ('default-distribution', {
1158                 'config_scenario_name': "exist-default-distribution-only",
1159                 'config_default_default_host_main': "consecteur",
1160                 'expected_host': "consecteur",
1161                 }),
1162             ]
1163
1164     def setUp(self):
1165         """ Set up test fixtures. """
1166         super(guess_upload_host_TestCase, self).setUp()
1167         patch_system_interfaces(self)
1168
1169         set_config(
1170                 self,
1171                 getattr(self, 'config_scenario_name', 'exist-minimal'))
1172         patch_runtime_config_options(self)
1173
1174         self.setup_changes_file_fixtures()
1175         set_changes_file_scenario(
1176                 self,
1177                 getattr(self, 'changes_file_scenario_name', 'no-distribution'))
1178
1179         self.set_test_args()
1180
1181     def set_test_args(self):
1182         """ Set the arguments for the test call to the function. """
1183         self.test_args = dict(
1184                 path=os.path.dirname(self.changes_file_double.path),
1185                 filename=os.path.basename(self.changes_file_double.path),
1186                 config=self.runtime_config_parser,
1187                 )
1188
1189     def setup_changes_file_fixtures(self):
1190         """ Set up fixtures for fake changes file. """
1191         file_path = make_changes_file_path()
1192
1193         scenarios = [s for (__, s) in self.changes_file_scenarios]
1194         for scenario in scenarios:
1195             scenario['file_double'] = FileDouble(
1196                     file_path, scenario['fake_file'])
1197         setup_file_double_behaviour(
1198                 self,
1199                 get_file_doubles_from_fake_file_scenarios(scenarios))
1200
1201     def test_calls_sys_exit_if_read_denied(self):
1202         """ Should call `sys.exit` if read permission denied. """
1203         self.changes_file_double.set_os_access_scenario('denied')
1204         self.changes_file_double.set_open_scenario('read_denied')
1205         with testtools.ExpectedException(FakeSystemExit):
1206             dput.dput.guess_upload_host(**self.test_args)
1207         expected_output = textwrap.dedent("""\
1208                 Can't open {path}
1209                 """).format(path=self.changes_file_double.path)
1210         self.assertIn(expected_output, sys.stdout.getvalue())
1211         sys.exit.assert_called_with(EXIT_STATUS_FAILURE)
1212
1213     def test_returns_expected_host(self):
1214         """ Should return expected host value. """
1215         result = dput.dput.guess_upload_host(**self.test_args)
1216         self.assertEqual(self.expected_host, result)
1217
1218     @mock.patch.object(dput.dput, 'debug', True)
1219     def test_emits_debug_message_for_host(self):
1220         """ Should emit a debug message for the discovered host. """
1221         config_parser = self.runtime_config_parser
1222         if not (
1223                 config_parser.has_section(self.expected_host)
1224                 and config_parser.get(self.expected_host, 'distributions')):
1225             self.skipTest("No distributions specified")
1226         dput.dput.guess_upload_host(**self.test_args)
1227         expected_output = textwrap.dedent("""\
1228                 D: guessing host {host} based on distribution {dist}
1229                 """).format(
1230                     host=self.expected_host, dist=self.test_distribution)
1231         self.assertIn(expected_output, sys.stdout.getvalue())
1232
1233 \f
1234 # Copyright © 2015–2016 Ben Finney <bignose@debian.org>
1235 #
1236 # This is free software: you may copy, modify, and/or distribute this work
1237 # under the terms of the GNU General Public License as published by the
1238 # Free Software Foundation; version 3 of that license or any later version.
1239 # No warranty expressed or implied. See the file ‘LICENSE.GPL-3’ for details.
1240
1241 \f
1242 # Local variables:
1243 # coding: utf-8
1244 # mode: python
1245 # End:
1246 # vim: fileencoding=utf-8 filetype=python :