piccolo/tests/apps/migrations/commands/test_clean.py at array_migrations · hipertracker/piccolo · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
from unittest import TestCase

from piccolo.apps.migrations.commands.clean import clean
from piccolo.apps.migrations.tables import Migration
from piccolo.utils.sync import run_sync


class TestCleanMigrationCommand(TestCase):
    def test_clean(self):
        Migration.create_table(if_not_exists=True).run_sync()

        real_migration_ids = [
            "2020-12-17T18:44:30",
            "2020-12-17T18:44:39",
            "2020-12-17T18:44:44",
        ]

        orphaned_migration_id = "2010-01-101T00:00:00"

        migration_ids = real_migration_ids + [orphaned_migration_id]

        Migration.insert(
            *[Migration(name=i, app_name="example_app") for i in migration_ids]
        ).run_sync()

        run_sync(clean(app_name="example_app", auto_agree=True))

        remaining_rows = (
            Migration.select(Migration.name)
            .where(Migration.app_name == "example_app")
            .output(as_list=True)
            .order_by(Migration.name)
            .run_sync()
        )
        self.assertEqual(remaining_rows, real_migration_ids)

        Migration.alter().drop_table(if_exists=True).run_sync()