import pytest from aerich.ddl.mysql import MysqlDDL from aerich.ddl.postgres import PostgresDDL from aerich.ddl.sqlite import SqliteDDL from aerich.exceptions import NotSupportError from aerich.migrate import Migrate from aerich.utils import get_models_describe old_models_describe = { "models.Category": { "name": "models.Category", "app": "models", "table": "category", "abstract": False, "description": None, "docstring": None, "unique_together": [], "pk_field": { "name": "id", "field_type": "IntField", "db_column": "id", "python_type": "int", "generated": True, "nullable": False, "unique": True, "indexed": True, "default": None, "description": None, "docstring": None, "constraints": {"ge": 1, "le": 2147483647}, "db_field_types": {"": "INT"}, }, "data_fields": [ { "name": "slug", "field_type": "CharField", "db_column": "slug", "python_type": "str", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {"max_length": 200}, "db_field_types": {"": "VARCHAR(200)"}, }, { "name": "name", "field_type": "CharField", "db_column": "name", "python_type": "str", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {"max_length": 200}, "db_field_types": {"": "VARCHAR(200)"}, }, { "name": "created_at", "field_type": "DatetimeField", "db_column": "created_at", "python_type": "datetime.datetime", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {"readOnly": True}, "db_field_types": { "": "TIMESTAMP", "mysql": "DATETIME(6)", "postgres": "TIMESTAMPTZ", }, }, { "name": "user_id", "field_type": "IntField", "db_column": "user_id", "python_type": "int", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": "User", "docstring": None, "constraints": {"ge": 1, "le": 2147483647}, "db_field_types": {"": "INT"}, }, ], "fk_fields": [ { "name": "user", "field_type": "ForeignKeyFieldInstance", "python_type": "models.User", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": "User", "docstring": None, "constraints": {}, "raw_field": "user_id", "on_delete": "CASCADE", } ], "backward_fk_fields": [], "o2o_fields": [], "backward_o2o_fields": [], "m2m_fields": [ { "name": "products", "field_type": "ManyToManyFieldInstance", "python_type": "models.Product", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {}, } ], }, "models.Config": { "name": "models.Config", "app": "models", "table": "config", "abstract": False, "description": None, "docstring": None, "unique_together": [], "pk_field": { "name": "id", "field_type": "IntField", "db_column": "id", "python_type": "int", "generated": True, "nullable": False, "unique": True, "indexed": True, "default": None, "description": None, "docstring": None, "constraints": {"ge": 1, "le": 2147483647}, "db_field_types": {"": "INT"}, }, "data_fields": [ { "name": "label", "field_type": "CharField", "db_column": "label", "python_type": "str", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {"max_length": 200}, "db_field_types": {"": "VARCHAR(200)"}, }, { "name": "key", "field_type": "CharField", "db_column": "key", "python_type": "str", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {"max_length": 20}, "db_field_types": {"": "VARCHAR(20)"}, }, { "name": "value", "field_type": "JSONField", "db_column": "value", "python_type": "Union[dict, list]", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {}, "db_field_types": {"": "TEXT", "postgres": "JSONB"}, }, { "name": "status", "field_type": "IntEnumFieldInstance", "db_column": "status", "python_type": "int", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": 1, "description": "on: 1\noff: 0", "docstring": None, "constraints": {"ge": -32768, "le": 32767}, "db_field_types": {"": "SMALLINT"}, }, ], "fk_fields": [], "backward_fk_fields": [], "o2o_fields": [], "backward_o2o_fields": [], "m2m_fields": [], }, "models.Email": { "name": "models.Email", "app": "models", "table": "email", "abstract": False, "description": None, "docstring": None, "unique_together": [], "pk_field": { "name": "id", "field_type": "IntField", "db_column": "id", "python_type": "int", "generated": True, "nullable": False, "unique": True, "indexed": True, "default": None, "description": None, "docstring": None, "constraints": {"ge": 1, "le": 2147483647}, "db_field_types": {"": "INT"}, }, "data_fields": [ { "name": "email", "field_type": "CharField", "db_column": "email", "python_type": "str", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {"max_length": 200}, "db_field_types": {"": "VARCHAR(200)"}, }, { "name": "is_primary", "field_type": "BooleanField", "db_column": "is_primary", "python_type": "bool", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": False, "description": None, "docstring": None, "constraints": {}, "db_field_types": {"": "BOOL", "sqlite": "INT"}, }, { "name": "user_id", "field_type": "IntField", "db_column": "user_id", "python_type": "int", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {"ge": 1, "le": 2147483647}, "db_field_types": {"": "INT"}, }, ], "fk_fields": [ { "name": "user", "field_type": "ForeignKeyFieldInstance", "python_type": "models.User", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {}, "raw_field": "user_id", "on_delete": "CASCADE", } ], "backward_fk_fields": [], "o2o_fields": [], "backward_o2o_fields": [], "m2m_fields": [], }, "models.Product": { "name": "models.Product", "app": "models", "table": "product", "abstract": False, "description": None, "docstring": None, "unique_together": [], "pk_field": { "name": "id", "field_type": "IntField", "db_column": "id", "python_type": "int", "generated": True, "nullable": False, "unique": True, "indexed": True, "default": None, "description": None, "docstring": None, "constraints": {"ge": 1, "le": 2147483647}, "db_field_types": {"": "INT"}, }, "data_fields": [ { "name": "name", "field_type": "CharField", "db_column": "name", "python_type": "str", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {"max_length": 50}, "db_field_types": {"": "VARCHAR(50)"}, }, { "name": "view_num", "field_type": "IntField", "db_column": "view_num", "python_type": "int", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": "View Num", "docstring": None, "constraints": {"ge": -2147483648, "le": 2147483647}, "db_field_types": {"": "INT"}, }, { "name": "sort", "field_type": "IntField", "db_column": "sort", "python_type": "int", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {"ge": -2147483648, "le": 2147483647}, "db_field_types": {"": "INT"}, }, { "name": "is_reviewed", "field_type": "BooleanField", "db_column": "is_reviewed", "python_type": "bool", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": "Is Reviewed", "docstring": None, "constraints": {}, "db_field_types": {"": "BOOL", "sqlite": "INT"}, }, { "name": "type", "field_type": "IntEnumFieldInstance", "db_column": "type", "python_type": "int", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": "Product Type", "docstring": None, "constraints": {"ge": -32768, "le": 32767}, "db_field_types": {"": "SMALLINT"}, }, { "name": "image", "field_type": "CharField", "db_column": "image", "python_type": "str", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {"max_length": 200}, "db_field_types": {"": "VARCHAR(200)"}, }, { "name": "body", "field_type": "TextField", "db_column": "body", "python_type": "str", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {}, "db_field_types": {"": "TEXT", "mysql": "LONGTEXT"}, }, { "name": "created_at", "field_type": "DatetimeField", "db_column": "created_at", "python_type": "datetime.datetime", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {"readOnly": True}, "db_field_types": { "": "TIMESTAMP", "mysql": "DATETIME(6)", "postgres": "TIMESTAMPTZ", }, }, ], "fk_fields": [], "backward_fk_fields": [], "o2o_fields": [], "backward_o2o_fields": [], "m2m_fields": [ { "name": "categories", "field_type": "ManyToManyFieldInstance", "python_type": "models.Category", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {}, } ], }, "models.User": { "name": "models.User", "app": "models", "table": "user", "abstract": False, "description": None, "docstring": None, "unique_together": [], "pk_field": { "name": "id", "field_type": "IntField", "db_column": "id", "python_type": "int", "generated": True, "nullable": False, "unique": True, "indexed": True, "default": None, "description": None, "docstring": None, "constraints": {"ge": 1, "le": 2147483647}, "db_field_types": {"": "INT"}, }, "data_fields": [ { "name": "username", "field_type": "CharField", "db_column": "username", "python_type": "str", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {"max_length": 20}, "db_field_types": {"": "VARCHAR(20)"}, }, { "name": "password", "field_type": "CharField", "db_column": "password", "python_type": "str", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {"max_length": 200}, "db_field_types": {"": "VARCHAR(200)"}, }, { "name": "last_login", "field_type": "DatetimeField", "db_column": "last_login", "python_type": "datetime.datetime", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": "", "description": "Last Login", "docstring": None, "constraints": {}, "db_field_types": { "": "TIMESTAMP", "mysql": "DATETIME(6)", "postgres": "TIMESTAMPTZ", }, }, { "name": "is_active", "field_type": "BooleanField", "db_column": "is_active", "python_type": "bool", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": True, "description": "Is Active", "docstring": None, "constraints": {}, "db_field_types": {"": "BOOL", "sqlite": "INT"}, }, { "name": "is_superuser", "field_type": "BooleanField", "db_column": "is_superuser", "python_type": "bool", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": False, "description": "Is SuperUser", "docstring": None, "constraints": {}, "db_field_types": {"": "BOOL", "sqlite": "INT"}, }, { "name": "avatar", "field_type": "CharField", "db_column": "avatar", "python_type": "str", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": "", "description": None, "docstring": None, "constraints": {"max_length": 200}, "db_field_types": {"": "VARCHAR(200)"}, }, { "name": "intro", "field_type": "TextField", "db_column": "intro", "python_type": "str", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": "", "description": None, "docstring": None, "constraints": {}, "db_field_types": {"": "TEXT", "mysql": "LONGTEXT"}, }, ], "fk_fields": [], "backward_fk_fields": [ { "name": "categorys", "field_type": "BackwardFKRelation", "python_type": "models.Category", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": "User", "docstring": None, "constraints": {}, }, { "name": "emails", "field_type": "BackwardFKRelation", "python_type": "models.Email", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {}, }, ], "o2o_fields": [], "backward_o2o_fields": [], "m2m_fields": [], }, "models.Aerich": { "name": "models.Aerich", "app": "models", "table": "aerich", "abstract": False, "description": None, "docstring": None, "unique_together": [], "pk_field": { "name": "id", "field_type": "IntField", "db_column": "id", "python_type": "int", "generated": True, "nullable": False, "unique": True, "indexed": True, "default": None, "description": None, "docstring": None, "constraints": {"ge": 1, "le": 2147483647}, "db_field_types": {"": "INT"}, }, "data_fields": [ { "name": "version", "field_type": "CharField", "db_column": "version", "python_type": "str", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {"max_length": 255}, "db_field_types": {"": "VARCHAR(255)"}, }, { "name": "app", "field_type": "CharField", "db_column": "app", "python_type": "str", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {"max_length": 20}, "db_field_types": {"": "VARCHAR(20)"}, }, { "name": "content", "field_type": "JSONField", "db_column": "content", "python_type": "Union[dict, list]", "generated": False, "nullable": False, "unique": False, "indexed": False, "default": None, "description": None, "docstring": None, "constraints": {}, "db_field_types": {"": "TEXT", "postgres": "JSONB"}, }, ], "fk_fields": [], "backward_fk_fields": [], "o2o_fields": [], "backward_o2o_fields": [], "m2m_fields": [], }, } def test_migrate(): """ models.py diff with old_models.py - add field: Email.address - add fk: Config.user - drop fk: Email.user - drop field: User.avatar - add index: Email.email - remove unique: User.username """ models_describe = get_models_describe("models") Migrate.diff_models(old_models_describe, models_describe) if isinstance(Migrate.ddl, SqliteDDL): with pytest.raises(NotSupportError): Migrate.diff_models(models_describe, old_models_describe, False) else: Migrate.diff_models(models_describe, old_models_describe, False) Migrate._merge_operators() if isinstance(Migrate.ddl, MysqlDDL): assert Migrate.upgrade_operators == [ "ALTER TABLE `email` DROP FOREIGN KEY `fk_email_user_5b58673d`", "ALTER TABLE `category` ADD `name` VARCHAR(200) NOT NULL", "ALTER TABLE `user` ADD UNIQUE INDEX `uid_user_usernam_9987ab` (`username`)", "ALTER TABLE `user` RENAME COLUMN `last_login_at` TO `last_login`", ] assert Migrate.downgrade_operators == [ "ALTER TABLE `category` DROP COLUMN `name`", "ALTER TABLE `user` DROP INDEX `uid_user_usernam_9987ab`", "ALTER TABLE `user` RENAME COLUMN `last_login` TO `last_login_at`", "ALTER TABLE `email` ADD CONSTRAINT `fk_email_user_5b58673d` FOREIGN KEY " "(`user_id`) REFERENCES `user` (`id`) ON DELETE CASCADE", ] elif isinstance(Migrate.ddl, PostgresDDL): assert Migrate.upgrade_operators == [ 'ALTER TABLE "email" DROP CONSTRAINT "fk_email_user_5b58673d"', 'ALTER TABLE "category" ADD "name" VARCHAR(200) NOT NULL', 'ALTER TABLE "user" ADD CONSTRAINT "uid_user_usernam_9987ab" UNIQUE ("username")', 'ALTER TABLE "user" RENAME COLUMN "last_login_at" TO "last_login"', ] assert Migrate.downgrade_operators == [ 'ALTER TABLE "category" DROP COLUMN "name"', 'ALTER TABLE "user" DROP CONSTRAINT "uid_user_usernam_9987ab"', 'ALTER TABLE "user" RENAME COLUMN "last_login" TO "last_login_at"', 'ALTER TABLE "email" ADD CONSTRAINT "fk_email_user_5b58673d" FOREIGN KEY ("user_id") REFERENCES "user" ("id") ON DELETE CASCADE', ] elif isinstance(Migrate.ddl, SqliteDDL): assert Migrate.upgrade_operators == [ 'ALTER TABLE "email" DROP FOREIGN KEY "fk_email_user_5b58673d"', 'ALTER TABLE "category" ADD "name" VARCHAR(200) NOT NULL', 'ALTER TABLE "user" ADD UNIQUE INDEX "uid_user_usernam_9987ab" ("username")', 'ALTER TABLE "user" RENAME COLUMN "last_login_at" TO "last_login"', ] assert Migrate.downgrade_operators == [] def test_sort_all_version_files(mocker): mocker.patch( "os.listdir", return_value=[ "1_datetime_update.sql", "11_datetime_update.sql", "10_datetime_update.sql", "2_datetime_update.sql", ], ) Migrate.migrate_location = "." assert Migrate.get_all_version_files() == [ "1_datetime_update.sql", "2_datetime_update.sql", "10_datetime_update.sql", "11_datetime_update.sql", ]