diff --git a/fastapi_startkit/tests/masoniteorm/query/__init__.py b/fastapi_startkit/tests/masoniteorm/query/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/fastapi_startkit/tests/masoniteorm/query/grammars/__init__.py b/fastapi_startkit/tests/masoniteorm/query/grammars/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/fastapi_startkit/tests/masoniteorm/query/grammars/test_mssql_grammar.py b/fastapi_startkit/tests/masoniteorm/query/grammars/test_mssql_grammar.py new file mode 100644 index 00000000..640d4488 --- /dev/null +++ b/fastapi_startkit/tests/masoniteorm/query/grammars/test_mssql_grammar.py @@ -0,0 +1,154 @@ +import unittest + +from fastapi_startkit.masoniteorm.models.builder import QueryBuilder +from fastapi_startkit.masoniteorm.query.grammars.MSSQLGrammar import MSSQLGrammar +from fastapi_startkit.masoniteorm.query.processors.MSSQLPostProcessor import MSSQLPostProcessor + + +class TestMSSQLGrammarSelect(unittest.TestCase): + def setUp(self): + query = QueryBuilder(connection=None, grammar=MSSQLGrammar, processor=MSSQLPostProcessor) + query._table = "users" + self.query = query + + def test_select_all(self): + self.assertEqual(self.query.to_sql(), "SELECT * FROM [users]") + + def test_select_columns(self): + self.assertEqual( + self.query.select("name", "email").to_sql(), + "SELECT [users].[name], [users].[email] FROM [users]", + ) + + def test_select_distinct(self): + self.assertEqual( + self.query.select("name").distinct().to_sql(), + "SELECT DISTINCT [users].[name] FROM [users]", + ) + + def test_where_with_operator(self): + self.assertEqual( + self.query.where("age", ">", 18).to_sql(), + "SELECT * FROM [users] WHERE [users].[age] > '18'", + ) + + def test_where_equals(self): + self.assertEqual( + self.query.where("name", "Joe").to_sql(), + "SELECT * FROM [users] WHERE [users].[name] = 'Joe'", + ) + + def test_or_where(self): + self.assertEqual( + self.query.where("age", 18).or_where("age", "<", 5).to_sql(), + "SELECT * FROM [users] WHERE [users].[age] = '18' OR [users].[age] < '5'", + ) + + def test_where_null(self): + self.assertEqual( + self.query.where_null("name").to_sql(), + "SELECT * FROM [users] WHERE [users].[name] IS NULL", + ) + + def test_where_in(self): + self.assertEqual( + self.query.where_in("id", [1, 2, 3]).to_sql(), + "SELECT * FROM [users] WHERE [users].[id] IN ('1','2','3')", + ) + + def test_where_between(self): + self.assertEqual( + self.query.between("age", 10, 20).to_sql(), + "SELECT * FROM [users] WHERE [users].[age] BETWEEN '10' AND '20'", + ) + + def test_where_like(self): + self.assertEqual( + self.query.where("name", "like", "%Jo%").to_sql(), + "SELECT * FROM [users] WHERE [users].[name] LIKE '%Jo%'", + ) + + def test_limit_uses_top(self): + self.assertEqual(self.query.limit(5).to_sql(), "SELECT TOP 5 * FROM [users]") + + def test_offset_uses_fetch_next(self): + self.assertEqual( + self.query.limit(5).offset(10).to_sql(), + "SELECT * FROM [users] OFFSET 10 ROWS FETCH NEXT 5 ROWS ONLY", + ) + + def test_order_by(self): + self.assertEqual( + self.query.order_by("name", "desc").to_sql(), + "SELECT * FROM [users] ORDER BY [name] DESC", + ) + + def test_group_by(self): + self.assertEqual( + self.query.group_by("age").to_sql(), + "SELECT * FROM [users] GROUP BY [users].[age]", + ) + + def test_inner_join(self): + qb = self.query.join("profiles", "users.id", "=", "profiles.user_id") + self.assertEqual( + qb.to_sql(), + "SELECT * FROM [users] INNER JOIN [profiles] ON [users].[id] = [profiles].[user_id]", + ) + + +class TestMSSQLGrammarWrites(unittest.TestCase): + def setUp(self): + query = QueryBuilder(connection=None, grammar=MSSQLGrammar, processor=MSSQLPostProcessor) + query._table = "users" + self.query = query + + def test_insert(self): + grammar = MSSQLGrammar(columns={"name": "Joe", "email": "j@x.com"}, table="users") + self.assertEqual( + grammar.compile("insert")._sql, + "INSERT INTO [users] ([users].[name], [users].[email]) VALUES ('Joe', 'j@x.com')", + ) + + def test_delete(self): + qb = self.query.where("id", 1).set_action("delete") + self.assertEqual(qb.to_sql(), "DELETE FROM [users] WHERE [users].[id] = '1'") + + +class TestMSSQLGrammarPrimitives(unittest.TestCase): + def setUp(self): + self.grammar = MSSQLGrammar() + + def test_table_string_wraps_in_brackets(self): + self.assertEqual(self.grammar.wrap_table("users"), "[users]") + + def test_truncate_table(self): + self.assertEqual(self.grammar.truncate_table("users"), "TRUNCATE TABLE [users]") + + def test_compile_random_uses_newid(self): + self.assertEqual(self.grammar.compile_random(5), "NEWID()") + + def test_limit_string_without_offset(self): + self.assertEqual(self.grammar.limit_string(), "TOP {limit}") + + def test_limit_string_with_offset_is_empty(self): + self.assertEqual(self.grammar.limit_string(offset=True), "") + + def test_increment_and_decrement_strings(self): + self.assertEqual(self.grammar.increment_string(), "{column} = {column} + {value}") + self.assertEqual(self.grammar.decrement_string(), "{column} = {column} - {value}") + + def test_insert_and_bulk_insert_formats(self): + self.assertEqual( + self.grammar.insert_format(), + "INSERT INTO {table} ({columns}) VALUES ({values})", + ) + self.assertEqual( + self.grammar.bulk_insert_format(), + "INSERT INTO {table} ({columns}) VALUES {values}", + ) + + def test_join_keywords_mapping(self): + self.assertEqual(self.grammar.join_keywords["left"], "LEFT JOIN") + self.assertEqual(self.grammar.join_keywords["right"], "RIGHT JOIN") + self.assertEqual(self.grammar.join_keywords["outer"], "OUTER JOIN") diff --git a/fastapi_startkit/tests/masoniteorm/query/processors/__init__.py b/fastapi_startkit/tests/masoniteorm/query/processors/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/fastapi_startkit/tests/masoniteorm/query/processors/test_post_processors.py b/fastapi_startkit/tests/masoniteorm/query/processors/test_post_processors.py new file mode 100644 index 00000000..bd7008af --- /dev/null +++ b/fastapi_startkit/tests/masoniteorm/query/processors/test_post_processors.py @@ -0,0 +1,157 @@ +import unittest +from unittest.mock import MagicMock + +from fastapi_startkit.masoniteorm.query.processors.MSSQLPostProcessor import MSSQLPostProcessor +from fastapi_startkit.masoniteorm.query.processors.MySQLPostProcessor import MySQLPostProcessor +from fastapi_startkit.masoniteorm.query.processors.PostgresPostProcessor import PostgresPostProcessor +from fastapi_startkit.masoniteorm.query.processors.SQLitePostProcessor import SQLitePostProcessor + + +class TestSQLitePostProcessor(unittest.TestCase): + """Post processor unit tests — connections are mocked, no live database.""" + + def setUp(self): + self.processor = SQLitePostProcessor() + + def test_process_insert_get_id_fetches_last_row_id(self): + builder = MagicMock() + builder.get_connection.return_value.get_last_row_id.return_value = 11 + + result = self.processor.process_insert_get_id(builder, {}, "id") + + self.assertEqual(result, {"id": 11}) + + def test_process_insert_get_id_keeps_existing_key(self): + builder = MagicMock() + + result = self.processor.process_insert_get_id(builder, {"id": 5}, "id") + + self.assertEqual(result, {"id": 5}) + builder.get_connection.assert_not_called() + + def test_get_column_value_refetches_from_row(self): + builder = MagicMock() + new_builder = builder.select.return_value + new_builder.first.return_value = {"name": "Joe"} + + result = self.processor.get_column_value(builder, "name", {}, "id", 1) + + self.assertEqual(result, "Joe") + new_builder.where.assert_called_once_with("id", 1) + + def test_get_column_value_without_id_returns_empty(self): + builder = MagicMock() + + result = self.processor.get_column_value(builder, "name", {}, None, None) + + self.assertEqual(result, {}) + + +class TestMySQLPostProcessor(unittest.TestCase): + def setUp(self): + self.processor = MySQLPostProcessor() + + def test_process_insert_get_id_uses_cursor_lastrowid(self): + builder = MagicMock() + builder._connection.get_cursor.return_value.lastrowid = 42 + + result = self.processor.process_insert_get_id(builder, {}, "id") + + self.assertEqual(result, {"id": 42}) + + def test_process_insert_get_id_keeps_existing_key(self): + builder = MagicMock() + + result = self.processor.process_insert_get_id(builder, {"id": 7}, "id") + + self.assertEqual(result, {"id": 7}) + builder._connection.get_cursor.assert_not_called() + + def test_get_column_value_refetches_from_row(self): + builder = MagicMock() + builder.select.return_value.first.return_value = {"email": "a@b.com"} + + result = self.processor.get_column_value(builder, "email", {}, "id", 3) + + self.assertEqual(result, "a@b.com") + + def test_get_column_value_without_id_returns_empty(self): + builder = MagicMock() + + result = self.processor.get_column_value(builder, "email", {}, None, None) + + self.assertEqual(result, {}) + + +class TestPostgresPostProcessor(unittest.TestCase): + def setUp(self): + self.processor = PostgresPostProcessor() + + def test_process_insert_get_id_maps_lastval_fallback(self): + result = self.processor.process_insert_get_id(MagicMock(), {"lastval": 9}, "id") + + self.assertEqual(result, {"id": 9}) + + def test_process_insert_get_id_passes_through_returning_row(self): + results = {"id": 3, "name": "Joe"} + + result = self.processor.process_insert_get_id(MagicMock(), results, "id") + + self.assertEqual(result, results) + + def test_get_column_value_reads_present_column(self): + result = self.processor.get_column_value(MagicMock(), "name", {"name": "Bob"}, "id", 1) + + self.assertEqual(result, "Bob") + + def test_get_column_value_refetches_missing_column(self): + builder = MagicMock() + builder.select.return_value.first.return_value = {"name": "Sue"} + + result = self.processor.get_column_value(builder, "name", {}, "id", 2) + + self.assertEqual(result, "Sue") + + def test_get_column_value_without_id_returns_empty(self): + builder = MagicMock() + + result = self.processor.get_column_value(builder, "name", {}, None, None) + + self.assertEqual(result, {}) + + +class TestMSSQLPostProcessor(unittest.TestCase): + def setUp(self): + self.processor = MSSQLPostProcessor() + + def test_process_insert_get_id_casts_numeric_identity(self): + builder = MagicMock() + builder.new_connection.return_value.query.return_value = {"id": "9"} + + result = self.processor.process_insert_get_id(builder, {}, "id") + + self.assertEqual(result, {"id": 9}) + self.assertIsInstance(result["id"], int) + + def test_process_insert_get_id_keeps_non_numeric_identity_as_string(self): + builder = MagicMock() + builder.new_connection.return_value.query.return_value = {"id": "abc"} + + result = self.processor.process_insert_get_id(builder, {}, "id") + + self.assertEqual(result, {"id": "abc"}) + + def test_get_column_value_refetches_from_row(self): + builder = MagicMock() + builder.select.return_value.first.return_value = {"name": "Zed"} + + result = self.processor.get_column_value(builder, "name", {}, "id", 4) + + self.assertEqual(result, "Zed") + + def test_get_column_value_without_id_returns_empty(self): + builder = MagicMock() + + result = self.processor.get_column_value(builder, "name", {}, None, None) + + self.assertEqual(result, {}) diff --git a/fastapi_startkit/tests/masoniteorm/schema/__init__.py b/fastapi_startkit/tests/masoniteorm/schema/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/fastapi_startkit/tests/masoniteorm/schema/platforms/__init__.py b/fastapi_startkit/tests/masoniteorm/schema/platforms/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/fastapi_startkit/tests/masoniteorm/schema/platforms/test_mssql_platform.py b/fastapi_startkit/tests/masoniteorm/schema/platforms/test_mssql_platform.py new file mode 100644 index 00000000..77aa80b8 --- /dev/null +++ b/fastapi_startkit/tests/masoniteorm/schema/platforms/test_mssql_platform.py @@ -0,0 +1,180 @@ +import unittest + +from fastapi_startkit.masoniteorm.schema.Column import Column +from fastapi_startkit.masoniteorm.schema.Table import Table +from fastapi_startkit.masoniteorm.schema.TableDiff import TableDiff +from fastapi_startkit.masoniteorm.schema.platforms import MSSQLPlatform + + +class TestMSSQLPlatformCreate(unittest.TestCase): + """Schema SQL generation for MSSQL — pure string compilation, no live connection.""" + + def setUp(self): + self.platform = MSSQLPlatform() + + def _basic_table(self): + table = Table("users") + table.add_column("id", "integer", primary=True) + table.add_column("name", "string", length=255) + table.add_column("bio", "string", length=255, nullable=True) + return table + + def test_compile_create_sql(self): + sql = self.platform.compile_create_sql(self._basic_table()) + self.assertEqual( + sql, + [ + "CREATE TABLE [users] ([id] INT NOT NULL PRIMARY KEY, [name] VARCHAR(255) NOT NULL, [bio] VARCHAR(255) NULL)" + ], + ) + + def test_compile_create_sql_if_not_exists(self): + sql = self.platform.compile_create_sql(self._basic_table(), if_not_exists=True) + self.assertTrue(sql[0].startswith("CREATE TABLE IF NOT EXISTS [users]")) + + def test_create_with_enum_check_constraint(self): + table = Table("users") + table.add_column("status", "enum", values=["on", "off"]) + sql = self.platform.compile_create_sql(table)[0] + self.assertIn("[status] VARCHAR NOT NULL CHECK([status] IN ('on', 'off'))", sql) + + def test_create_with_foreign_key(self): + table = self._basic_table() + table.add_foreign_key("profile_id", table="profiles", foreign_column="id").on_delete("cascade") + sql = self.platform.compile_create_sql(table)[0] + self.assertIn( + "CONSTRAINT users_profile_id_foreign FOREIGN KEY ([profile_id]) REFERENCES [profiles]([id]) ON DELETE CASCADE", + sql, + ) + + def test_create_appends_index_statement(self): + table = self._basic_table() + table.add_index(["email"], "users_email_index", "index") + sql = self.platform.compile_create_sql(table) + self.assertEqual(sql[1], "CREATE INDEX users_email_index ON [users](email)") + + +class TestMSSQLPlatformAlter(unittest.TestCase): + def setUp(self): + self.platform = MSSQLPlatform() + + def test_add_column(self): + diff = TableDiff("users") + diff.add_column("age", "integer") + self.assertEqual(self.platform.compile_alter_sql(diff), ["ALTER TABLE [users] ADD [age] INT NOT NULL"]) + + def test_change_column(self): + diff = TableDiff("users") + diff.changed_columns = {"age": Column("age", "integer")} + self.assertEqual( + self.platform.compile_alter_sql(diff), + ["ALTER TABLE [users] ALTER COLUMN [age] INT NOT NULL"], + ) + + def test_rename_column(self): + diff = TableDiff("users") + diff.rename_column("old_name", "new_name", "string", length=255) + self.assertEqual( + self.platform.compile_alter_sql(diff), + ["EXEC sp_rename 'users.old_name', 'new_name', 'COLUMN'"], + ) + + def test_drop_column(self): + diff = TableDiff("users") + diff.drop_column("age") + self.assertEqual(self.platform.compile_alter_sql(diff), ["ALTER TABLE [users] DROP COLUMN age"]) + + def test_add_foreign_key(self): + diff = TableDiff("users") + diff.add_foreign_key("profile_id", table="profiles", foreign_column="id").on_delete("cascade") + self.assertEqual( + self.platform.compile_alter_sql(diff), + [ + "ALTER TABLE [users] ADD CONSTRAINT users_profile_id_foreign FOREIGN KEY ([profile_id]) " + "REFERENCES [profiles]([id]) ON DELETE CASCADE" + ], + ) + + def test_add_primary_key_constraint(self): + diff = TableDiff("users") + diff.add_constraint("pk", "primary_key", columns=["id"]) + self.assertEqual( + self.platform.compile_alter_sql(diff), + ["ALTER TABLE [users] ADD CONSTRAINT pk PRIMARY KEY (id)"], + ) + + def test_drop_foreign_key(self): + diff = TableDiff("users") + diff.drop_foreign("fk_1") + self.assertEqual( + self.platform.compile_alter_sql(diff), + ["ALTER TABLE [users] DROP CONSTRAINT fk_1"], + ) + + def test_remove_index(self): + diff = TableDiff("users") + diff.remove_index("idx_1") + self.assertEqual(self.platform.compile_alter_sql(diff), ["DROP INDEX [users].[idx_1]"]) + + +class TestMSSQLPlatformHelpers(unittest.TestCase): + def setUp(self): + self.platform = MSSQLPlatform() + + def test_compile_drop_table(self): + self.assertEqual(self.platform.compile_drop_table("users"), "DROP TABLE [users]") + + def test_compile_drop_table_if_exists(self): + self.assertEqual(self.platform.compile_drop_table_if_exists("users"), "DROP TABLE IF EXISTS [users]") + + def test_compile_rename_table(self): + self.assertEqual( + self.platform.compile_rename_table("users", "clients"), + "EXEC sp_rename [users], [clients]", + ) + + def test_compile_truncate(self): + self.assertEqual(self.platform.compile_truncate("users"), "TRUNCATE TABLE [users]") + + def test_compile_truncate_with_foreign_keys(self): + self.assertEqual( + self.platform.compile_truncate("users", foreign_keys=True), + [ + "ALTER TABLE [users] NOCHECK CONSTRAINT ALL", + "TRUNCATE TABLE [users]", + "ALTER TABLE [users] WITH CHECK CHECK CONSTRAINT ALL", + ], + ) + + def test_compile_table_exists(self): + self.assertEqual( + self.platform.compile_table_exists("users"), + "SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 'users'", + ) + + def test_compile_column_exists(self): + self.assertEqual( + self.platform.compile_column_exists("users", "name"), + "SELECT 1 FROM sys.columns WHERE Name = N'name' AND Object_ID = Object_ID(N'users')", + ) + + def test_compile_get_all_tables(self): + self.assertEqual(self.platform.compile_get_all_tables("app"), "SELECT name FROM app.sys.tables") + + def test_foreign_key_toggles_are_noops(self): + self.assertEqual(self.platform.enable_foreign_key_constraints(), "") + self.assertEqual(self.platform.disable_foreign_key_constraints(), "") + + def test_rename_column_string(self): + self.assertEqual( + self.platform.rename_column_string("users", "old", "new"), + "EXEC sp_rename 'users.old', 'new', 'COLUMN'", + ) + + def test_get_current_schema_returns_table(self): + table = self.platform.get_current_schema(connection=None, table_name="users") + self.assertEqual(table.name, "users") + + def test_wrap_helpers(self): + self.assertEqual(self.platform.wrap_table("users"), "[users]") + self.assertEqual(self.platform.wrap_column("name"), "[name]") diff --git a/fastapi_startkit/tests/masoniteorm/schema/platforms/test_mysql_platform.py b/fastapi_startkit/tests/masoniteorm/schema/platforms/test_mysql_platform.py new file mode 100644 index 00000000..9ba78a59 --- /dev/null +++ b/fastapi_startkit/tests/masoniteorm/schema/platforms/test_mysql_platform.py @@ -0,0 +1,224 @@ +import unittest + +from fastapi_startkit.masoniteorm.schema.Table import Table +from fastapi_startkit.masoniteorm.schema.TableDiff import TableDiff +from fastapi_startkit.masoniteorm.schema.platforms import MySQLPlatform + + +class TestMySQLPlatformCreate(unittest.TestCase): + """Schema SQL generation for MySQL — pure string compilation, no live connection.""" + + def setUp(self): + self.platform = MySQLPlatform() + + def _basic_table(self): + table = Table("users") + table.add_column("id", "integer", primary=True) + table.add_column("name", "string", length=255) + table.add_column("bio", "string", length=255, nullable=True) + return table + + def test_compile_create_sql(self): + sql = self.platform.compile_create_sql(self._basic_table()) + self.assertEqual( + sql, + [ + "CREATE TABLE `users` (`id` INT NOT NULL PRIMARY KEY, `name` VARCHAR(255) NOT NULL, `bio` VARCHAR(255) NULL)" + ], + ) + + def test_compile_create_sql_if_not_exists(self): + sql = self.platform.compile_create_sql(self._basic_table(), if_not_exists=True) + self.assertTrue(sql[0].startswith("CREATE TABLE IF NOT EXISTS `users`")) + + def test_create_with_defaults_enum_and_foreign_key(self): + table = Table("users") + table.add_column("active", "integer", default=0) + table.add_column("role", "string", default="admin") + table.add_column("status", "enum", values=["on", "off"]) + table.add_column("created", "timestamp", default="current") + table.add_foreign_key("profile_id", table="profiles", foreign_column="id").on_delete("cascade") + + sql = self.platform.compile_create_sql(table)[0] + + self.assertIn("`active` INT NOT NULL DEFAULT 0", sql) + self.assertIn("`role` VARCHAR NOT NULL DEFAULT 'admin'", sql) + self.assertIn("`status` ENUM('on', 'off') NOT NULL", sql) + self.assertIn("`created` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP", sql) + self.assertIn( + "CONSTRAINT users_profile_id_foreign FOREIGN KEY (`profile_id`) REFERENCES `profiles`(`id`) ON DELETE CASCADE", + sql, + ) + + def test_create_appends_index_statement(self): + table = self._basic_table() + table.add_index(["email"], "users_email_index", "index") + + sql = self.platform.compile_create_sql(table) + + self.assertEqual(sql[1], "CREATE INDEX users_email_index ON `users`(email)") + + def test_columnize_unsigned_and_comment(self): + table = Table("stats") + column = table.add_column("views", "integer") + column.unsigned() + column.add_comment("hits") + + self.assertEqual( + self.platform.columnize(table.get_added_columns()), + ["`views` INT UNSIGNED NOT NULL COMMENT 'hits'"], + ) + + +class TestMySQLPlatformAlter(unittest.TestCase): + def setUp(self): + self.platform = MySQLPlatform() + + def test_add_column(self): + diff = TableDiff("users") + diff.add_column("age", "integer") + self.assertEqual(self.platform.compile_alter_sql(diff), ["ALTER TABLE `users` ADD `age` INT NOT NULL"]) + + def test_add_column_with_default_and_enum(self): + diff = TableDiff("users") + diff.add_column("role", "string", default="admin") + diff.add_column("st", "enum", values=["a", "b"]) + self.assertEqual( + self.platform.compile_alter_sql(diff), + ["ALTER TABLE `users` ADD `role` VARCHAR NOT NULL DEFAULT 'admin', ADD `st` ENUM('a', 'b') NOT NULL"], + ) + + def test_rename_column(self): + diff = TableDiff("users") + diff.rename_column("old", "new", "string", length=255) + self.assertEqual( + self.platform.compile_alter_sql(diff), + ["ALTER TABLE `users` CHANGE `old` `new` VARCHAR(255) NOT NULL"], + ) + + def test_change_column(self): + from fastapi_startkit.masoniteorm.schema.Column import Column + + diff = TableDiff("users") + diff.changed_columns = {"age": Column("age", "integer")} + self.assertEqual( + self.platform.compile_alter_sql(diff), + ["ALTER TABLE `users` MODIFY `age` INT NOT NULL"], + ) + + def test_drop_column(self): + diff = TableDiff("users") + diff.drop_column("age") + self.assertEqual(self.platform.compile_alter_sql(diff), ["ALTER TABLE `users` DROP COLUMN `age`"]) + + def test_add_foreign_key(self): + diff = TableDiff("users") + diff.add_foreign_key("profile_id", table="profiles", foreign_column="id").on_delete("cascade") + self.assertEqual( + self.platform.compile_alter_sql(diff), + [ + "ALTER TABLE `users` ADD CONSTRAINT users_profile_id_foreign FOREIGN KEY (profile_id) " + "REFERENCES profiles(id) ON DELETE CASCADE" + ], + ) + + def test_add_index(self): + diff = TableDiff("users") + diff.add_index(["email"], "users_email_index", "index") + self.assertEqual( + self.platform.compile_alter_sql(diff), + ["CREATE INDEX users_email_index ON `users`(email)"], + ) + + def test_add_unique_constraint(self): + diff = TableDiff("users") + diff.add_constraint("u_uniq", "unique", columns=["email"]) + self.assertEqual( + self.platform.compile_alter_sql(diff), + ["ALTER TABLE `users` ADD CONSTRAINT UNIQUE INDEX u_uniq(email)"], + ) + + def test_drop_foreign_key_and_index(self): + diff = TableDiff("users") + diff.drop_foreign("fk_1") + diff.remove_index("idx_1") + self.assertEqual( + self.platform.compile_alter_sql(diff), + [ + "ALTER TABLE `users` DROP FOREIGN KEY fk_1", + "ALTER TABLE `users` DROP INDEX idx_1", + ], + ) + + def test_table_comment(self): + diff = TableDiff("users") + diff.add_column("x", "integer") + diff.add_comment("a table") + self.assertEqual( + self.platform.compile_alter_sql(diff), + [ + "ALTER TABLE `users` ADD `x` INT NOT NULL", + "ALTER TABLE `users` COMMENT 'a table'", + ], + ) + + +class TestMySQLPlatformHelpers(unittest.TestCase): + def setUp(self): + self.platform = MySQLPlatform() + + def test_compile_drop_table(self): + self.assertEqual(self.platform.compile_drop_table("users"), "DROP TABLE `users`") + + def test_compile_drop_table_if_exists(self): + self.assertEqual(self.platform.compile_drop_table_if_exists("users"), "DROP TABLE IF EXISTS `users`") + + def test_compile_rename_table(self): + self.assertEqual( + self.platform.compile_rename_table("users", "clients"), + "ALTER TABLE `users` RENAME TO `clients`", + ) + + def test_compile_truncate(self): + self.assertEqual(self.platform.compile_truncate("users"), "TRUNCATE `users`") + + def test_compile_truncate_with_foreign_keys(self): + self.assertEqual( + self.platform.compile_truncate("users", foreign_keys=True), + ["SET FOREIGN_KEY_CHECKS=0", "TRUNCATE `users`", "SET FOREIGN_KEY_CHECKS=1"], + ) + + def test_compile_table_exists(self): + self.assertEqual( + self.platform.compile_table_exists("users", database="app"), + "SELECT * from information_schema.tables where table_name='users' AND table_schema = 'app'", + ) + + def test_compile_column_exists(self): + self.assertEqual( + self.platform.compile_column_exists("users", "name"), + "SELECT column_name FROM information_schema.columns WHERE table_name='users' and column_name='name'", + ) + + def test_compile_get_all_tables(self): + self.assertEqual( + self.platform.compile_get_all_tables("app"), + "SELECT table_name FROM information_schema.tables WHERE table_schema = 'app'", + ) + + def test_foreign_key_toggles(self): + self.assertEqual(self.platform.enable_foreign_key_constraints(), "SET FOREIGN_KEY_CHECKS=1") + self.assertEqual(self.platform.disable_foreign_key_constraints(), "SET FOREIGN_KEY_CHECKS=0") + + def test_wrap_helpers(self): + self.assertEqual(self.platform.wrap_table("users"), "`users`") + self.assertEqual(self.platform.wrap_column("name"), "`name`") + + def test_get_column_length(self): + self.assertEqual(self.platform.get_column_length("VARCHAR(255)"), "255") + self.assertIsNone(self.platform.get_column_length("INT")) + + def test_get_column_type(self): + reversed_map = {v: k for k, v in self.platform.type_map.items()} + self.assertEqual(self.platform.get_column_type(reversed_map, "CHAR(1)"), "char") + self.assertEqual(self.platform.get_column_type(reversed_map, "INT"), "integer") diff --git a/fastapi_startkit/tests/masoniteorm/schema/platforms/test_postgres_platform.py b/fastapi_startkit/tests/masoniteorm/schema/platforms/test_postgres_platform.py new file mode 100644 index 00000000..a6e83a0d --- /dev/null +++ b/fastapi_startkit/tests/masoniteorm/schema/platforms/test_postgres_platform.py @@ -0,0 +1,222 @@ +import unittest + +from fastapi_startkit.masoniteorm.schema.Table import Table +from fastapi_startkit.masoniteorm.schema.TableDiff import TableDiff +from fastapi_startkit.masoniteorm.schema.platforms import PostgresPlatform + + +class TestPostgresPlatformCreate(unittest.TestCase): + """Schema SQL generation for Postgres — pure string compilation, no live connection.""" + + def setUp(self): + self.platform = PostgresPlatform() + + def _basic_table(self): + table = Table("users") + table.add_column("id", "integer", primary=True) + table.add_column("name", "string", length=255) + table.add_column("bio", "string", length=255, nullable=True) + return table + + def test_compile_create_sql(self): + sql = self.platform.compile_create_sql(self._basic_table()) + self.assertEqual( + sql, + [ + 'CREATE TABLE "users" ("id" INTEGER NOT NULL PRIMARY KEY, "name" VARCHAR(255) NOT NULL, "bio" VARCHAR(255) NULL)' + ], + ) + + def test_compile_create_sql_if_not_exists(self): + sql = self.platform.compile_create_sql(self._basic_table(), if_not_exists=True) + self.assertTrue(sql[0].startswith('CREATE TABLE IF NOT EXISTS "users"')) + + def test_create_with_enum_check_constraint(self): + table = Table("users") + table.add_column("status", "enum", values=["on", "off"]) + sql = self.platform.compile_create_sql(table)[0] + self.assertIn("\"status\" VARCHAR CHECK(status IN ('on', 'off')) NOT NULL", sql) + + def test_create_with_column_comment(self): + table = self._basic_table() + table.get_added_columns()["name"].add_comment("full name") + sql = self.platform.compile_create_sql(table) + self.assertIn("""COMMENT ON COLUMN "users"."name" is 'full name'""", sql) + + def test_create_appends_index_statement(self): + table = self._basic_table() + table.add_index(["email"], "users_email_index", "index") + sql = self.platform.compile_create_sql(table) + self.assertEqual(sql[1], 'CREATE INDEX users_email_index ON "users"(email)') + + def test_integer_types_have_no_length(self): + table = Table("users") + table.add_column("count", "integer", length=11) + sql = self.platform.compile_create_sql(table)[0] + self.assertIn('"count" INTEGER NOT NULL', sql) + + +class TestPostgresPlatformAlter(unittest.TestCase): + def setUp(self): + self.platform = PostgresPlatform() + + def test_add_column(self): + diff = TableDiff("users") + diff.add_column("age", "integer") + self.assertEqual( + self.platform.compile_alter_sql(diff), + ['ALTER TABLE "users" ADD COLUMN "age" INTEGER NOT NULL'], + ) + + def test_drop_column(self): + diff = TableDiff("users") + diff.drop_column("age") + self.assertEqual( + self.platform.compile_alter_sql(diff), + ['ALTER TABLE "users" DROP COLUMN "age"'], + ) + + def test_add_foreign_key(self): + diff = TableDiff("users") + diff.add_foreign_key("profile_id", table="profiles", foreign_column="id").on_delete("cascade") + self.assertEqual( + self.platform.compile_alter_sql(diff), + [ + 'ALTER TABLE "users" ADD CONSTRAINT users_profile_id_foreign FOREIGN KEY ("profile_id") ' + 'REFERENCES "profiles"("id") ON DELETE CASCADE' + ], + ) + + def test_rename_column(self): + diff = TableDiff("users") + diff.rename_column("old", "new", "string", length=255) + self.assertEqual( + self.platform.compile_alter_sql(diff), + ['ALTER TABLE "users" RENAME COLUMN "old" TO "new"'], + ) + + def test_change_column(self): + from fastapi_startkit.masoniteorm.schema.Column import Column + + column = Column("age", "integer") + column.default = 5 + diff = TableDiff("users") + diff.changed_columns = {"age": column} + self.assertEqual( + self.platform.compile_alter_sql(diff), + [ + 'ALTER TABLE "users" ALTER COLUMN "age" TYPE INTEGER, ' + 'ALTER COLUMN "age" SET NOT NULL, ALTER COLUMN "age" SET DEFAULT 5' + ], + ) + + def test_add_primary_key_constraint(self): + diff = TableDiff("users") + diff.add_constraint("pk", "primary_key", columns=["id"]) + self.assertEqual( + self.platform.compile_alter_sql(diff), + ['ALTER TABLE "users" ADD CONSTRAINT pk PRIMARY KEY (id)'], + ) + + def test_add_index(self): + diff = TableDiff("users") + diff.add_index(["email"], "users_email_index", "index") + self.assertEqual( + self.platform.compile_alter_sql(diff), + ['CREATE INDEX users_email_index ON "users"(email)'], + ) + + def test_add_unique_constraint(self): + diff = TableDiff("users") + diff.add_constraint("u_uniq", "unique", columns=["email"]) + self.assertEqual( + self.platform.compile_alter_sql(diff), + ['ALTER TABLE "users" ADD CONSTRAINT u_uniq UNIQUE(email)'], + ) + + def test_drop_foreign_key_and_index(self): + diff = TableDiff("users") + diff.drop_foreign("fk_1") + diff.remove_index("idx_1") + self.assertEqual( + self.platform.compile_alter_sql(diff), + ["DROP INDEX idx_1", 'ALTER TABLE "users" DROP CONSTRAINT fk_1'], + ) + + def test_table_comment(self): + diff = TableDiff("users") + diff.add_column("x", "integer") + diff.add_comment("a table") + self.assertEqual( + self.platform.compile_alter_sql(diff), + [ + 'ALTER TABLE "users" ADD COLUMN "x" INTEGER NOT NULL', + "COMMENT ON TABLE \"users\" is 'a table'", + ], + ) + + +class TestPostgresPlatformHelpers(unittest.TestCase): + def setUp(self): + self.platform = PostgresPlatform() + + def test_compile_drop_table(self): + self.assertEqual(self.platform.compile_drop_table("users"), 'DROP TABLE "users" CASCADE') + + def test_compile_drop_table_if_exists(self): + self.assertEqual( + self.platform.compile_drop_table_if_exists("users"), + 'DROP TABLE IF EXISTS "users" CASCADE', + ) + + def test_compile_rename_table(self): + self.assertEqual( + self.platform.compile_rename_table("users", "clients"), + 'ALTER TABLE "users" RENAME TO "clients"', + ) + + def test_compile_truncate(self): + self.assertEqual(self.platform.compile_truncate("users"), 'TRUNCATE "users"') + + def test_compile_truncate_with_foreign_keys(self): + self.assertEqual( + self.platform.compile_truncate("users", foreign_keys=True), + [ + 'ALTER TABLE "users" DISABLE TRIGGER ALL', + 'TRUNCATE "users"', + 'ALTER TABLE "users" ENABLE TRIGGER ALL', + ], + ) + + def test_compile_table_exists_defaults_to_public(self): + self.assertEqual( + self.platform.compile_table_exists("users"), + "SELECT * from information_schema.tables where table_name='users' AND table_schema = 'public'", + ) + + def test_compile_table_exists_with_schema(self): + self.assertEqual( + self.platform.compile_table_exists("users", schema="app"), + "SELECT * from information_schema.tables where table_name='users' AND table_schema = 'app'", + ) + + def test_compile_column_exists(self): + self.assertEqual( + self.platform.compile_column_exists("users", "name"), + "SELECT column_name FROM information_schema.columns WHERE table_name='users' and column_name='name'", + ) + + def test_compile_get_all_tables(self): + self.assertEqual( + self.platform.compile_get_all_tables(database="app"), + "SELECT table_name FROM information_schema.tables WHERE table_schema = 'public' " + "AND table_catalog = 'app' AND table_type = 'BASE TABLE'", + ) + + def test_foreign_key_toggles_are_noops(self): + self.assertEqual(self.platform.enable_foreign_key_constraints(), "") + self.assertEqual(self.platform.disable_foreign_key_constraints(), "") + + def test_wrap_helpers(self): + self.assertEqual(self.platform.wrap_table("users"), '"users"') + self.assertEqual(self.platform.wrap_column("name"), '"name"')