Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use test_header_matches method in tempest

Best Python code snippet using tempest_python
test_migrations.py
Source:test_migrations.py
1# Copyright (C) 2015-2016 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 under6# the terms of the GNU General Public License as published by the Free7# 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 WITHOUT11# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or12# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for13# more details.14#15# You should have received a copy of the GNU General Public License along with16# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.17"""Test database schema migrations with Alembic"""18import os19import unittest20import sqlalchemy as sa21import alembic.command22from mailman.app.lifecycle import create_list23from mailman.config import config24from mailman.database.alembic import alembic_cfg25from mailman.database.helpers import exists_in_db26from mailman.database.model import Model27from mailman.database.transaction import transaction28from mailman.database.types import Enum29from mailman.interfaces.action import Action30from mailman.interfaces.member import MemberRole31from mailman.interfaces.usermanager import IUserManager32from mailman.testing.layers import ConfigLayer33from zope.component import getUtility34class TestMigrations(unittest.TestCase):35    layer = ConfigLayer36    def setUp(self):37        alembic.command.stamp(alembic_cfg, 'head')38    def tearDown(self):39        # Drop and restore a virgin database.40        config.db.store.rollback()41        md = sa.MetaData(bind=config.db.engine)42        md.reflect()43        # We have circular dependencies between user and address, thus we can't44        # use drop_all() without getting a warning.  Setting use_alter to True45        # on the foreign keys helps SQLAlchemy mark those loops as known.46        for tablename in ('user', 'address'):47            if tablename not in md.tables:48                continue49            for fk in md.tables[tablename].foreign_keys:50                fk.constraint.use_alter = True51        md.drop_all()52        Model.metadata.create_all(config.db.engine)53    def test_all_migrations(self):54        script_dir = alembic.script.ScriptDirectory.from_config(alembic_cfg)55        revisions = [sc.revision for sc in script_dir.walk_revisions()]56        for revision in revisions:57            alembic.command.downgrade(alembic_cfg, revision)58        revisions.reverse()59        for revision in revisions:60            alembic.command.upgrade(alembic_cfg, revision)61    def test_42756496720_header_matches(self):62        test_header_matches = [63            ('test-header-1', 'test-pattern-1'),64            ('test-header-2', 'test-pattern-2'),65            ('test-header-3', 'test-pattern-3'),66            ]67        mlist_table = sa.sql.table(68            'mailinglist',69            sa.sql.column('id', sa.Integer),70            sa.sql.column('header_matches', sa.PickleType)71            )72        header_match_table = sa.sql.table(73            'headermatch',74            sa.sql.column('mailing_list_id', sa.Integer),75            sa.sql.column('header', sa.Unicode),76            sa.sql.column('pattern', sa.Unicode),77            )78        # Bring the DB to the revision that is being tested.79        alembic.command.downgrade(alembic_cfg, '42756496720')80        # Test downgrading.81        config.db.store.execute(mlist_table.insert().values(id=1))82        config.db.store.execute(header_match_table.insert().values(83            [{'mailing_list_id': 1, 'header': hm[0], 'pattern': hm[1]}84             for hm in test_header_matches]))85        config.db.store.commit()86        alembic.command.downgrade(alembic_cfg, '2bb9b382198')87        results = config.db.store.execute(88            mlist_table.select()).fetchall()89        self.assertEqual(results[0].header_matches, test_header_matches)90        self.assertFalse(exists_in_db(config.db.engine, 'headermatch'))91        config.db.store.commit()92        # Test upgrading.93        alembic.command.upgrade(alembic_cfg, '42756496720')94        results = config.db.store.execute(95            header_match_table.select()).fetchall()96        self.assertEqual(97            results,98            [(1, hm[0], hm[1]) for hm in test_header_matches])99    def test_47294d3a604_pendable_keyvalues(self):100        # We have 5 pended items:101        # - one is a probe request102        # - one is a subscription request103        # - one is a moderation request104        # - one is a held message105        # - one is a registration request in the new format106        #107        # The first three used to have no 'type' key and must be properly108        # typed, the held message used to have a type key, but in JSON, and109        # must be converted.110        pended_table = sa.sql.table(111            'pended',112            sa.sql.column('id', sa.Integer),113            )114        keyvalue_table = sa.sql.table(115            'pendedkeyvalue',116            sa.sql.column('id', sa.Integer),117            sa.sql.column('key', sa.Unicode),118            sa.sql.column('value', sa.Unicode),119            sa.sql.column('pended_id', sa.Integer),120            )121        def get_from_db():                                 # noqa122            results = {}123            for i in range(1, 6):124                query = sa.sql.select(125                    [keyvalue_table.c.key, keyvalue_table.c.value]126                ).where(127                    keyvalue_table.c.pended_id == i128                )129                results[i] = dict([130                    (r['key'], r['value']) for r in131                    config.db.store.execute(query).fetchall()132                    ])133            return results134        # Start at the previous revision135        with transaction():136            alembic.command.downgrade(alembic_cfg, '33bc0099223')137            for i in range(1, 6):138                config.db.store.execute(pended_table.insert().values(id=i))139            config.db.store.execute(keyvalue_table.insert().values([140                {'pended_id': 1, 'key': 'member_id', 'value': 'test-value'},141                {'pended_id': 2, 'key': 'token_owner', 'value': 'test-value'},142                {'pended_id': 3, 'key': '_mod_message_id',143                                 'value': 'test-value'},144                {'pended_id': 4, 'key': 'type', 'value': '"held message"'},145                {'pended_id': 5, 'key': 'type', 'value': 'registration'},146                ]))147        # Upgrading.148        with transaction():149            alembic.command.upgrade(alembic_cfg, '47294d3a604')150            results = get_from_db()151        for i in range(1, 5):152            self.assertIn('type', results[i])153        self.assertEqual(results[1]['type'], 'probe')154        self.assertEqual(results[2]['type'], 'subscription')155        self.assertEqual(results[3]['type'], 'data')156        self.assertEqual(results[4]['type'], 'held message')157        self.assertEqual(results[5]['type'], 'registration')158        # Downgrading.159        with transaction():160            alembic.command.downgrade(alembic_cfg, '33bc0099223')161            results = get_from_db()162        for i in range(1, 4):163            self.assertNotIn('type', results[i])164        self.assertEqual(results[4]['type'], '"held message"')165        self.assertEqual(results[5]['type'], '"registration"')166    def test_70af5a4e5790_digests(self):167        IDS_TO_DIGESTABLE = [168            (1, True),169            (2, False),170            (3, False),171            (4, True),172            ]173        mlist_table = sa.sql.table(174            'mailinglist',175            sa.sql.column('id', sa.Integer),176            sa.sql.column('digests_enabled', sa.Boolean)177            )178        # Downgrading.179        with transaction():180            for table_id, enabled in IDS_TO_DIGESTABLE:181                config.db.store.execute(mlist_table.insert().values(182                    id=table_id, digests_enabled=enabled))183        with transaction():184            alembic.command.downgrade(alembic_cfg, '47294d3a604')185            results = config.db.store.execute(186                'SELECT id, digestable FROM mailinglist').fetchall()187        self.assertEqual(results, IDS_TO_DIGESTABLE)188        # Upgrading.189        with transaction():190            alembic.command.upgrade(alembic_cfg, '70af5a4e5790')191        results = config.db.store.execute(192            'SELECT id, digests_enabled FROM mailinglist').fetchall()193        self.assertEqual(results, IDS_TO_DIGESTABLE)194    def test_70af5a4e5790_data_paths(self):195        # Create a couple of mailing lists through the standard API.196        with transaction():197            ant = create_list('ant@example.com')198            bee = create_list('bee@example.com')199        # Downgrade and verify that the old data paths exist.200        alembic.command.downgrade(alembic_cfg, '47294d3a604')201        self.assertTrue(os.path.exists(202            os.path.join(config.LIST_DATA_DIR, 'ant@example.com')))203        self.assertTrue(os.path.exists(204            os.path.join(config.LIST_DATA_DIR, 'ant@example.com')))205        # Upgrade and verify that the new data paths exists and the old ones206        # no longer do.207        alembic.command.upgrade(alembic_cfg, '70af5a4e5790')208        self.assertFalse(os.path.exists(209            os.path.join(config.LIST_DATA_DIR, 'ant@example.com')))210        self.assertFalse(os.path.exists(211            os.path.join(config.LIST_DATA_DIR, 'ant@example.com')))212        self.assertTrue(os.path.exists(ant.data_path))213        self.assertTrue(os.path.exists(bee.data_path))214    def test_7b254d88f122_moderation_action(self):215        mailinglist_table = sa.sql.table(           # noqa216            'mailinglist',217            sa.sql.column('id', sa.Integer),218            sa.sql.column('list_id', sa.Unicode),219            sa.sql.column('default_member_action', Enum(Action)),220            sa.sql.column('default_nonmember_action', Enum(Action)),221            )222        member_table = sa.sql.table(223            'member',224            sa.sql.column('id', sa.Integer),225            sa.sql.column('list_id', sa.Unicode),226            sa.sql.column('address_id', sa.Integer),227            sa.sql.column('role', Enum(MemberRole)),228            sa.sql.column('moderation_action', Enum(Action)),229            )230        user_manager = getUtility(IUserManager)231        with transaction():232            # Start at the previous revision.233            alembic.command.downgrade(alembic_cfg, 'd4fbb4fd34ca')234            # Create a mailing list through the standard API.235            ant = create_list('ant@example.com')236            # Create some members.237            anne = user_manager.create_address('anne@example.com')238            bart = user_manager.create_address('bart@example.com')239            cris = user_manager.create_address('cris@example.com')240            dana = user_manager.create_address('dana@example.com')241            # Flush the database to get the last auto-increment id.242            config.db.store.flush()243            # Assign some moderation actions to the members created above.244            config.db.store.execute(member_table.insert().values([245                {'address_id': anne.id, 'role': MemberRole.owner,246                 'list_id': ant.list_id, 'moderation_action': Action.accept},247                {'address_id': bart.id, 'role': MemberRole.moderator,248                 'list_id': ant.list_id, 'moderation_action': Action.accept},249                {'address_id': cris.id, 'role': MemberRole.member,250                 'list_id': ant.list_id, 'moderation_action': Action.defer},251                {'address_id': dana.id, 'role': MemberRole.nonmember,252                 'list_id': ant.list_id, 'moderation_action': Action.hold},253                ]))254        # Cris and Dana have actions which match the list default action for255        # members and nonmembers respectively.256        self.assertEqual(257            ant.members.get_member('cris@example.com').moderation_action,258            ant.default_member_action)259        self.assertEqual(260            ant.nonmembers.get_member('dana@example.com').moderation_action,261            ant.default_nonmember_action)262        # Upgrade and check the moderation_actions.   Cris's and Dana's263        # actions have been set to None to fall back to the list defaults.264        alembic.command.upgrade(alembic_cfg, '7b254d88f122')265        members = config.db.store.execute(sa.select([266            member_table.c.address_id, member_table.c.moderation_action,267            ])).fetchall()268        self.assertEqual(members, [269            (anne.id, Action.accept),270            (bart.id, Action.accept),271            (cris.id, None),272            (dana.id, None),273            ])274        # Downgrade and check that Cris's and Dana's actions have been set275        # explicitly.276        alembic.command.downgrade(alembic_cfg, 'd4fbb4fd34ca')277        members = config.db.store.execute(sa.select([278            member_table.c.address_id, member_table.c.moderation_action,279            ])).fetchall()280        self.assertEqual(members, [281            (anne.id, Action.accept),282            (bart.id, Action.accept),283            (cris.id, Action.defer),284            (dana.id, Action.hold),...
Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.