summaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
authorJohn Whitman <jwhitman@galileo.io>2023-08-28 14:16:38 -0400
committerMariusz Felisiak <felisiak.mariusz@gmail.com>2023-08-30 12:46:15 +0200
commit68a8996bdfce2d191decd7b1c1a2b9fdea8e4b2f (patch)
tree0722940994234b53db11002f8317404e7b4426a4 /tests
parent3f8dbe267d35f0219277f0fe2d79915a4fb2b045 (diff)
Fixed #34799 -- Made MySQL introspection skip cross-database relations.
Diffstat (limited to 'tests')
-rw-r--r--tests/backends/mysql/test_introspection.py60
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}")