1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
|
# Copyright (C) 2012 by the Free Software Foundation, Inc.
#
# This file is part of GNU Mailman.
#
# GNU Mailman is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free
# Software Foundation, either version 3 of the License, or (at your option)
# any later version.
#
# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# GNU Mailman. If not, see <http://www.gnu.org/licenses/>.
"""Test schema migrations."""
from __future__ import absolute_import, print_function, unicode_literals
__metaclass__ = type
__all__ = [
'TestMigration20120407ArchiveData',
'TestMigration20120407Data',
'TestMigration20120407Schema',
]
import unittest
from pkg_resources import resource_string
from storm.exceptions import DatabaseError
from zope.component import getUtility
from mailman.config import config
from mailman.interfaces.domain import IDomainManager
from mailman.interfaces.archiver import ArchivePolicy
from mailman.interfaces.listmanager import IListManager
from mailman.interfaces.mailinglist import IAcceptableAliasSet
from mailman.testing.helpers import temporary_db
from mailman.testing.layers import ConfigLayer
from mailman.utilities.modules import call_name
class MigrationTestBase(unittest.TestCase):
"""Test the dated migration (LP: #971013)
Circa: 3.0b1 -> 3.0b2
table mailinglist:
* news_moderation -> newsgroup_moderation
* news_prefix_subject_too -> nntp_prefix_subject_too
* ADD archive_policy
* REMOVE archive
* REMOVE archive_private
* REMOVE archive_volume_frequency
* REMOVE nntp_host
"""
layer = ConfigLayer
def setUp(self):
database_class_name = config.database['class']
self._database_class = call_name(database_class_name)
self._testdb = self._database_class._make_testdb()
self._database = self._testdb.database
def tearDown(self):
self._testdb.cleanup()
class TestMigration20120407Schema(MigrationTestBase):
"""Test column migrations."""
def test_pre_upgrade_columns_base(self):
# Test that before the migration, the old table columns are present
# and the new database columns are not.
#
# Load all the migrations to just before the one we're testing.
self._database.load_migrations('20120406999999')
# Verify that the database has not yet been migrated.
for missing in ('archive_policy',
'nntp_prefix_subject_too'):
self.assertRaises(DatabaseError,
self._database.store.execute,
'select {0} from mailinglist;'.format(missing))
self._testdb.abort()
for present in ('archive',
'archive_private',
'archive_volume_frequency',
'news_moderation',
'news_prefix_subject_too',
'nntp_host'):
# This should not produce an exception. Is there some better test
# that we can perform?
self._database.store.execute(
'select {0} from mailinglist;'.format(present))
def test_post_upgrade_columns_migration(self):
# Test that after the migration, the old table columns are missing
# and the new database columns are present.
#
# Load all the migrations up to and including the one we're testing.
self._database.load_migrations('20120406999999')
self._database.load_migrations('20120407000000')
# Verify that the database has been migrated.
for present in ('archive_policy',
'nntp_prefix_subject_too'):
# This should not produce an exception. Is there some better test
# that we can perform?
self._database.store.execute(
'select {0} from mailinglist;'.format(present))
for missing in ('archive',
'archive_private',
'archive_volume_frequency',
'news_moderation',
'news_prefix_subject_too',
'nntp_host'):
self.assertRaises(DatabaseError,
self._database.store.execute,
'select {0} from mailinglist;'.format(missing))
self._testdb.abort()
class TestMigration20120407Data(MigrationTestBase):
"""Test non-migrated data."""
def setUp(self):
MigrationTestBase.setUp(self)
# Load all the migrations to just before the one we're testing.
self._database.load_migrations('20120406999999')
# Load the previous schema's sample data.
sample_data = resource_string(
'mailman.database.tests.data',
'migration_{0}_1.sql'.format(self._database_class.TAG))
self._database.load_sql(self._database.store, sample_data)
# Update to the current migration we're testing.
self._database.load_migrations('20120407000000')
def test_migration_domains(self):
# Test that the domains table, which isn't touched, doesn't change.
with temporary_db(self._database):
# Check that the domains survived the migration. This table
# was not touched so it should be fine.
domains = list(getUtility(IDomainManager))
self.assertEqual(len(domains), 1)
self.assertEqual(domains[0].mail_host, 'example.com')
def test_migration_mailing_lists(self):
# Test that the mailing lists survive migration.
with temporary_db(self._database):
# There should be exactly one mailing list defined.
mlists = list(getUtility(IListManager).mailing_lists)
self.assertEqual(len(mlists), 1)
self.assertEqual(mlists[0].fqdn_listname, 'test@example.com')
def test_migration_acceptable_aliases(self):
# Test that the mailing list's acceptable aliases survive migration.
# This proves that foreign key references are migrated properly.
with temporary_db(self._database):
mlist = getUtility(IListManager).get('test@example.com')
aliases_set = IAcceptableAliasSet(mlist)
self.assertEqual(set(aliases_set.aliases),
set(['foo@example.com', 'bar@example.com']))
def test_migration_members(self):
# Test that the members of a mailing list all survive migration.
with temporary_db(self._database):
mlist = getUtility(IListManager).get('test@example.com')
# Test that all the members we expect are still there. Start with
# the two list delivery members.
addresses = set(address.email
for address in mlist.members.addresses)
self.assertEqual(addresses,
set(['anne@example.com', 'bart@example.com']))
# There is one owner.
owners = set(address.email for address in mlist.owners.addresses)
self.assertEqual(len(owners), 1)
self.assertEqual(owners.pop(), 'anne@example.com')
# There is one moderator.
moderators = set(address.email
for address in mlist.moderators.addresses)
self.assertEqual(len(moderators), 1)
self.assertEqual(moderators.pop(), 'bart@example.com')
class TestMigration20120407ArchiveData(MigrationTestBase):
"""Test affected migration data."""
def setUp(self):
MigrationTestBase.setUp(self)
# Load all the migrations to just before the one we're testing.
self._database.load_migrations('20120406999999')
# Load the previous schema's sample data.
sample_data = resource_string(
'mailman.database.tests.data',
'migration_{0}_1.sql'.format(self._database_class.TAG))
self._database.load_sql(self._database.store, sample_data)
def _upgrade(self):
# Update to the current migration we're testing.
self._database.load_migrations('20120407000000')
def test_migration_archive_policy_never_0(self):
# Test that the new archive_policy value is updated correctly. In the
# case of old column archive=0, the archive_private column is
# ignored. This test sets it to 0 to ensure it's ignored.
self._database.store.execute(
'UPDATE mailinglist SET archive = {0}, archive_private = {0} '
'WHERE id = 1;'.format(self._testdb.FALSE))
# Complete the migration
self._upgrade()
with temporary_db(self._database):
mlist = getUtility(IListManager).get('test@example.com')
self.assertEqual(mlist.archive_policy, ArchivePolicy.never)
def test_migration_archive_policy_never_1(self):
# Test that the new archive_policy value is updated correctly. In the
# case of old column archive=0, the archive_private column is
# ignored. This test sets it to 1 to ensure it's ignored.
self._database.store.execute(
'UPDATE mailinglist SET archive = {0}, archive_private = {1} '
'WHERE id = 1;'.format(self._testdb.FALSE,
self._testdb.TRUE))
# Complete the migration
self._upgrade()
with temporary_db(self._database):
mlist = getUtility(IListManager).get('test@example.com')
self.assertEqual(mlist.archive_policy, ArchivePolicy.never)
def test_archive_policy_private(self):
# Test that the new archive_policy value is updated correctly for
# private archives.
self._database.store.execute(
'UPDATE mailinglist SET archive = {0}, archive_private = {0} '
'WHERE id = 1;'.format(self._testdb.TRUE))
# Complete the migration
self._upgrade()
with temporary_db(self._database):
mlist = getUtility(IListManager).get('test@example.com')
self.assertEqual(mlist.archive_policy, ArchivePolicy.private)
def test_archive_policy_public(self):
# Test that the new archive_policy value is updated correctly for
# public archives.
self._database.store.execute(
'UPDATE mailinglist SET archive = {1}, archive_private = {0} '
'WHERE id = 1;'.format(self._testdb.FALSE,
self._testdb.TRUE))
# Complete the migration
self._upgrade()
with temporary_db(self._database):
mlist = getUtility(IListManager).get('test@example.com')
self.assertEqual(mlist.archive_policy, ArchivePolicy.public)
|
