Cannot test migrations if there is a collation in the migrations

[jrobichaud]

I have a migration with a collation:

Ex:

# Generated by Django 4.1.5 on 2023-02-04 09:26

from django.contrib.postgres.operations import CreateCollation
from django.db import migrations


class Migration(migrations.Migration):

    initial = True

    dependencies = []

    operations = [
        CreateCollation(
            "french",
            provider="icu",
            locale="fr-x-icu",
        )
    ]

When running the migration tests I have this error:

Error
Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/django/db/backends/utils.py", line 89, in _execute
    return self.cursor.execute(sql, params)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
psycopg2.errors.DuplicateObject: collation "french" already exists


The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/django_test_migrations/contrib/unittest_case.py", line 37, in setUp
    self.old_state = self._migrator.apply_initial_migration(
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/django_test_migrations/migrator.py", line 61, in apply_initial_migration
    return self._migrate(targets, plan=plan)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/django_test_migrations/migrator.py", line 84, in _migrate
    return self._executor.migrate(migration_targets, plan=plan)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/django/db/migrations/executor.py", line 135, in migrate
    state = self._migrate_all_forwards(
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/django/db/migrations/executor.py", line 167, in _migrate_all_forwards
    state = self.apply_migration(
            ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/django/db/migrations/executor.py", line 252, in apply_migration
    state = migration.apply(state, schema_editor)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/django/db/migrations/migration.py", line 130, in apply
    operation.database_forwards(
  File "/usr/local/lib/python3.11/site-packages/django/contrib/postgres/operations.py", line 223, in database_forwards
    self.create_collation(schema_editor)
  File "/usr/local/lib/python3.11/site-packages/django/contrib/postgres/operations.py", line 199, in create_collation
    schema_editor.execute(
  File "/usr/local/lib/python3.11/site-packages/django/db/backends/base/schema.py", line 199, in execute
    cursor.execute(sql, params)
  File "/usr/local/lib/python3.11/site-packages/django/db/backends/utils.py", line 67, in execute
    return self._execute_with_wrappers(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/django/db/backends/utils.py", line 80, in _execute_with_wrappers
    return executor(sql, params, many, context)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/django/db/backends/utils.py", line 84, in _execute
    with self.db.wrap_database_errors:
  File "/usr/local/lib/python3.11/site-packages/django/db/utils.py", line 91, in __exit__
    raise dj_exc_value.with_traceback(traceback) from exc_value
  File "/usr/local/lib/python3.11/site-packages/django/db/backends/utils.py", line 89, in _execute
    return self.cursor.execute(sql, params)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
django.db.utils.ProgrammingError: collation "french" already exists

[jrobichaud]

I found this workaround:

class TestMyMigration(MigratorTestCase):
    databases = ("default", "my_db",)
    database_name = "my_db"
    # ...
    def setUp(self) -> None:
        sql.drop_models_tables(self.database_name)
        sql.flush_django_migrations_table(self.database_name)
        with connections["my_db"].cursor() as cursor:
            cursor.execute("DROP COLLATION french")
        super().setUp()
        
    def test_my_migration(self):
        pass

[skarzi]

hi 👋

Creating a collation is tricky because it's not a data migration, it's just changing the DB configuration/metadata. Probably the best solution will be to add some hook (something like, migrator.apply_*_migration(..., run_before_apply=drop_collaction)), so the developers can define their own code that will be run before applying initial/target migration. What do you think about it @sobolevn and @jrobichaud?

[sobolevn]

Yes, seems like a good idea. Something like setup= and teardown=

[jrobichaud]

Sounds great!

[sobolevn]

@jrobichaud would you like to send a PR? :)

[jrobichaud]

I could give it a try.

I have trouble finding meaningful names.

For the TestCase it could work:

• setUpBeforeApply
• tearDownAfterApply

For the migrator:

• run_before_apply (its not clear its for forward migration)
• run_after_apply (its not clear its for reverse migration)

In theory someone could want to run 4 different things depending if its before/after apply in forward/reverse migration.

I only need before apply in forward migration.

Any thoughts about this?

[jrobichaud]

Is anything needed for the migrator at all considering the user call it himself directly? He could just add the just before the function.

I believe only the TestCase needs new functions.

[skarzi]

yes, that's true that when the developers are using the Migrator instance directly they can add custom code before/after the apply_*_migration(), but I think that for apply_initial_migration we will usually want to run "before apply" hook/code just after dropping tables and flushing Django migrations table - so exactly here, then adding arguments to apply_*_migration makes sense. We can add identical arguments to the apply_target_migration to make the interface more intuitive.

So I would add following arguments to apply_*_migration and pass it down to _migrate():

• before_hook: Callable[[PlanType], None](executed in_migrate` before doing migration)
• • after_hook: Callable[[PlanType], None](executed in_migrate` after doing migration)

and for the unittest test case just 1 method (optionally 2), taking plan as well (like you proposed):

• run_before_initial_migration

Optionally also run_after_initial_migration, but this can be replaced by adding the required code to prepare (of course assuming the migration plan is not important).

[medihack]

I am also having problems using Procrastinate (see mention above) because it creates some custom tables, functions and types in PostgreSQL. I wonder why only the Django models are dropped and not the whole database itself is reset (by dropping and recreating). It would be at least nice to have the option.

[sobolevn]

PR is welcome.

[medihack]

Sure, I can look into it (it will take some days to work on it). As you are deeper into it, what was the initial reason not to drop the whole database and then start from scratch but instead drop the model tables? On the other hand, I am not sure if dropping the database inside a test is even possible. I tried to call reset_db (a django-extensions command) inside a test but ended up with this error:

_________________________________________________________________________________________________________________________________ test_foobar __________________________________________________________________________________________________________________________________

    @pytest.mark.django_db
    def test_foobar():
>       call_command("reset_db", "--noinput")

adit_radis_shared/accounts/tests/test_migrations.py:44: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/opt/pysetup/.venv/lib/python3.12/site-packages/django/core/management/__init__.py:194: in call_command
    return command.execute(*args, **defaults)
/opt/pysetup/.venv/lib/python3.12/site-packages/django/core/management/base.py:459: in execute
    output = self.handle(*args, **options)
/opt/pysetup/.venv/lib/python3.12/site-packages/django_extensions/management/utils.py:62: in inner
    ret = func(self, *args, **kwargs)
/opt/pysetup/.venv/lib/python3.12/site-packages/django_extensions/management/commands/reset_db.py:183: in handle
    cursor.execute(drop_query)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <psycopg.Cursor [no result] [IDLE] (host=postgres.local user=postgres database=template1) at 0x77e05cad6f90>, query = 'DROP DATABASE "test_postgres";', params = None

    def execute(
        self,
        query: Query,
        params: Optional[Params] = None,
        *,
        prepare: Optional[bool] = None,
        binary: Optional[bool] = None,
    ) -> Self:
        """
        Execute a query or command to the database.
        """
        try:
            with self._conn.lock:
                self._conn.wait(
                    self._execute_gen(query, params, prepare=prepare, binary=binary)
                )
        except e._NO_TRACEBACK as ex:
>           raise ex.with_traceback(None)
E           psycopg.errors.ObjectInUse: database "test_postgres" is being accessed by other users
E           DETAIL:  There is 1 other session using the database.

/opt/pysetup/.venv/lib/python3.12/site-packages/psycopg/cursor.py:732: ObjectInUse
------------------------------------------------------------------------------------------------------------------------------ Captured log call -------------------------------------------------------------------------------------------------------------------------------
INFO     root:reset_db.py:181 Executing... "DROP DATABASE "test_postgres";"
=========================================================================================================================== short test summary info ============================================================================================================================
FAILED adit_radis_shared/accounts/tests/test_migrations.py::test_foobar - psycopg.errors.ObjectInUse: database "test_postgres" is being accessed by other users

EDIT: I worked around it by closing the old connections before resetting the database:

@pytest.mark.django_db
def test_foobar(migrator: Migrator):
    close_old_connections()
    call_command("reset_db", "--noinput")
    assert True

So, does anything speak against dropping the whole database and integrating something like the code of reset_db into django-reset-migrations? Should it be optional or replace the current drop_models_tables?