codekingpro/portable-devtools

1# testing/suite/test_sequence.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
9from .. import config
10from .. import fixtures
11from ..assertions import eq_
12from ..assertions import is_true
13from ..config import requirements
14from ..provision import normalize_sequence
15from ..schema import Column
16from ..schema import Table
17from ... import inspect
18from ... import Integer
19from ... import MetaData
20from ... import Sequence
21from ... import String
22from ... import testing
23
24
25class SequenceTest(fixtures.TablesTest):
26    __requires__ = ("sequences",)
27    __sparse_driver_backend__ = True
28
29    run_create_tables = "each"
30
31    @classmethod
32    def define_tables(cls, metadata):
33        Table(
34            "seq_pk",
35            metadata,
36            Column(
37                "id",
38                Integer,
39                normalize_sequence(config, Sequence("tab_id_seq")),
40                primary_key=True,
41            ),
42            Column("data", String(50)),
43        )
44
45        Table(
46            "seq_opt_pk",
47            metadata,
48            Column(
49                "id",
50                Integer,
51                normalize_sequence(
52                    config,
53                    Sequence("tab_id_seq", data_type=Integer, optional=True),
54                ),
55                primary_key=True,
56            ),
57            Column("data", String(50)),
58        )
59
60        Table(
61            "seq_no_returning",
62            metadata,
63            Column(
64                "id",
65                Integer,
66                normalize_sequence(config, Sequence("noret_id_seq")),
67                primary_key=True,
68            ),
69            Column("data", String(50)),
70            implicit_returning=False,
71        )
72
73        if testing.requires.schemas.enabled:
74            Table(
75                "seq_no_returning_sch",
76                metadata,
77                Column(
78                    "id",
79                    Integer,
80                    normalize_sequence(
81                        config,
82                        Sequence(
83                            "noret_sch_id_seq", schema=config.test_schema
84                        ),
85                    ),
86                    primary_key=True,
87                ),
88                Column("data", String(50)),
89                implicit_returning=False,
90                schema=config.test_schema,
91            )
92
93    def test_insert_roundtrip(self, connection):
94        connection.execute(self.tables.seq_pk.insert(), dict(data="some data"))
95        self._assert_round_trip(self.tables.seq_pk, connection)
96
97    def test_insert_lastrowid(self, connection):
98        r = connection.execute(
99            self.tables.seq_pk.insert(), dict(data="some data")
100        )
101        eq_(
102            r.inserted_primary_key, (testing.db.dialect.default_sequence_base,)
103        )
104
105    def test_nextval_direct(self, connection):
106        r = connection.scalar(self.tables.seq_pk.c.id.default)
107        eq_(r, testing.db.dialect.default_sequence_base)
108
109    @requirements.sequences_optional
110    def test_optional_seq(self, connection):
111        r = connection.execute(
112            self.tables.seq_opt_pk.insert(), dict(data="some data")
113        )
114        eq_(r.inserted_primary_key, (1,))
115
116    def _assert_round_trip(self, table, conn):
117        row = conn.execute(table.select()).first()
118        eq_(row, (testing.db.dialect.default_sequence_base, "some data"))
119
120    def test_insert_roundtrip_no_implicit_returning(self, connection):
121        connection.execute(
122            self.tables.seq_no_returning.insert(), dict(data="some data")
123        )
124        self._assert_round_trip(self.tables.seq_no_returning, connection)
125
126    @testing.combinations((True,), (False,), argnames="implicit_returning")
127    @testing.requires.schemas
128    def test_insert_roundtrip_translate(self, connection, implicit_returning):
129        seq_no_returning = Table(
130            "seq_no_returning_sch",
131            MetaData(),
132            Column(
133                "id",
134                Integer,
135                normalize_sequence(
136                    config, Sequence("noret_sch_id_seq", schema="alt_schema")
137                ),
138                primary_key=True,
139            ),
140            Column("data", String(50)),
141            implicit_returning=implicit_returning,
142            schema="alt_schema",
143        )
144
145        connection = connection.execution_options(
146            schema_translate_map={"alt_schema": config.test_schema}
147        )
148        connection.execute(seq_no_returning.insert(), dict(data="some data"))
149        self._assert_round_trip(seq_no_returning, connection)
150
151    @testing.requires.schemas
152    def test_nextval_direct_schema_translate(self, connection):
153        seq = normalize_sequence(
154            config, Sequence("noret_sch_id_seq", schema="alt_schema")
155        )
156        connection = connection.execution_options(
157            schema_translate_map={"alt_schema": config.test_schema}
158        )
159
160        r = connection.scalar(seq)
161        eq_(r, testing.db.dialect.default_sequence_base)
162
163
164class SequenceCompilerTest(testing.AssertsCompiledSQL, fixtures.TestBase):
165    __requires__ = ("sequences",)
166    __sparse_driver_backend__ = True
167
168    def test_literal_binds_inline_compile(self, connection):
169        table = Table(
170            "x",
171            MetaData(),
172            Column(
173                "y", Integer, normalize_sequence(config, Sequence("y_seq"))
174            ),
175            Column("q", Integer),
176        )
177
178        stmt = table.insert().values(q=5)
179
180        seq_nextval = connection.dialect.statement_compiler(
181            statement=None, dialect=connection.dialect
182        ).visit_sequence(normalize_sequence(config, Sequence("y_seq")))
183        self.assert_compile(
184            stmt,
185            "INSERT INTO x (y, q) VALUES (%s, 5)" % (seq_nextval,),
186            literal_binds=True,
187            dialect=connection.dialect,
188        )
189
190
191class HasSequenceTest(fixtures.TablesTest):
192    run_deletes = None
193
194    __requires__ = ("sequences",)
195    __sparse_driver_backend__ = True
196
197    @classmethod
198    def define_tables(cls, metadata):
199        normalize_sequence(config, Sequence("user_id_seq", metadata=metadata))
200        normalize_sequence(
201            config,
202            Sequence(
203                "other_seq",
204                metadata=metadata,
205                nomaxvalue=True,
206                nominvalue=True,
207            ),
208        )
209        if testing.requires.schemas.enabled:
210            normalize_sequence(
211                config,
212                Sequence(
213                    "user_id_seq", schema=config.test_schema, metadata=metadata
214                ),
215            )
216            normalize_sequence(
217                config,
218                Sequence(
219                    "schema_seq", schema=config.test_schema, metadata=metadata
220                ),
221            )
222        Table(
223            "user_id_table",
224            metadata,
225            Column("id", Integer, primary_key=True),
226        )
227
228    def test_has_sequence(self, connection):
229        eq_(inspect(connection).has_sequence("user_id_seq"), True)
230
231    def test_has_sequence_cache(self, connection, metadata):
232        insp = inspect(connection)
233        eq_(insp.has_sequence("user_id_seq"), True)
234        ss = normalize_sequence(config, Sequence("new_seq", metadata=metadata))
235        eq_(insp.has_sequence("new_seq"), False)
236        ss.create(connection)
237        try:
238            eq_(insp.has_sequence("new_seq"), False)
239            insp.clear_cache()
240            eq_(insp.has_sequence("new_seq"), True)
241        finally:
242            ss.drop(connection)
243
244    def test_has_sequence_other_object(self, connection):
245        eq_(inspect(connection).has_sequence("user_id_table"), False)
246
247    @testing.requires.schemas
248    def test_has_sequence_schema(self, connection):
249        eq_(
250            inspect(connection).has_sequence(
251                "user_id_seq", schema=config.test_schema
252            ),
253            True,
254        )
255
256    def test_has_sequence_neg(self, connection):
257        eq_(inspect(connection).has_sequence("some_sequence"), False)
258
259    @testing.requires.schemas
260    def test_has_sequence_schemas_neg(self, connection):
261        eq_(
262            inspect(connection).has_sequence(
263                "some_sequence", schema=config.test_schema
264            ),
265            False,
266        )
267
268    @testing.requires.schemas
269    def test_has_sequence_default_not_in_remote(self, connection):
270        eq_(
271            inspect(connection).has_sequence(
272                "other_sequence", schema=config.test_schema
273            ),
274            False,
275        )
276
277    @testing.requires.schemas
278    def test_has_sequence_remote_not_in_default(self, connection):
279        eq_(inspect(connection).has_sequence("schema_seq"), False)
280
281    def test_get_sequence_names(self, connection):
282        exp = {"other_seq", "user_id_seq"}
283
284        res = set(inspect(connection).get_sequence_names())
285        is_true(res.intersection(exp) == exp)
286        is_true("schema_seq" not in res)
287
288    @testing.requires.schemas
289    def test_get_sequence_names_no_sequence_schema(self, connection):
290        eq_(
291            inspect(connection).get_sequence_names(
292                schema=config.test_schema_2
293            ),
294            [],
295        )
296
297    @testing.requires.schemas
298    def test_get_sequence_names_sequences_schema(self, connection):
299        eq_(
300            sorted(
301                inspect(connection).get_sequence_names(
302                    schema=config.test_schema
303                )
304            ),
305            ["schema_seq", "user_id_seq"],
306        )
307
308
309class HasSequenceTestEmpty(fixtures.TestBase):
310    __requires__ = ("sequences",)
311    __sparse_driver_backend__ = True
312
313    def test_get_sequence_names_no_sequence(self, connection):
314        eq_(
315            inspect(connection).get_sequence_names(),
316            [],
317        )
318