codekingpro/portable-devtools

1# testing/suite/test_ddl.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: ignore-errors
8
9import random
10
11from . import testing
12from .. import config
13from .. import fixtures
14from .. import util
15from ..assertions import eq_
16from ..assertions import is_false
17from ..assertions import is_true
18from ..config import requirements
19from ..schema import Table
20from ... import CheckConstraint
21from ... import Column
22from ... import ForeignKeyConstraint
23from ... import Index
24from ... import inspect
25from ... import Integer
26from ... import schema
27from ... import String
28from ... import UniqueConstraint
29
30
31class TableDDLTest(fixtures.TestBase):
32    __sparse_driver_backend__ = True
33
34    def _simple_fixture(self, schema=None):
35        return Table(
36            "test_table",
37            self.metadata,
38            Column("id", Integer, primary_key=True, autoincrement=False),
39            Column("data", String(50)),
40            schema=schema,
41        )
42
43    def _underscore_fixture(self):
44        return Table(
45            "_test_table",
46            self.metadata,
47            Column("id", Integer, primary_key=True, autoincrement=False),
48            Column("_data", String(50)),
49        )
50
51    def _table_index_fixture(self, schema=None):
52        table = self._simple_fixture(schema=schema)
53        idx = Index("test_index", table.c.data)
54        return table, idx
55
56    def _simple_roundtrip(self, table):
57        with config.db.begin() as conn:
58            conn.execute(table.insert().values((1, "some data")))
59            result = conn.execute(table.select())
60            eq_(result.first(), (1, "some data"))
61
62    @requirements.create_table
63    @util.provide_metadata
64    def test_create_table(self):
65        table = self._simple_fixture()
66        table.create(config.db, checkfirst=False)
67        self._simple_roundtrip(table)
68
69    @requirements.create_table
70    @requirements.schemas
71    @util.provide_metadata
72    def test_create_table_schema(self):
73        table = self._simple_fixture(schema=config.test_schema)
74        table.create(config.db, checkfirst=False)
75        self._simple_roundtrip(table)
76
77    @requirements.drop_table
78    @util.provide_metadata
79    def test_drop_table(self):
80        table = self._simple_fixture()
81        table.create(config.db, checkfirst=False)
82        table.drop(config.db, checkfirst=False)
83
84    @requirements.create_table
85    @util.provide_metadata
86    def test_underscore_names(self):
87        table = self._underscore_fixture()
88        table.create(config.db, checkfirst=False)
89        self._simple_roundtrip(table)
90
91    @requirements.comment_reflection
92    @util.provide_metadata
93    def test_add_table_comment(self, connection):
94        table = self._simple_fixture()
95        table.create(connection, checkfirst=False)
96        table.comment = "a comment"
97        connection.execute(schema.SetTableComment(table))
98        eq_(
99            inspect(connection).get_table_comment("test_table"),
100            {"text": "a comment"},
101        )
102
103    @requirements.comment_reflection
104    @util.provide_metadata
105    def test_drop_table_comment(self, connection):
106        table = self._simple_fixture()
107        table.create(connection, checkfirst=False)
108        table.comment = "a comment"
109        connection.execute(schema.SetTableComment(table))
110        connection.execute(schema.DropTableComment(table))
111        eq_(
112            inspect(connection).get_table_comment("test_table"), {"text": None}
113        )
114
115    @requirements.table_ddl_if_exists
116    @util.provide_metadata
117    def test_create_table_if_not_exists(self, connection):
118        table = self._simple_fixture()
119
120        connection.execute(schema.CreateTable(table, if_not_exists=True))
121
122        is_true(inspect(connection).has_table("test_table"))
123        connection.execute(schema.CreateTable(table, if_not_exists=True))
124
125    @requirements.index_ddl_if_exists
126    @util.provide_metadata
127    def test_create_index_if_not_exists(self, connection):
128        table, idx = self._table_index_fixture()
129
130        connection.execute(schema.CreateTable(table, if_not_exists=True))
131        is_true(inspect(connection).has_table("test_table"))
132        is_false(
133            "test_index"
134            in [
135                ix["name"]
136                for ix in inspect(connection).get_indexes("test_table")
137            ]
138        )
139
140        connection.execute(schema.CreateIndex(idx, if_not_exists=True))
141
142        is_true(
143            "test_index"
144            in [
145                ix["name"]
146                for ix in inspect(connection).get_indexes("test_table")
147            ]
148        )
149
150        connection.execute(schema.CreateIndex(idx, if_not_exists=True))
151
152    @requirements.table_ddl_if_exists
153    @util.provide_metadata
154    def test_drop_table_if_exists(self, connection):
155        table = self._simple_fixture()
156
157        table.create(connection)
158
159        is_true(inspect(connection).has_table("test_table"))
160
161        connection.execute(schema.DropTable(table, if_exists=True))
162
163        is_false(inspect(connection).has_table("test_table"))
164
165        connection.execute(schema.DropTable(table, if_exists=True))
166
167    @requirements.index_ddl_if_exists
168    @util.provide_metadata
169    def test_drop_index_if_exists(self, connection):
170        table, idx = self._table_index_fixture()
171
172        table.create(connection)
173
174        is_true(
175            "test_index"
176            in [
177                ix["name"]
178                for ix in inspect(connection).get_indexes("test_table")
179            ]
180        )
181
182        connection.execute(schema.DropIndex(idx, if_exists=True))
183
184        is_false(
185            "test_index"
186            in [
187                ix["name"]
188                for ix in inspect(connection).get_indexes("test_table")
189            ]
190        )
191
192        connection.execute(schema.DropIndex(idx, if_exists=True))
193
194
195class FutureTableDDLTest(fixtures.FutureEngineMixin, TableDDLTest):
196    pass
197
198
199class LongNameBlowoutTest(fixtures.TestBase):
200    """test the creation of a variety of DDL structures and ensure
201    label length limits pass on backends
202
203    """
204
205    __sparse_driver_backend__ = True
206
207    def fk(self, metadata, connection):
208        convention = {
209            "fk": "foreign_key_%(table_name)s_"
210            "%(column_0_N_name)s_"
211            "%(referred_table_name)s_"
212            + (
213                "_".join(
214                    "".join(random.choice("abcdef") for j in range(20))
215                    for i in range(10)
216                )
217            ),
218        }
219        metadata.naming_convention = convention
220
221        Table(
222            "a_things_with_stuff",
223            metadata,
224            Column("id_long_column_name", Integer, primary_key=True),
225            test_needs_fk=True,
226        )
227
228        cons = ForeignKeyConstraint(
229            ["aid"], ["a_things_with_stuff.id_long_column_name"]
230        )
231        Table(
232            "b_related_things_of_value",
233            metadata,
234            Column(
235                "aid",
236            ),
237            cons,
238            test_needs_fk=True,
239        )
240        actual_name = cons.name
241
242        metadata.create_all(connection)
243
244        if testing.requires.foreign_key_constraint_name_reflection.enabled:
245            insp = inspect(connection)
246            fks = insp.get_foreign_keys("b_related_things_of_value")
247            reflected_name = fks[0]["name"]
248
249            return actual_name, reflected_name
250        else:
251            return actual_name, None
252
253    def pk(self, metadata, connection):
254        convention = {
255            "pk": "primary_key_%(table_name)s_"
256            "%(column_0_N_name)s"
257            + (
258                "_".join(
259                    "".join(random.choice("abcdef") for j in range(30))
260                    for i in range(10)
261                )
262            ),
263        }
264        metadata.naming_convention = convention
265
266        a = Table(
267            "a_things_with_stuff",
268            metadata,
269            Column("id_long_column_name", Integer, primary_key=True),
270            Column("id_another_long_name", Integer, primary_key=True),
271        )
272        cons = a.primary_key
273        actual_name = cons.name
274
275        metadata.create_all(connection)
276        insp = inspect(connection)
277        pk = insp.get_pk_constraint("a_things_with_stuff")
278        reflected_name = pk["name"]
279        return actual_name, reflected_name
280
281    def ix(self, metadata, connection):
282        convention = {
283            "ix": "index_%(table_name)s_"
284            "%(column_0_N_name)s"
285            + (
286                "_".join(
287                    "".join(random.choice("abcdef") for j in range(30))
288                    for i in range(10)
289                )
290            ),
291        }
292        metadata.naming_convention = convention
293
294        a = Table(
295            "a_things_with_stuff",
296            metadata,
297            Column("id_long_column_name", Integer, primary_key=True),
298            Column("id_another_long_name", Integer),
299        )
300        cons = Index(None, a.c.id_long_column_name, a.c.id_another_long_name)
301        actual_name = cons.name
302
303        metadata.create_all(connection)
304        insp = inspect(connection)
305        ix = insp.get_indexes("a_things_with_stuff")
306        reflected_name = ix[0]["name"]
307        return actual_name, reflected_name
308
309    def uq(self, metadata, connection):
310        convention = {
311            "uq": "unique_constraint_%(table_name)s_"
312            "%(column_0_N_name)s"
313            + (
314                "_".join(
315                    "".join(random.choice("abcdef") for j in range(30))
316                    for i in range(10)
317                )
318            ),
319        }
320        metadata.naming_convention = convention
321
322        cons = UniqueConstraint("id_long_column_name", "id_another_long_name")
323        Table(
324            "a_things_with_stuff",
325            metadata,
326            Column("id_long_column_name", Integer, primary_key=True),
327            Column("id_another_long_name", Integer),
328            cons,
329        )
330        actual_name = cons.name
331
332        metadata.create_all(connection)
333        insp = inspect(connection)
334        uq = insp.get_unique_constraints("a_things_with_stuff")
335        reflected_name = uq[0]["name"]
336        return actual_name, reflected_name
337
338    def ck(self, metadata, connection):
339        convention = {
340            "ck": "check_constraint_%(table_name)s"
341            + (
342                "_".join(
343                    "".join(random.choice("abcdef") for j in range(30))
344                    for i in range(10)
345                )
346            ),
347        }
348        metadata.naming_convention = convention
349
350        cons = CheckConstraint("some_long_column_name > 5")
351        Table(
352            "a_things_with_stuff",
353            metadata,
354            Column("id_long_column_name", Integer, primary_key=True),
355            Column("some_long_column_name", Integer),
356            cons,
357        )
358        actual_name = cons.name
359
360        metadata.create_all(connection)
361        insp = inspect(connection)
362        ck = insp.get_check_constraints("a_things_with_stuff")
363        reflected_name = ck[0]["name"]
364        return actual_name, reflected_name
365
366    @testing.combinations(
367        ("fk",),
368        ("pk",),
369        ("ix",),
370        ("ck", testing.requires.check_constraint_reflection.as_skips()),
371        ("uq", testing.requires.unique_constraint_reflection.as_skips()),
372        argnames="type_",
373    )
374    def test_long_convention_name(self, type_, metadata, connection):
375        actual_name, reflected_name = getattr(self, type_)(
376            metadata, connection
377        )
378
379        assert len(actual_name) > 255
380
381        if reflected_name is not None:
382            overlap = actual_name[0 : len(reflected_name)]
383            if len(overlap) < len(actual_name):
384                eq_(overlap[0:-5], reflected_name[0 : len(overlap) - 5])
385            else:
386                eq_(overlap, reflected_name)
387
388
389__all__ = ("TableDDLTest", "FutureTableDDLTest", "LongNameBlowoutTest")
390