src/mailman/utilities/tests/test_import.py

   1 # Copyright (C) 2010-2015 by the Free Software Foundation, Inc.
   2 #
   3 # This file is part of GNU Mailman.
   4 #
   5 # GNU Mailman is free software: you can redistribute it and/or modify it under
   6 # the terms of the GNU General Public License as published by the Free
   7 # Software Foundation, either version 3 of the License, or (at your option)
   8 # any later version.
   9 #
  10 # GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
  11 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  12 # FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
  13 # more details.
  14 #
  15 # You should have received a copy of the GNU General Public License along with
  16 # GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
  17
  18 """Tests for config.pck imports."""
  19
  20 __all__ = [
  21     'TestArchiveImport',
  22     'TestBasicImport',
  23     'TestConvertToURI',
  24     'TestFilterActionImport',
  25     'TestMemberActionImport',
  26     'TestPreferencesImport',
  27     'TestRosterImport',
  28     ]
  29
  30
  31 import os
  32 import mock
  33 import unittest
  34
  35 from datetime import timedelta, datetime
  36 from enum import Enum
  37 from mailman.app.lifecycle import create_list
  38 from mailman.config import config
  39 from mailman.handlers.decorate import decorate
  40 from mailman.interfaces.action import Action, FilterAction
  41 from mailman.interfaces.address import InvalidEmailAddressError
  42 from mailman.interfaces.archiver import ArchivePolicy
  43 from mailman.interfaces.autorespond import ResponseAction
  44 from mailman.interfaces.bans import IBanManager
  45 from mailman.interfaces.bounce import UnrecognizedBounceDisposition
  46 from mailman.interfaces.languages import ILanguageManager
  47 from mailman.interfaces.mailinglist import (
  48     IAcceptableAliasSet, SubscriptionPolicy)
  49 from mailman.interfaces.member import DeliveryMode, DeliveryStatus
  50 from mailman.interfaces.nntp import NewsgroupModeration
  51 from mailman.interfaces.templates import ITemplateLoader
  52 from mailman.interfaces.usermanager import IUserManager
  53 from mailman.testing.layers import ConfigLayer
  54 from mailman.utilities.filesystem import makedirs
  55 from mailman.utilities.importer import import_config_pck, Import21Error
  56 from mailman.utilities.string import expand
  57 from pickle import load
  58 from pkg_resources import resource_filename
  59 from zope.component import getUtility
  60
  61
  62 \f
  63 NL = '\n'
  64
  65
  66 class DummyEnum(Enum):
  67     # For testing purposes
  68     val = 42
  69
  70
  71 def list_to_string(data):
  72     return NL.join(data).encode('utf-8')
  73
  74
  75 \f
  76 class TestBasicImport(unittest.TestCase):
  77     layer = ConfigLayer
  78
  79     def setUp(self):
  80         self._mlist = create_list('blank@example.com')
  81         pickle_file = resource_filename('mailman.testing', 'config.pck')
  82         with open(pickle_file, 'rb') as fp:
  83             self._pckdict = load(fp)
  84
  85     def _import(self):
  86         import_config_pck(self._mlist, self._pckdict)
  87
  88     def test_display_name(self):
  89         # The mlist.display_name gets set from the old list's real_name.
  90         self.assertEqual(self._mlist.display_name, 'Blank')
  91         self._import()
  92         self.assertEqual(self._mlist.display_name, 'Test')
  93
  94     def test_mail_host_invariant(self):
  95         # The mlist.mail_host must not be updated when importing (it will
  96         # change the list_id property, which is supposed to be read-only).
  97         self.assertEqual(self._mlist.mail_host, 'example.com')
  98         self._import()
  99         self.assertEqual(self._mlist.mail_host, 'example.com')
 100
 101     def test_rfc2369_headers(self):
 102         self._mlist.allow_list_posts = False
 103         self._mlist.include_rfc2369_headers = False
 104         self._import()
 105         self.assertTrue(self._mlist.allow_list_posts)
 106         self.assertTrue(self._mlist.include_rfc2369_headers)
 107
 108     def test_no_overwrite_rosters(self):
 109         # The mlist.members and mlist.digest_members rosters must not be
 110         # overwritten.
 111         for rname in ('members', 'digest_members'):
 112             roster = getattr(self._mlist, rname)
 113             self.assertFalse(isinstance(roster, dict))
 114             # Suppress warning messages in test output.
 115             with mock.patch('sys.stderr'):
 116                 self._import()
 117             self.assertFalse(
 118                 isinstance(roster, dict),
 119                 'The %s roster has been overwritten by the import' % rname)
 120
 121     def test_last_post_time(self):
 122         # last_post_time -> last_post_at
 123         self._pckdict['last_post_time'] = 1270420800.274485
 124         self.assertEqual(self._mlist.last_post_at, None)
 125         self._import()
 126         # convert 1270420800.2744851 to datetime
 127         expected = datetime(2010, 4, 4, 22, 40, 0, 274485)
 128         self.assertEqual(self._mlist.last_post_at, expected)
 129
 130     def test_autoresponse_grace_period(self):
 131         # autoresponse_graceperiod -> autoresponse_grace_period
 132         # must be a timedelta, not an int
 133         self._mlist.autoresponse_grace_period = timedelta(days=42)
 134         self._import()
 135         self.assertTrue(
 136             isinstance(self._mlist.autoresponse_grace_period, timedelta))
 137         self.assertEqual(self._mlist.autoresponse_grace_period,
 138                          timedelta(days=90))
 139
 140     def test_autoresponse_admin_to_owner(self):
 141         # admin -> owner
 142         self._mlist.autorespond_owner = DummyEnum.val
 143         self._mlist.autoresponse_owner_text = 'DUMMY'
 144         self._import()
 145         self.assertEqual(self._mlist.autorespond_owner, ResponseAction.none)
 146         self.assertEqual(self._mlist.autoresponse_owner_text, '')
 147
 148     def test_administrativia(self):
 149        self._mlist.administrivia = None
 150        self._import()
 151        self.assertTrue(self._mlist.administrivia)
 152
 153     def test_filter_pass_renames(self):
 154         # mime_types -> types
 155         # filename_extensions -> extensions
 156         self._mlist.filter_types = ['dummy']
 157         self._mlist.pass_types = ['dummy']
 158         self._mlist.filter_extensions = ['dummy']
 159         self._mlist.pass_extensions = ['dummy']
 160         self._import()
 161         self.assertEqual(list(self._mlist.filter_types), [])
 162         self.assertEqual(list(self._mlist.filter_extensions),
 163                          ['exe', 'bat', 'cmd', 'com', 'pif',
 164                           'scr', 'vbs', 'cpl'])
 165         self.assertEqual(list(self._mlist.pass_types),
 166                 ['multipart/mixed', 'multipart/alternative', 'text/plain'])
 167         self.assertEqual(list(self._mlist.pass_extensions), [])
 168
 169     def test_process_bounces(self):
 170         # bounce_processing -> process_bounces
 171         self._mlist.process_bounces = None
 172         self._import()
 173         self.assertTrue(self._mlist.process_bounces)
 174
 175     def test_forward_unrecognized_bounces_to(self):
 176         # bounce_unrecognized_goes_to_list_owner
 177         #   -> forward_unrecognized_bounces_to
 178         self._mlist.forward_unrecognized_bounces_to = DummyEnum.val
 179         self._import()
 180         self.assertEqual(self._mlist.forward_unrecognized_bounces_to,
 181                          UnrecognizedBounceDisposition.administrators)
 182
 183     def test_moderator_password(self):
 184         # mod_password -> moderator_password
 185         self._mlist.moderator_password = b'TESTDATA'
 186         self._import()
 187         self.assertEqual(self._mlist.moderator_password, None)
 188
 189     def test_moderator_password_str(self):
 190         # moderator_password must not be unicode
 191         self._pckdict['mod_password'] = b'TESTVALUE'
 192         self._import()
 193         self.assertNotIsInstance(self._mlist.moderator_password, str)
 194         self.assertEqual(self._mlist.moderator_password, b'TESTVALUE')
 195
 196     def test_newsgroup_moderation(self):
 197         # news_moderation -> newsgroup_moderation
 198         # news_prefix_subject_too -> nntp_prefix_subject_too
 199         self._mlist.newsgroup_moderation = DummyEnum.val
 200         self._mlist.nntp_prefix_subject_too = None
 201         self._import()
 202         self.assertEqual(self._mlist.newsgroup_moderation,
 203                          NewsgroupModeration.none)
 204         self.assertTrue(self._mlist.nntp_prefix_subject_too)
 205
 206     def test_msg_to_message(self):
 207         # send_welcome_msg -> send_welcome_message
 208         # send_goodbye_msg -> send_goodbye_message
 209         self._mlist.send_welcome_message = None
 210         self._mlist.send_goodbye_message = None
 211         self._import()
 212         self.assertTrue(self._mlist.send_welcome_message)
 213         self.assertTrue(self._mlist.send_goodbye_message)
 214
 215     def test_ban_list(self):
 216         banned = [
 217             ('anne@example.com', 'anne@example.com'),
 218             ('^.*@example.com', 'bob@example.com'),
 219             ('non-ascii-\xe8@example.com', 'non-ascii-\ufffd@example.com'),
 220             ]
 221         self._pckdict['ban_list'] = [b[0].encode('iso-8859-1') for b in banned]
 222         self._import()
 223         for _pattern, addr in banned:
 224             self.assertTrue(IBanManager(self._mlist).is_banned(addr))
 225
 226     def test_acceptable_aliases(self):
 227         # This used to be a plain-text field (values are newline-separated).
 228         aliases = ['alias1@example.com',
 229                    'alias2@exemple.com',
 230                    'non-ascii-\xe8@example.com',
 231                    ]
 232         self._pckdict['acceptable_aliases'] = list_to_string(aliases)
 233         self._import()
 234         alias_set = IAcceptableAliasSet(self._mlist)
 235         self.assertEqual(sorted(alias_set.aliases), aliases)
 236
 237     def test_acceptable_aliases_invalid(self):
 238         # Values without an '@' sign used to be matched against the local
 239         # part, now we need to add the '^' sign to indicate it's a regexp.
 240         aliases = ['invalid-value']
 241         self._pckdict['acceptable_aliases'] = list_to_string(aliases)
 242         self._import()
 243         alias_set = IAcceptableAliasSet(self._mlist)
 244         self.assertEqual(sorted(alias_set.aliases),
 245                          [('^' + alias) for alias in aliases])
 246
 247     def test_acceptable_aliases_as_list(self):
 248         # In some versions of the pickle, this can be a list, not a string
 249         # (seen in the wild).
 250         aliases = [b'alias1@example.com', b'alias2@exemple.com' ]
 251         self._pckdict['acceptable_aliases'] = aliases
 252         self._import()
 253         alias_set = IAcceptableAliasSet(self._mlist)
 254         self.assertEqual(sorted(alias_set.aliases),
 255                          sorted(a.decode('utf-8') for a in aliases))
 256
 257     def test_info_non_ascii(self):
 258         # info can contain non-ascii characters.
 259         info = 'O idioma aceito \xe9 somente Portugu\xeas do Brasil'
 260         self._pckdict['info'] = info.encode('utf-8')
 261         self._import()
 262         self.assertEqual(self._mlist.info, info,
 263                          'Encoding to UTF-8 is not handled')
 264         # Test fallback to ascii with replace.
 265         self._pckdict['info'] = info.encode('iso-8859-1')
 266         # Suppress warning messages in test output.
 267         with mock.patch('sys.stderr'):
 268             self._import()
 269         self.assertEqual(
 270             self._mlist.info,
 271             self._pckdict['info'].decode('ascii', 'replace'),
 272             "We don't fall back to replacing non-ascii chars")
 273
 274     def test_preferred_language(self):
 275         self._pckdict['preferred_language'] = b'ja'
 276         english = getUtility(ILanguageManager).get('en')
 277         japanese = getUtility(ILanguageManager).get('ja')
 278         self.assertEqual(self._mlist.preferred_language, english)
 279         self._import()
 280         self.assertEqual(self._mlist.preferred_language, japanese)
 281
 282     def test_preferred_language_unknown_previous(self):
 283         # When the previous language is unknown, it should not fail.
 284         self._mlist._preferred_language = 'xx'
 285         self._import()
 286         english = getUtility(ILanguageManager).get('en')
 287         self.assertEqual(self._mlist.preferred_language, english)
 288
 289     def test_new_language(self):
 290         self._pckdict['preferred_language'] = b'xx_XX'
 291         try:
 292             self._import()
 293         except Import21Error as error:
 294             # Check the message.
 295             self.assertIn('[language.xx_XX]', str(error))
 296         else:                                       # pragma: no cover
 297             self.fail('Import21Error was not raised')
 298
 299     def test_encode_ascii_prefixes(self):
 300         self._pckdict['encode_ascii_prefixes'] = 2
 301         self.assertEqual(self._mlist.encode_ascii_prefixes, False)
 302         self._import()
 303         self.assertEqual(self._mlist.encode_ascii_prefixes, True)
 304
 305     def test_subscription_policy_open(self):
 306         self._mlist.subscription_policy = SubscriptionPolicy.confirm
 307         self._pckdict['subscribe_policy'] = 0
 308         self._import()
 309         self.assertEqual(self._mlist.subscription_policy,
 310                          SubscriptionPolicy.open)
 311
 312     def test_subscription_policy_confirm(self):
 313         self._mlist.subscription_policy = SubscriptionPolicy.open
 314         self._pckdict['subscribe_policy'] = 1
 315         self._import()
 316         self.assertEqual(self._mlist.subscription_policy,
 317                          SubscriptionPolicy.confirm)
 318
 319     def test_subscription_policy_moderate(self):
 320         self._mlist.subscription_policy = SubscriptionPolicy.open
 321         self._pckdict['subscribe_policy'] = 2
 322         self._import()
 323         self.assertEqual(self._mlist.subscription_policy,
 324                          SubscriptionPolicy.moderate)
 325
 326     def test_subscription_policy_confirm_then_moderate(self):
 327         self._mlist.subscription_policy = SubscriptionPolicy.open
 328         self._pckdict['subscribe_policy'] = 3
 329         self._import()
 330         self.assertEqual(self._mlist.subscription_policy,
 331                          SubscriptionPolicy.confirm_then_moderate)
 332
 333
 334 \f
 335 class TestArchiveImport(unittest.TestCase):
 336     """Test conversion of the archive policies.
 337
 338     Mailman 2.1 had two variables `archive` and `archive_private`.  Now
 339     there's just a single `archive_policy` enum.
 340     """
 341     layer = ConfigLayer
 342
 343     def setUp(self):
 344         self._mlist = create_list('blank@example.com')
 345         self._mlist.archive_policy = DummyEnum.val
 346
 347     def _do_test(self, pckdict, expected):
 348         import_config_pck(self._mlist, pckdict)
 349         self.assertEqual(self._mlist.archive_policy, expected)
 350
 351     def test_public(self):
 352         self._do_test(dict(archive=True, archive_private=False),
 353                       ArchivePolicy.public)
 354
 355     def test_private(self):
 356         self._do_test(dict(archive=True, archive_private=True),
 357                       ArchivePolicy.private)
 358
 359     def test_no_archive(self):
 360         self._do_test(dict(archive=False, archive_private=False),
 361                       ArchivePolicy.never)
 362
 363     def test_bad_state(self):
 364         # For some reason, the old list has the invalid archiving state where
 365         # `archive` is False and `archive_private` is True.  It doesn't matter
 366         # because this still collapses to the same enum value.
 367         self._do_test(dict(archive=False, archive_private=True),
 368                       ArchivePolicy.never)
 369
 370     def test_missing_archive_key(self):
 371         # For some reason, the old list didn't have an `archive` key.  We
 372         # treat this as if no archiving is done.
 373         self._do_test(dict(archive_private=False), ArchivePolicy.never)
 374
 375     def test_missing_archive_key_archive_public(self):
 376         # For some reason, the old list didn't have an `archive` key, and it
 377         # has weird value for archive_private.  We treat this as if no
 378         # archiving is done.
 379         self._do_test(dict(archive_private=True), ArchivePolicy.never)
 380
 381     def test_missing_archive_private_key(self):
 382         # For some reason, the old list was missing an `archive_private` key.
 383         # For maximum safety, we treat this as private archiving.
 384         self._do_test(dict(archive=True), ArchivePolicy.private)
 385
 386
 387 \f
 388 class TestFilterActionImport(unittest.TestCase):
 389     # The mlist.filter_action enum values have changed.  In Mailman 2.1 the
 390     # order was 'Discard', 'Reject', 'Forward to List Owner', 'Preserve'.
 391
 392     layer = ConfigLayer
 393
 394     def setUp(self):
 395         self._mlist = create_list('blank@example.com')
 396         self._mlist.filter_action = DummyEnum.val
 397
 398     def _do_test(self, original, expected):
 399         import_config_pck(self._mlist, dict(filter_action=original))
 400         self.assertEqual(self._mlist.filter_action, expected)
 401
 402     def test_discard(self):
 403         self._do_test(0, FilterAction.discard)
 404
 405     def test_reject(self):
 406         self._do_test(1, FilterAction.reject)
 407
 408     def test_forward(self):
 409         self._do_test(2, FilterAction.forward)
 410
 411     def test_preserve(self):
 412         self._do_test(3, FilterAction.preserve)
 413
 414
 415 \f
 416 class TestMemberActionImport(unittest.TestCase):
 417     # The mlist.default_member_action and mlist.default_nonmember_action enum
 418     # values are different in Mailman 2.1; they have been merged into a
 419     # single enum in Mailman 3.
 420     #
 421     # For default_member_action, which used to be called
 422     # member_moderation_action, the values were:
 423     # 0==Hold, 1=Reject, 2==Discard
 424     #
 425     # For default_nonmember_action, which used to be called
 426     # generic_nonmember_action, the values were:
 427     # 0==Accept, 1==Hold, 2==Reject, 3==Discard
 428
 429     layer = ConfigLayer
 430
 431     def setUp(self):
 432         self._mlist = create_list('blank@example.com')
 433         self._mlist.default_member_action = DummyEnum.val
 434         self._mlist.default_nonmember_action = DummyEnum.val
 435         self._pckdict = dict(
 436             member_moderation_action=DummyEnum.val,
 437             generic_nonmember_action=DummyEnum.val,
 438             )
 439
 440     def _do_test(self, expected):
 441         # Suppress warning messages in the test output.
 442         with mock.patch('sys.stderr'):
 443             import_config_pck(self._mlist, self._pckdict)
 444         for key, value in expected.items():
 445             self.assertEqual(getattr(self._mlist, key), value)
 446
 447     def test_member_hold(self):
 448         self._pckdict['member_moderation_action'] = 0
 449         self._do_test(dict(default_member_action=Action.hold))
 450
 451     def test_member_reject(self):
 452         self._pckdict['member_moderation_action'] = 1
 453         self._do_test(dict(default_member_action=Action.reject))
 454
 455     def test_member_discard(self):
 456         self._pckdict['member_moderation_action'] = 2
 457         self._do_test(dict(default_member_action=Action.discard))
 458
 459     def test_nonmember_accept(self):
 460         self._pckdict['generic_nonmember_action'] = 0
 461         self._do_test(dict(default_nonmember_action=Action.accept))
 462
 463     def test_nonmember_hold(self):
 464         self._pckdict['generic_nonmember_action'] = 1
 465         self._do_test(dict(default_nonmember_action=Action.hold))
 466
 467     def test_nonmember_reject(self):
 468         self._pckdict['generic_nonmember_action'] = 2
 469         self._do_test(dict(default_nonmember_action=Action.reject))
 470
 471     def test_nonmember_discard(self):
 472         self._pckdict['generic_nonmember_action'] = 3
 473         self._do_test(dict(default_nonmember_action=Action.discard))
 474
 475
 476 \f
 477 class TestConvertToURI(unittest.TestCase):
 478     # The following values were plain text, and are now URIs in Mailman 3:
 479     # - welcome_message_uri
 480     # - goodbye_message_uri
 481     # - header_uri
 482     # - footer_uri
 483     # - digest_header_uri
 484     # - digest_footer_uri
 485     #
 486     # The templates contain variables that must be replaced:
 487     # - %(real_name)s -> %(display_name)s
 488     # - %(real_name)s@%(host_name)s -> %(fqdn_listname)s
 489     # - %(web_page_url)slistinfo%(cgiext)s/%(_internal_name)s
 490     #       -> %(listinfo_uri)s
 491
 492     layer = ConfigLayer
 493
 494     def setUp(self):
 495         self._mlist = create_list('blank@example.com')
 496         self._conf_mapping = dict(
 497             welcome_msg='welcome_message_uri',
 498             goodbye_msg='goodbye_message_uri',
 499             msg_header='header_uri',
 500             msg_footer='footer_uri',
 501             digest_header='digest_header_uri',
 502             digest_footer='digest_footer_uri',
 503             )
 504         self._pckdict = dict()
 505
 506     def test_text_to_uri(self):
 507         for oldvar, newvar in self._conf_mapping.items():
 508             self._pckdict[str(oldvar)] = b'TEST VALUE'
 509             import_config_pck(self._mlist, self._pckdict)
 510             newattr = getattr(self._mlist, newvar)
 511             text = decorate(self._mlist, newattr)
 512             self.assertEqual(text, 'TEST VALUE',
 513                     'Old variable %s was not properly imported to %s'
 514                     % (oldvar, newvar))
 515
 516     def test_substitutions(self):
 517         test_text = ('UNIT TESTING %(real_name)s mailing list\n'
 518                      '%(real_name)s@%(host_name)s\n'
 519                      '%(web_page_url)slistinfo%(cgiext)s/%(_internal_name)s')
 520         expected_text = ('UNIT TESTING $display_name mailing list\n'
 521                          '$fqdn_listname\n'
 522                          '$listinfo_uri')
 523         for oldvar, newvar in self._conf_mapping.items():
 524             self._pckdict[str(oldvar)] = str(test_text)
 525             import_config_pck(self._mlist, self._pckdict)
 526             newattr = getattr(self._mlist, newvar)
 527             template_uri = expand(newattr, dict(
 528                 listname=self._mlist.fqdn_listname,
 529                 language=self._mlist.preferred_language.code,
 530                 ))
 531             loader = getUtility(ITemplateLoader)
 532             text = loader.get(template_uri)
 533             self.assertEqual(
 534                 text, expected_text,
 535                 'Old variables were not converted for %s' % newvar)
 536
 537     def test_keep_default(self):
 538         # If the value was not changed from MM2.1's default, don't import it.
 539         default_msg_footer = (
 540             '_______________________________________________\n'
 541             '%(real_name)s mailing list\n'
 542             '%(real_name)s@%(host_name)s\n'
 543             '%(web_page_url)slistinfo%(cgiext)s/%(_internal_name)s'
 544             )
 545         for oldvar in ('msg_footer', 'digest_footer'):
 546             newvar = self._conf_mapping[oldvar]
 547             self._pckdict[str(oldvar)] = str(default_msg_footer)
 548             old_value = getattr(self._mlist, newvar)
 549             import_config_pck(self._mlist, self._pckdict)
 550             new_value = getattr(self._mlist, newvar)
 551             self.assertEqual(old_value, new_value,
 552                     'Default value was not preserved for %s' % newvar)
 553
 554     def test_keep_default_if_fqdn_changed(self):
 555         # Use case: importing the old a@ex.com into b@ex.com.  We can't check
 556         # if it changed from the default so don't import.  We may do more harm
 557         # than good and it's easy to change if needed.
 558         test_value = b'TEST-VALUE'
 559         for oldvar, newvar in self._conf_mapping.items():
 560             self._mlist.mail_host = 'example.com'
 561             self._pckdict['mail_host'] = b'test.example.com'
 562             self._pckdict[str(oldvar)] = test_value
 563             old_value = getattr(self._mlist, newvar)
 564             # Suppress warning messages in the test output.
 565             with mock.patch('sys.stderr'):
 566                 import_config_pck(self._mlist, self._pckdict)
 567             new_value = getattr(self._mlist, newvar)
 568             self.assertEqual(old_value, new_value,
 569                 'Default value was not preserved for %s' % newvar)
 570
 571     def test_unicode(self):
 572         # non-ascii templates
 573         for oldvar in self._conf_mapping:
 574             self._pckdict[str(oldvar)] = b'Ol\xe1!'
 575         import_config_pck(self._mlist, self._pckdict)
 576         for oldvar, newvar in self._conf_mapping.items():
 577             newattr = getattr(self._mlist, newvar)
 578             text = decorate(self._mlist, newattr)
 579             expected = u'Ol\ufffd!'
 580             self.assertEqual(text, expected)
 581
 582     def test_unicode_in_default(self):
 583         # What if the default template is already in UTF-8?   For example, if
 584         # you import it twice.
 585         footer = b'\xe4\xb8\xad $listinfo_uri'
 586         footer_path = os.path.join(
 587             config.VAR_DIR, 'templates', 'lists',
 588             'blank@example.com', 'en', 'footer-generic.txt')
 589         makedirs(os.path.dirname(footer_path))
 590         with open(footer_path, 'wb') as fp:
 591             fp.write(footer)
 592         self._pckdict['msg_footer'] = b'NEW-VALUE'
 593         import_config_pck(self._mlist, self._pckdict)
 594         text = decorate(self._mlist, self._mlist.footer_uri)
 595         self.assertEqual(text, 'NEW-VALUE')
 596
 597 \f
 598 class TestRosterImport(unittest.TestCase):
 599     """Test that rosters are imported correctly."""
 600
 601     layer = ConfigLayer
 602
 603     def setUp(self):
 604         self._mlist = create_list('blank@example.com')
 605         self._pckdict = {
 606             'members': {
 607                 'anne@example.com': 0,
 608                 'bob@example.com': b'bob@ExampLe.Com',
 609             },
 610             'digest_members': {
 611                 'cindy@example.com': 0,
 612                 'dave@example.com': b'dave@ExampLe.Com',
 613             },
 614             'passwords': {
 615                 'anne@example.com' : b'annepass',
 616                 'bob@example.com'  : b'bobpass',
 617                 'cindy@example.com': b'cindypass',
 618                 'dave@example.com' : b'davepass',
 619             },
 620             'language': {
 621                 'anne@example.com' : b'fr',
 622                 'bob@example.com'  : b'de',
 623                 'cindy@example.com': b'es',
 624                 'dave@example.com' : b'it',
 625             },
 626             'usernames': { # Usernames are unicode strings in the pickle
 627                 'anne@example.com' : 'Anne',
 628                 'bob@example.com'  : 'Bob',
 629                 'cindy@example.com': 'Cindy',
 630                 'dave@example.com' : 'Dave',
 631             },
 632             'owner': [
 633                 'anne@example.com',
 634                 'emily@example.com',
 635             ],
 636             'moderator': [
 637                 'bob@example.com',
 638                 'fred@example.com',
 639             ],
 640         }
 641         self._usermanager = getUtility(IUserManager)
 642         language_manager = getUtility(ILanguageManager)
 643         for code in self._pckdict['language'].values():
 644             if isinstance(code, bytes):
 645                 code = code.decode('utf-8')
 646             if code not in language_manager.codes:
 647                 language_manager.add(code, 'utf-8', code)
 648
 649     def test_member(self):
 650         import_config_pck(self._mlist, self._pckdict)
 651         for name in ('anne', 'bob', 'cindy', 'dave'):
 652             addr = '%s@example.com' % name
 653             self.assertIn(addr,
 654                           [a.email for a in self._mlist.members.addresses],
 655                           'Address %s was not imported' % addr)
 656         self.assertIn('anne@example.com',
 657                       [a.email for a in self._mlist.regular_members.addresses])
 658         self.assertIn('bob@example.com',
 659                       [a.email for a in self._mlist.regular_members.addresses])
 660         self.assertIn('cindy@example.com',
 661                       [a.email for a in self._mlist.digest_members.addresses])
 662         self.assertIn('dave@example.com',
 663                       [a.email for a in self._mlist.digest_members.addresses])
 664
 665     def test_original_email(self):
 666         import_config_pck(self._mlist, self._pckdict)
 667         bob = self._usermanager.get_address('bob@example.com')
 668         self.assertEqual(bob.original_email, 'bob@ExampLe.Com')
 669         dave = self._usermanager.get_address('dave@example.com')
 670         self.assertEqual(dave.original_email, 'dave@ExampLe.Com')
 671
 672     def test_language(self):
 673         import_config_pck(self._mlist, self._pckdict)
 674         for name in ('anne', 'bob', 'cindy', 'dave'):
 675             addr = '%s@example.com' % name
 676             member = self._mlist.members.get_member(addr)
 677             self.assertIsNotNone(member, 'Address %s was not imported' % addr)
 678             code = self._pckdict['language'][addr]
 679             if isinstance(code, bytes):
 680                 code = code.decode('utf-8')
 681             self.assertEqual(member.preferred_language.code, code)
 682
 683     def test_new_language(self):
 684         self._pckdict['language']['anne@example.com'] = b'xx_XX'
 685         try:
 686             import_config_pck(self._mlist, self._pckdict)
 687         except Import21Error as error:
 688             self.assertIn('[language.xx_XX]', str(error))
 689         else:                                       # pragma: no cover
 690             self.fail('Import21Error was not raised')
 691
 692     def test_username(self):
 693         import_config_pck(self._mlist, self._pckdict)
 694         for name in ('anne', 'bob', 'cindy', 'dave'):
 695             addr = '%s@example.com' % name
 696             user = self._usermanager.get_user(addr)
 697             address = self._usermanager.get_address(addr)
 698             self.assertIsNotNone(user, 'User %s was not imported' % addr)
 699             self.assertIsNotNone(address, 'Address %s was not imported' % addr)
 700             display_name = self._pckdict['usernames'][addr]
 701             self.assertEqual(user.display_name, display_name,
 702                 'The display name was not set for User %s' % addr)
 703             self.assertEqual(address.display_name, display_name,
 704                 'The display name was not set for Address %s' % addr)
 705
 706     def test_owner(self):
 707         import_config_pck(self._mlist, self._pckdict)
 708         for name in ('anne', 'emily'):
 709             addr = '%s@example.com' % name
 710             self.assertIn(addr,
 711                           [a.email for a in self._mlist.owners.addresses],
 712                           'Address %s was not imported as owner' % addr)
 713         self.assertNotIn(
 714             'emily@example.com',
 715             [a.email for a in self._mlist.members.addresses],
 716             'Address emily@ was wrongly added to the members list')
 717
 718     def test_moderator(self):
 719         import_config_pck(self._mlist, self._pckdict)
 720         for name in ('bob', 'fred'):
 721             addr = '%s@example.com' % name
 722             self.assertIn(addr,
 723                           [a.email for a in self._mlist.moderators.addresses],
 724                           'Address %s was not imported as moderator' % addr)
 725         self.assertNotIn('fred@example.com',
 726                          [a.email for a in self._mlist.members.addresses],
 727                          'Address fred@ was wrongly added to the members list')
 728
 729     def test_password(self):
 730         #self.anne.password = config.password_context.encrypt('abc123')
 731         import_config_pck(self._mlist, self._pckdict)
 732         for name in ('anne', 'bob', 'cindy', 'dave'):
 733             addr = '%s@example.com' % name
 734             user = self._usermanager.get_user(addr)
 735             self.assertIsNotNone(user, 'Address %s was not imported' % addr)
 736             self.assertEqual(
 737                 user.password, '{plaintext}%spass' % name,
 738                 'Password for %s was not imported' % addr)
 739
 740     def test_same_user(self):
 741         # Adding the address of an existing User must not create another user.
 742         user = self._usermanager.create_user('anne@example.com', 'Anne')
 743         user.register('bob@example.com') # secondary email
 744         import_config_pck(self._mlist, self._pckdict)
 745         member = self._mlist.members.get_member('bob@example.com')
 746         self.assertEqual(member.user, user)
 747
 748     def test_owner_and_moderator_not_lowercase(self):
 749         # In the v2.1 pickled dict, the owner and moderator lists are not
 750         # necessarily lowercased already.
 751         self._pckdict['owner'] = [b'Anne@example.com']
 752         self._pckdict['moderator'] = [b'Anne@example.com']
 753         import_config_pck(self._mlist, self._pckdict)
 754         self.assertIn('anne@example.com',
 755                       [a.email for a in self._mlist.owners.addresses])
 756         self.assertIn('anne@example.com',
 757                       [a.email for a in self._mlist.moderators.addresses])
 758
 759     def test_address_already_exists_but_no_user(self):
 760         # An address already exists, but it is not linked to a user nor
 761         # subscribed.
 762         anne_addr = self._usermanager.create_address(
 763             'anne@example.com', 'Anne')
 764         import_config_pck(self._mlist, self._pckdict)
 765         anne = self._usermanager.get_user('anne@example.com')
 766         self.assertTrue(anne.controls('anne@example.com'))
 767         self.assertIn(anne_addr, self._mlist.regular_members.addresses)
 768
 769     def test_address_already_subscribed_but_no_user(self):
 770         # An address is already subscribed, but it is not linked to a user.
 771         anne_addr = self._usermanager.create_address(
 772             'anne@example.com', 'Anne')
 773         self._mlist.subscribe(anne_addr)
 774         # Suppress warning messages in test output.
 775         with mock.patch('sys.stderr'):
 776             import_config_pck(self._mlist, self._pckdict)
 777         anne = self._usermanager.get_user('anne@example.com')
 778         self.assertTrue(anne.controls('anne@example.com'))
 779
 780     def test_invalid_original_email(self):
 781         # When the member has an original email address (i.e. the
 782         # case-preserved version) that is invalid, their new address record's
 783         # original_email attribute will only be the case insensitive version.
 784         self._pckdict['members']['anne@example.com'] = b'invalid email address'
 785         try:
 786             import_config_pck(self._mlist, self._pckdict)
 787         except InvalidEmailAddressError as error:
 788             self.fail(error)
 789         self.assertIn('anne@example.com',
 790                       [a.email for a in self._mlist.members.addresses])
 791         anne = self._usermanager.get_address('anne@example.com')
 792         self.assertEqual(anne.original_email, 'anne@example.com')
 793
 794     def test_invalid_email(self):
 795         # When a member's email address is invalid, that member is skipped
 796         # during the import.
 797         self._pckdict['members'] = {
 798             'anne@example.com': 0,
 799             'invalid email address': b'invalid email address'
 800             }
 801         self._pckdict['digest_members'] = {}
 802         try:
 803             import_config_pck(self._mlist, self._pckdict)
 804         except InvalidEmailAddressError as error:
 805             self.fail(error)
 806         self.assertEqual(['anne@example.com'],
 807                          [a.email for a in self._mlist.members.addresses])
 808
 809     def test_no_email_sent(self):
 810         # No welcome message is sent to newly imported members.
 811         self.assertTrue(self._mlist.send_welcome_message)
 812         import_config_pck(self._mlist, self._pckdict)
 813         self.assertIn('anne@example.com',
 814                       [a.email for a in self._mlist.members.addresses])
 815         # There are no messages in any of the queues.
 816         for queue, switchboard in config.switchboards.items():
 817             file_count = len(switchboard.files)
 818             self.assertEqual(file_count, 0,
 819                              "Unexpected queue '{}' file count: {}".format(
 820                                  queue, file_count))
 821         self.assertTrue(self._mlist.send_welcome_message)
 822
 823
 824 \f
 825 class TestPreferencesImport(unittest.TestCase):
 826     """Preferences get imported too."""
 827
 828     layer = ConfigLayer
 829
 830     def setUp(self):
 831         self._mlist = create_list('blank@example.com')
 832         self._pckdict = dict(
 833             members={'anne@example.com': 0},
 834             user_options=dict(),
 835             delivery_status=dict(),
 836             )
 837         self._usermanager = getUtility(IUserManager)
 838
 839     def _do_test(self, oldvalue, expected):
 840         self._pckdict['user_options']['anne@example.com'] = oldvalue
 841         import_config_pck(self._mlist, self._pckdict)
 842         user = self._usermanager.get_user('anne@example.com')
 843         self.assertIsNotNone(user, 'User was not imported')
 844         member = self._mlist.members.get_member('anne@example.com')
 845         self.assertIsNotNone(member, 'Address was not subscribed')
 846         for exp_name, exp_val in expected.items():
 847             try:
 848                 currentval = getattr(member, exp_name)
 849             except AttributeError:
 850                 # hide_address has no direct getter
 851                 currentval = getattr(member.preferences, exp_name)
 852             self.assertEqual(
 853                 currentval, exp_val,
 854                 'Preference %s was not imported' % exp_name)
 855         # XXX: should I check that other params are still equal to
 856         # mailman.core.constants.system_preferences?
 857
 858     def test_acknowledge_posts(self):
 859         # AcknowledgePosts
 860         self._do_test(4, dict(acknowledge_posts=True))
 861
 862     def test_hide_address(self):
 863         # ConcealSubscription
 864         self._do_test(16, dict(hide_address=True))
 865
 866     def test_receive_own_postings(self):
 867         # DontReceiveOwnPosts
 868         self._do_test(2, dict(receive_own_postings=False))
 869
 870     def test_receive_list_copy(self):
 871         # DontReceiveDuplicates
 872         self._do_test(256, dict(receive_list_copy=False))
 873
 874     def test_digest_plain(self):
 875         # Digests & DisableMime
 876         self._pckdict['digest_members'] = self._pckdict['members'].copy()
 877         self._pckdict['members'] = dict()
 878         self._do_test(8, dict(delivery_mode=DeliveryMode.plaintext_digests))
 879
 880     def test_digest_mime(self):
 881         # Digests & not DisableMime
 882         self._pckdict['digest_members'] = self._pckdict['members'].copy()
 883         self._pckdict['members'] = dict()
 884         self._do_test(0, dict(delivery_mode=DeliveryMode.mime_digests))
 885
 886     def test_delivery_status(self):
 887         # Look for the pckdict['delivery_status'] key which will look like
 888         # (status, time) where status is among the following:
 889         # ENABLED  = 0 # enabled
 890         # UNKNOWN  = 1 # legacy disabled
 891         # BYUSER   = 2 # disabled by user choice
 892         # BYADMIN  = 3 # disabled by admin choice
 893         # BYBOUNCE = 4 # disabled by bounces
 894         for oldval, expected in enumerate((
 895                 DeliveryStatus.enabled,
 896                 DeliveryStatus.unknown, DeliveryStatus.by_user,
 897                 DeliveryStatus.by_moderator, DeliveryStatus.by_bounces)):
 898             self._pckdict['delivery_status']['anne@example.com'] = (oldval, 0)
 899             import_config_pck(self._mlist, self._pckdict)
 900             member = self._mlist.members.get_member('anne@example.com')
 901             self.assertIsNotNone(member, 'Address was not subscribed')
 902             self.assertEqual(member.delivery_status, expected)
 903             member.unsubscribe()
 904
 905     def test_moderate(self):
 906         # Option flag Moderate is translated to
 907         # member.moderation_action = Action.hold
 908         self._do_test(128, dict(moderation_action=Action.hold))
 909
 910     def test_multiple_options(self):
 911         # DontReceiveDuplicates & DisableMime & SuppressPasswordReminder
 912         # Keys might be Python 2 str/bytes or unicode.
 913         members = self._pckdict['members']
 914         self._pckdict['digest_members'] = members.copy()
 915         self._pckdict['members'] = dict()
 916         self._do_test(296, dict(
 917                 receive_list_copy=False,
 918                 delivery_mode=DeliveryMode.plaintext_digests,
 919                 ))