Commit 0fcb68ac authored by Aurélien Bompard's avatar Aurélien Bompard Committed by Barry Warsaw

Test schema migration for the header matches

parent 87f2f50b
......@@ -10,9 +10,11 @@ Create Date: 2015-09-11 10:11:38.310315
revision = '42756496720'
down_revision = '2bb9b382198'
from alembic import op
import sqlalchemy as sa
from alembic import op
from mailman.database.helpers import is_sqlite, exists_in_db
def upgrade():
# Create the new table
......@@ -46,12 +48,12 @@ def upgrade():
# Now that data is migrated, drop the old column (except on SQLite which
# does not support this)
if op.get_bind() != 'sqlite':
if not is_sqlite(connection):
op.drop_column('mailinglist', 'header_matches')
def downgrade():
if op.get_bind() != 'sqlite':
if not exists_in_db(op.get_bind(), 'mailinglist', 'header_matches'):
# SQLite will not have deleted the former column, since it does not
# support column deletion.
op.add_column('mailinglist', sa.Column(
......@@ -72,11 +74,15 @@ def downgrade():
sa.sql.column('pattern', sa.Unicode),
for mlist_id, header, pattern in connection.execute(
mlist = connection.execute( == mlist_id)).fetchone()
if not mlist["header_matches"]:
mlist["header_matches"] = []
mlist["header_matches"].append((header, pattern))
header_matches = mlist['header_matches']
if not header_matches:
header_matches = []
header_matches.append((header, pattern))
connection.execute(mlist_table.update().where( == mlist_id).values(
......@@ -28,6 +28,7 @@ import sqlalchemy as sa
from mailman.config import config
from mailman.database.alembic import alembic_cfg
from mailman.database.helpers import exists_in_db
from mailman.database.model import Model
from mailman.testing.layers import ConfigLayer
......@@ -41,8 +42,17 @@ class TestMigrations(unittest.TestCase):
def tearDown(self):
# Drop and restore a virgin database.
md = sa.MetaData(bind=config.db.engine)
# We have circular dependencies between user and address, thus we can't
# use drop_all() without getting a warning. Setting use_alter to True
# on the foreign keys helps SQLAlchemy mark those loops as known.
for tablename in ("user", "address"):
if tablename not in md.tables:
for fk in md.tables[tablename].foreign_key_constraints:
fk.use_alter = True
......@@ -55,3 +65,37 @@ class TestMigrations(unittest.TestCase):
for revision in revisions:
alembic.command.upgrade(alembic_cfg, revision)
def test_42756496720_header_matches(self):
test_header_matches = [
('test-header-1', 'test-pattern-1'),
('test-header-2', 'test-pattern-2'),
('test-header-3', 'test-pattern-3'),
mlist_table = sa.sql.table('mailinglist',
sa.sql.column('id', sa.Integer),
sa.sql.column('header_matches', sa.PickleType)
header_match_table = sa.sql.table('headermatch',
sa.sql.column('mailing_list_id', sa.Integer),
sa.sql.column('header', sa.Unicode),
sa.sql.column('pattern', sa.Unicode),
# Downgrading
[{'mailing_list_id': 1, 'header': hm[0], 'pattern': hm[1]}
for hm in test_header_matches]))
alembic.command.downgrade(alembic_cfg, '2bb9b382198')
results =
self.assertEqual(results[0].header_matches, test_header_matches)
self.assertFalse(exists_in_db(config.db.engine, 'headermatch'))
# Upgrading
alembic.command.upgrade(alembic_cfg, '42756496720')
results =
[(1, hm[0], hm[1]) for hm in test_header_matches])
Markdown is supported
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment