如何针对Django数据迁移运行测试？

Question

使用文档中的以下示例：

def combine_names(apps, schema_editor):
    Person = apps.get_model("yourappname", "Person")
    for person in Person.objects.all():
        person.name = "%s %s" % (person.first_name, person.last_name)
        person.save()

class Migration(migrations.Migration):    
    dependencies = [
        ('yourappname', '0001_initial'),
    ]    
    operations = [
        migrations.RunPython(combine_names),
    ]

如何创建并运行针对此迁移的测试，以确认数据已正确迁移？

Answer 1

编辑：

这些其他答案更有意义：

• https://stackoverflow.com/a/56212859
• https://stackoverflow.com/a/59016744，如果您不介意额外的（dev）依赖关系的话

原创：

在实际应用之前，通过一些基本单元测试来运行数据迁移功能（例如OP示例中的组合_name），对我来说也是有意义的。

乍一看，这应该不会比普通的Django单元测试困难很多：迁移是Python模块，migrations/文件夹是一个包，因此可以从中导入内容。然而，这需要一些时间才能实现。

第一个困难是因为默认迁移文件名以数字开头。例如，假设OP（即Django）数据迁移示例中的代码位于0002_my_data_迁移中。py，则很容易使用

from yourappname.migrations.0002_my_data_migration import combine_names

但这会引发一个SyntaxError，因为模块名称以数字（0）开头。

至少有两种方法可以实现这一点：

>

如果您想坚持使用默认编号的迁移文件名，您可以使用Python的import_module（请参阅文档和这个SO问题）。

第二个困难来自这样一个事实：数据迁移函数被设计为传递到RunPython（docs）中，因此默认情况下它们需要两个输入参数：apps和schema\u editor。要查看这些数据的来源，可以检查源代码。

现在，我不确定这是否适用于每个案例（请，任何人，如果你能澄清，请发表评论），但是对于我们的案例，从django.apps导入应用程序并从活动数据库获取schema_editor就足够了连接（django.db.connection）。

下面是一个简化的示例，展示了如何为OP示例实现此功能，假设迁移文件名为0002\u my\u data\u migration。py：

from importlib import import_module
from django.test import TestCase
from django.apps import apps
from django.db import connection
from yourappname.models import Person
# Our filename starts with a number, so we use import_module
data_migration = import_module('yourappname.migrations.0002_my_data_migration')


class DataMigrationTests(TestCase):
    def __init__(self, *args, **kwargs):
        super(DataMigrationTests, self).__init__(*args, **kwargs)
        # Some test values
        self.first_name = 'John'
        self.last_name = 'Doe'
        
    def test_combine_names(self):
        # Create a dummy Person
        Person.objects.create(first_name=self.first_name,
                              last_name=self.last_name, 
                              name=None)
        # Run the data migration function
        data_migration.combine_names(apps, connection.schema_editor())
        # Test the result
        person = Person.objects.get(id=1)
        self.assertEqual('{} {}'.format(self.first_name, self.last_name), person.name)
        

Answer 2

可以使用django-test-迁移包。它适用于测试：数据迁移、模式迁移和迁移顺序。

下面是它的工作原理：

from django_test_migrations.migrator import Migrator

# You can specify any database alias you need:
migrator = Migrator(database='default')

old_state = migrator.before(('main_app', '0002_someitem_is_clean'))
SomeItem = old_state.apps.get_model('main_app', 'SomeItem')

# One instance will be `clean`, the other won't be:
SomeItem.objects.create(string_field='a')
SomeItem.objects.create(string_field='a b')

assert SomeItem.objects.count() == 2
assert SomeItem.objects.filter(is_clean=True).count() == 2

new_state = migrator.after(('main_app', '0003_auto_20191119_2125'))
SomeItem = new_state.apps.get_model('main_app', 'SomeItem')

assert SomeItem.objects.count() == 2
# One instance is clean, the other is not:
assert SomeItem.objects.filter(is_clean=True).count() == 1
assert SomeItem.objects.filter(is_clean=False).count() == 1

我们还为pytest提供了本机集成：

@pytest.mark.django_db
def test_main_migration0002(migrator):
    """Ensures that the second migration works."""
    old_state = migrator.before(('main_app', '0002_someitem_is_clean'))
    SomeItem = old_state.apps.get_model('main_app', 'SomeItem')
    ...

和unittest：

from django_test_migrations.contrib.unittest_case import MigratorTestCase

class TestDirectMigration(MigratorTestCase):
    """This class is used to test direct migrations."""

    migrate_from = ('main_app', '0002_someitem_is_clean')
    migrate_to = ('main_app', '0003_auto_20191119_2125')

    def prepare(self):
        """Prepare some data before the migration."""
        SomeItem = self.old_state.apps.get_model('main_app', 'SomeItem')
        SomeItem.objects.create(string_field='a')
        SomeItem.objects.create(string_field='a b')

    def test_migration_main0003(self):
        """Run the test itself."""
        SomeItem = self.new_state.apps.get_model('main_app', 'SomeItem')

        assert SomeItem.objects.count() == 2
        assert SomeItem.objects.filter(is_clean=True).count() == 1

• 完整指南：https://sobolevn.me/2019/10/testing-django-migrations
• Github：https://github.com/wemake-services/django-test-migrations
• PyPI：https://pypi.org/project/django-test-migrations/

Answer 3

我做了一些google来回答同样的问题，发现一篇文章对我来说一锤定音，看起来不像现有的答案那么俗气。所以，把这个放在这里，以防它能帮助其他人。

Django的TestCase的以下子类：

from django.apps import apps
from django.test import TestCase
from django.db.migrations.executor import MigrationExecutor
from django.db import connection


class TestMigrations(TestCase):

    @property
    def app(self):
        return apps.get_containing_app_config(type(self).__module__).name

    migrate_from = None
    migrate_to = None

    def setUp(self):
        assert self.migrate_from and self.migrate_to, \
            "TestCase '{}' must define migrate_from and migrate_to     properties".format(type(self).__name__)
        self.migrate_from = [(self.app, self.migrate_from)]
        self.migrate_to = [(self.app, self.migrate_to)]
        executor = MigrationExecutor(connection)
        old_apps = executor.loader.project_state(self.migrate_from).apps

        # Reverse to the original migration
        executor.migrate(self.migrate_from)

        self.setUpBeforeMigration(old_apps)

        # Run the migration to test
        executor = MigrationExecutor(connection)
        executor.loader.build_graph()  # reload.
        executor.migrate(self.migrate_to)

        self.apps = executor.loader.project_state(self.migrate_to).apps

    def setUpBeforeMigration(self, apps):
        pass

他们提出的一个示例用例是：

class TagsTestCase(TestMigrations):

    migrate_from = '0009_previous_migration'
    migrate_to = '0010_migration_being_tested'

    def setUpBeforeMigration(self, apps):
        BlogPost = apps.get_model('blog', 'Post')
        self.post_id = BlogPost.objects.create(
            title = "A test post with tags",
            body = "",
            tags = "tag1 tag2",
        ).id

    def test_tags_migrated(self):
        BlogPost = self.apps.get_model('blog', 'Post')
        post = BlogPost.objects.get(id=self.post_id)

        self.assertEqual(post.tags.count(), 2)
        self.assertEqual(post.tags.all()[0].name, "tag1")
        self.assertEqual(post.tags.all()[1].name, "tag2")