MigrationTestCase

[Tom Linford]

At Robinhood, we've been using a custom in-house MigrationTestCase for testing migrations that we'd like to contribute, but want to check the API of it before contributing it. Here's the definition of the class:

class MigrationTestCase(TransactionTestCase):

    """

    app_label: name of app (ie. "users" or "polls")

    (start|dest)_migration_name: name of migration in app

        (e.g. "0001_initial")

    additional_dependencies: list of tuples of `(app_label, migration_name)`.

        Add any additional migrations here that need to be included in the

        generation of the model states.

    Usage:

    class TestCase(MigrationTestCase):

        app_label = ...

        start_migration_name = ...

        dest_migration_name = ...

        additional_dependencies = ...

        def setUp(self):

            # Create models with regular orm

            super(TestCase, self).setUp()

            # Create models with start orm. Access model with:

            # self.start_models["<app_label>"]["<model_name>"]

            # Note that model_name must be all lower case, you can just do:

            # <instance>._meta.model_name to get the model_name

        def test(self):

            # Still using start orm

            self.migrate_to_dest()

            # Now, you can access dest models with:

            # self.dest_models["<app_label>"]["<model_name>"]

    """

    app_label = None

    start_migration_name = None

    dest_migration_name = None

    additional_dependencies = []

Let me know if this API is agreeable and I can make a PR for this feature.

[Andrew Godwin]

Hi Tom,

When you say "testing migrations", what do you mean exactly? The migration framework itself is heavily unit-tested, so I presume you intend to test things like custom RunPython function bodies and the like?

Andrew

--
You received this message because you are subscribed to the Google Groups "Django developers (Contributions to Django itself)" group.
To unsubscribe from this group and stop receiving emails from it, send an email to django-develop...@googlegroups.com.
To post to this group, send email to django-d...@googlegroups.com.
Visit this group at http://groups.google.com/group/django-developers.
To view this discussion on the web visit https://groups.google.com/d/msgid/django-developers/8818bf72-aa66-4351-9177-e6e0f6605386%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Sean Briceland]

I believe Tom is referring to testing their migration files in order to ensure DB is migrated accordingly.

For example, at our company we test all of our source code & Migrations are code too! Most of the time we test rolling migrations forwards and backwards to ensure they will run without a hitch once deployed to production. 

For data migrations we use the MigrationLoader to generate models at a given state of the migration history. Then we can verify we our code within the migration mutates the data as desired.

While Django Migrations are tested, stable, & kicka$$, they do not prevent developers from generating erroneous data migrations or even irreversible schema migrations.

I jumped on this post because we now have a pretty beefy Django Application and as a result our Migration Tests take forever. Without getting into to much detail, I want to make sure that it would be okay to post here? or should I open a new thread?

[Tim Graham]

Sure... what do you think of the API that Tom proposed? Did you have something different in mind?

[Sean Briceland]

Let me run it by my CTO. But I should be able to send over our migration test class.

You received this message because you are subscribed to a topic in the Google Groups "Django developers (Contributions to Django itself)" group.
To unsubscribe from this topic, visit https://groups.google.com/d/topic/django-developers/181BkMhFUwo/unsubscribe.
To unsubscribe from this group and all its topics, send an email to django-develop...@googlegroups.com.

To post to this group, send email to django-d...@googlegroups.com.
Visit this group at http://groups.google.com/group/django-developers.

To view this discussion on the web visit https://groups.google.com/d/msgid/django-developers/e4aadb20-30ec-4968-8564-c6652dafeaf9%40googlegroups.com.

[Sean Briceland]

I like Tom's initial proposition. As mentioned ours is very similar. Here is currently what we are using:

class MigrationTestBase(TransactionTestCase):

    """

    Custom TestCase containing an extended set of asserts for testing

    migrations and schema operations.

    Most of this code was derived from Django's private MigrationTestBase:

    https://github.com/django/django/blob/stable/1.7.x/tests/migrations/test_base.py

    Notes:

        If you would like to override setUp() in the test, you will want to

        call super().setup() or explicitly invoke self.migrate_all_the_way() to

        ensure a clean history at the start of each test.

    """

    # MUST specify which apps we will need to test within the

    app_label = None

    test_migration_name = None

    # last_migration_nodes is a list of tuples assigned after each time we

    # migrate.

    # if last_migration_node is None, we will use apps from django.apps which

    # is the fully migrated App Registry created via our existing source code.

    last_migration_nodes = None

    @classmethod

    def setUpClass(cls):

        super(MigrationTestBase, cls).setUpClass()

        if not cls.app_label or not cls.test_migration_name:

            raise NotImplementedError('Must define class defaults: app_label, '

                                      'test_migration_name')

    def setUp(self):

        super(MigrationTestBase, self).setUp()

        self.migrate_all_the_way()

    def tearDown(self):

        self.migrate_all_the_way()

        super(MigrationTestBase, self).tearDown()

    def migrate_to(self, app_label, migration_name):

        call_command('migrate', app_label, migration_name, verbosity=0)

        self.last_migration_nodes = [(app_label, migration_name)]

    def migrate_to_test_migration(self):

        self.migrate_to(self.app_label, self.test_migration_name)

    def migrate_all_the_way(self):

        call_command('migrate', verbosity=0)

        self.last_migration_nodes = None

    def get_current_model(self, app_label, model):

        if self.last_migration_nodes is not None:

            conn = connections[DEFAULT_DB_ALIAS]

            loader = MigrationLoader(conn)

            proj_state = loader.project_state(self.last_migration_nodes)

            self.apps = proj_state.render()

            return self.apps.get_model(app_label, model)

        return dj_apps.get_model(app_label, model)

    def get_table_description(self, table):

        with connection.cursor() as cursor:

            return connection.introspection.get_table_description(

                cursor, table

            )

    def assertTableExists(self, table):

        with connection.cursor() as cursor:

            self.assertIn(

                table,

                connection.introspection.get_table_list(cursor)

            )

    def assertTableNotExists(self, table):

        with connection.cursor() as cursor:

            self.assertNotIn(

                table,

                connection.introspection.get_table_list(cursor)

            )

    def assertColumnExists(self, table, column):

        self.assertIn(

            column,

            [c.name for c in self.get_table_description(table)]

        )

    def assertColumnNotExists(self, table, column):

        self.assertNotIn(

            column,

            [c.name for c in self.get_table_description(table)]

        )

    def assertColumnNull(self, table, column):

        self.assertEqual(

            [

                c.null_ok

                for c in self.get_table_description(table) if c.name == column

            ][0],

            True

        )

    def assertColumnNotNull(self, table, column):

        self.assertEqual(

            [

                c.null_ok

                for c in self.get_table_description(table)

                if c.name == column

            ][0],

            False

        )

    def assertIndexExists(self, table, columns, value=True):

        with connection.cursor() as cursor:

            self.assertEqual(

                value,

                any(

                    c["index"]

                    for c in connection.introspection.get_constraints(

                        cursor, table).values()

                    if c['columns'] == list(columns)

                ),

            )

    def assertIndexNotExists(self, table, columns):

        return self.assertIndexExists(table, columns, False)

    def assertFKExists(self, table, columns, to, value=True):

        with connection.cursor() as cursor:

            self.assertEqual(

                value,

                any(

                    c["foreign_key"] == to

                    for c in connection.introspection.get_constraints(

                        cursor, table).values()

                    if c['columns'] == list(columns)

                ),

            )

    def assertFKNotExists(self, table, columns, to, value=True):

        return self.assertFKExists(table, columns, to, False)

[Daniel Hahler]

Interesting!

I came up with the following today, but calling the 'migrate' command via 'call_command' (like in Sean's code) is probably cleaner?!

"""
Test (data) migrations in Django.

This uses py.test/pytest-django (the `transactional_db` fixture comes from there),
but could be easily adopted for Django's testrunner:

    from django.test.testcases import TransactionTestCase

    class FooTestcase(TransactionTestCase):
        def test_with_django(self):
        …

This example tests that some fields are properly migrated from a `Profile` model
to `User`.
"""

from django.db import connection
from django.db.migrations.executor import MigrationExecutor


def test_migrate_profile_to_user(transactional_db):
    executor = MigrationExecutor(connection)
    app = "YOUR_APP"
    migrate_from = [(app, "000X_before")]
    migrate_to = [(app, "000X_after")]

    executor.migrate(migrate_from)
    old_apps = executor.loader.project_state(migrate_from).apps

    # Create some old data.
    Profile = old_apps.get_model(app, "Profile")
    old_profile = Profile.objects.create(email="email",
                                         firstname="firstname",
                                         lastname="lastname")
    # Migrate forwards.
    executor.loader.build_graph()  # reload.
    executor.migrate(migrate_to)
    new_apps = executor.loader.project_state(migrate_to).apps

    # Test the new data.
    Profile = new_apps.get_model(app, "Profile")
    User = new_apps.get_model(app, "UserEntry")
    assert 'firstname' not in Profile._meta.get_all_field_names()

    user = User.objects.get(email='email')
    profile = Profile.objects.get(user__email='email')
    assert user.profile.pk == old_profile.pk == profile.pk
    assert profile.user.email == 'email'
    assert profile.user.first_name == 'firstname'
    assert profile.user.last_name == 'lastname'
 

https://gist.github.com/blueyed/4fb0a807104551f103e6


Cheers,
Daniel.