diff options
| author | John Whitman <jwhitman@galileo.io> | 2023-08-28 14:16:38 -0400 |
|---|---|---|
| committer | Mariusz Felisiak <felisiak.mariusz@gmail.com> | 2023-08-30 12:46:15 +0200 |
| commit | 68a8996bdfce2d191decd7b1c1a2b9fdea8e4b2f (patch) | |
| tree | 0722940994234b53db11002f8317404e7b4426a4 /tests | |
| parent | 3f8dbe267d35f0219277f0fe2d79915a4fb2b045 (diff) | |
Fixed #34799 -- Made MySQL introspection skip cross-database relations.
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/backends/mysql/test_introspection.py | 60 |
1 files changed, 60 insertions, 0 deletions
diff --git a/tests/backends/mysql/test_introspection.py b/tests/backends/mysql/test_introspection.py index c1247de232..a2e2938451 100644 --- a/tests/backends/mysql/test_introspection.py +++ b/tests/backends/mysql/test_introspection.py @@ -61,3 +61,63 @@ class StorageEngineTests(TestCase): cursor.execute(drop_sql) with other_connection.cursor() as cursor: cursor.execute(drop_sql) + + +@skipUnless(connection.vendor == "mysql", "MySQL specific SQL") +class TestCrossDatabaseRelations(TestCase): + databases = {"default", "other"} + + def test_omit_cross_database_relations(self): + default_connection = connections["default"] + other_connection = connections["other"] + main_table = "cross_schema_get_relations_main_table" + main_table_quoted = default_connection.ops.quote_name(main_table) + other_schema_quoted = other_connection.ops.quote_name( + other_connection.settings_dict["NAME"] + ) + rel_table = "cross_schema_get_relations_rel_table" + rel_table_quoted = other_connection.ops.quote_name(rel_table) + rel_column = "cross_schema_get_relations_rel_table_id" + rel_column_quoted = other_connection.ops.quote_name(rel_column) + try: + with other_connection.cursor() as other_cursor: + other_cursor.execute( + f""" + CREATE TABLE {rel_table_quoted} ( + id integer AUTO_INCREMENT, + PRIMARY KEY (id) + ) + """ + ) + with default_connection.cursor() as default_cursor: + # Create table in the default schema with a cross-database + # relation. + default_cursor.execute( + f""" + CREATE TABLE {main_table_quoted} ( + id integer AUTO_INCREMENT, + {rel_column_quoted} integer NOT NULL, + PRIMARY KEY (id), + FOREIGN KEY ({rel_column_quoted}) + REFERENCES {other_schema_quoted}.{rel_table_quoted}(id) + ) + """ + ) + relations = default_connection.introspection.get_relations( + default_cursor, main_table + ) + constraints = default_connection.introspection.get_constraints( + default_cursor, main_table + ) + self.assertEqual(len(relations), 0) + rel_column_fk_constraints = [ + spec + for name, spec in constraints.items() + if spec["columns"] == [rel_column] and spec["foreign_key"] is not None + ] + self.assertEqual(len(rel_column_fk_constraints), 0) + finally: + with default_connection.cursor() as default_cursor: + default_cursor.execute(f"DROP TABLE IF EXISTS {main_table_quoted}") + with other_connection.cursor() as other_cursor: + other_cursor.execute(f"DROP TABLE IF EXISTS {rel_table_quoted}") |
