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
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
|
from unittest import skipUnless
from django.db import connection, connections
from django.test import TestCase
@skipUnless(connection.vendor == "mysql", "MySQL tests")
class ParsingTests(TestCase):
def test_parse_constraint_columns(self):
_parse_constraint_columns = connection.introspection._parse_constraint_columns
tests = (
("`height` >= 0", ["height"], ["height"]),
("`cost` BETWEEN 1 AND 10", ["cost"], ["cost"]),
("`ref1` > `ref2`", ["id", "ref1", "ref2"], ["ref1", "ref2"]),
(
"`start` IS NULL OR `end` IS NULL OR `start` < `end`",
["id", "start", "end"],
["start", "end"],
),
("JSON_VALID(`json_field`)", ["json_field"], ["json_field"]),
("CHAR_LENGTH(`name`) > 2", ["name"], ["name"]),
("lower(`ref1`) != 'test'", ["id", "owe", "ref1"], ["ref1"]),
("lower(`ref1`) != 'test'", ["id", "lower", "ref1"], ["ref1"]),
("`name` LIKE 'test%'", ["name"], ["name"]),
)
for check_clause, table_columns, expected_columns in tests:
with self.subTest(check_clause):
check_columns = _parse_constraint_columns(check_clause, table_columns)
self.assertEqual(list(check_columns), expected_columns)
@skipUnless(connection.vendor == "mysql", "MySQL tests")
class StorageEngineTests(TestCase):
databases = {"default", "other"}
def test_get_storage_engine(self):
table_name = "test_storage_engine"
create_sql = "CREATE TABLE %s (id INTEGER) ENGINE = %%s" % table_name
drop_sql = "DROP TABLE %s" % table_name
default_connection = connections["default"]
other_connection = connections["other"]
try:
with default_connection.cursor() as cursor:
cursor.execute(create_sql % "InnoDB")
self.assertEqual(
default_connection.introspection.get_storage_engine(
cursor, table_name
),
"InnoDB",
)
with other_connection.cursor() as cursor:
cursor.execute(create_sql % "MyISAM")
self.assertEqual(
other_connection.introspection.get_storage_engine(
cursor, table_name
),
"MyISAM",
)
finally:
with default_connection.cursor() as cursor:
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}")
|
