codekingpro/portable-devtools
114k
1# testing/suite/test_reflection.py
2# Copyright (C) 2005-2024 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 operator
10import re
11
12import sqlalchemy as sa
13from .. import config
14from .. import engines
15from .. import eq_
16from .. import expect_raises
17from .. import expect_raises_message
18from .. import expect_warnings
19from .. import fixtures
20from .. import is_
21from ..provision import get_temp_table_name
22from ..provision import temp_table_keyword_args
23from ..schema import Column
24from ..schema import Table
25from ... import event
26from ... import ForeignKey
27from ... import func
28from ... import Identity
29from ... import inspect
30from ... import Integer
31from ... import MetaData
32from ... import String
33from ... import testing
34from ... import types as sql_types
35from ...engine import Inspector
36from ...engine import ObjectKind
37from ...engine import ObjectScope
38from ...exc import NoSuchTableError
39from ...exc import UnreflectableTableError
40from ...schema import DDL
41from ...schema import Index
42from ...sql.elements import quoted_name
43from ...sql.schema import BLANK_SCHEMA
44from ...testing import ComparesIndexes
45from ...testing import ComparesTables
46from ...testing import is_false
47from ...testing import is_true
48from ...testing import mock
49
50
51metadata, users = None, None
52
53
54class OneConnectionTablesTest(fixtures.TablesTest):
55 @classmethod
56 def setup_bind(cls):
57 # TODO: when temp tables are subject to server reset,
58 # this will also have to disable that server reset from
59 # happening
60 if config.requirements.independent_connections.enabled:
61 from sqlalchemy import pool
62
63 return engines.testing_engine(
64 options=dict(poolclass=pool.StaticPool, scope="class"),
65 )
66 else:
67 return config.db
68
69
70class HasTableTest(OneConnectionTablesTest):
71 __backend__ = True
72
73 @classmethod
74 def define_tables(cls, metadata):
75 Table(
76 "test_table",
77 metadata,
78 Column("id", Integer, primary_key=True),
79 Column("data", String(50)),
80 )
81 if testing.requires.schemas.enabled:
82 Table(
83 "test_table_s",
84 metadata,
85 Column("id", Integer, primary_key=True),
86 Column("data", String(50)),
87 schema=config.test_schema,
88 )
89
90 if testing.requires.view_reflection:
91 cls.define_views(metadata)
92 if testing.requires.has_temp_table.enabled:
93 cls.define_temp_tables(metadata)
94
95 @classmethod
96 def define_views(cls, metadata):
97 query = "CREATE VIEW vv AS SELECT id, data FROM test_table"
98
99 event.listen(metadata, "after_create", DDL(query))
100 event.listen(metadata, "before_drop", DDL("DROP VIEW vv"))
101
102 if testing.requires.schemas.enabled:
103 query = (
104 "CREATE VIEW %s.vv AS SELECT id, data FROM %s.test_table_s"
105 % (
106 config.test_schema,
107 config.test_schema,
108 )
109 )
110 event.listen(metadata, "after_create", DDL(query))
111 event.listen(
112 metadata,
113 "before_drop",
114 DDL("DROP VIEW %s.vv" % (config.test_schema)),
115 )
116
117 @classmethod
118 def temp_table_name(cls):
119 return get_temp_table_name(
120 config, config.db, f"user_tmp_{config.ident}"
121 )
122
123 @classmethod
124 def define_temp_tables(cls, metadata):
125 kw = temp_table_keyword_args(config, config.db)
126 table_name = cls.temp_table_name()
127 user_tmp = Table(
128 table_name,
129 metadata,
130 Column("id", sa.INT, primary_key=True),
131 Column("name", sa.VARCHAR(50)),
132 **kw,
133 )
134 if (
135 testing.requires.view_reflection.enabled
136 and testing.requires.temporary_views.enabled
137 ):
138 event.listen(
139 user_tmp,
140 "after_create",
141 DDL(
142 "create temporary view user_tmp_v as "
143 "select * from user_tmp_%s" % config.ident
144 ),
145 )
146 event.listen(user_tmp, "before_drop", DDL("drop view user_tmp_v"))
147
148 def test_has_table(self):
149 with config.db.begin() as conn:
150 is_true(config.db.dialect.has_table(conn, "test_table"))
151 is_false(config.db.dialect.has_table(conn, "test_table_s"))
152 is_false(config.db.dialect.has_table(conn, "nonexistent_table"))
153
154 def test_has_table_cache(self, metadata):
155 insp = inspect(config.db)
156 is_true(insp.has_table("test_table"))
157 nt = Table("new_table", metadata, Column("col", Integer))
158 is_false(insp.has_table("new_table"))
159 nt.create(config.db)
160 try:
161 is_false(insp.has_table("new_table"))
162 insp.clear_cache()
163 is_true(insp.has_table("new_table"))
164 finally:
165 nt.drop(config.db)
166
167 @testing.requires.schemas
168 def test_has_table_schema(self):
169 with config.db.begin() as conn:
170 is_false(
171 config.db.dialect.has_table(
172 conn, "test_table", schema=config.test_schema
173 )
174 )
175 is_true(
176 config.db.dialect.has_table(
177 conn, "test_table_s", schema=config.test_schema
178 )
179 )
180 is_false(
181 config.db.dialect.has_table(
182 conn, "nonexistent_table", schema=config.test_schema
183 )
184 )
185
186 @testing.requires.schemas
187 def test_has_table_nonexistent_schema(self):
188 with config.db.begin() as conn:
189 is_false(
190 config.db.dialect.has_table(
191 conn, "test_table", schema="nonexistent_schema"
192 )
193 )
194
195 @testing.requires.views
196 def test_has_table_view(self, connection):
197 insp = inspect(connection)
198 is_true(insp.has_table("vv"))
199
200 @testing.requires.has_temp_table
201 def test_has_table_temp_table(self, connection):
202 insp = inspect(connection)
203 temp_table_name = self.temp_table_name()
204 is_true(insp.has_table(temp_table_name))
205
206 @testing.requires.has_temp_table
207 @testing.requires.view_reflection
208 @testing.requires.temporary_views
209 def test_has_table_temp_view(self, connection):
210 insp = inspect(connection)
211 is_true(insp.has_table("user_tmp_v"))
212
213 @testing.requires.views
214 @testing.requires.schemas
215 def test_has_table_view_schema(self, connection):
216 insp = inspect(connection)
217 is_true(insp.has_table("vv", config.test_schema))
218
219
220class HasIndexTest(fixtures.TablesTest):
221 __backend__ = True
222
223 @classmethod
224 def define_tables(cls, metadata):
225 tt = Table(
226 "test_table",
227 metadata,
228 Column("id", Integer, primary_key=True),
229 Column("data", String(50)),
230 Column("data2", String(50)),
231 )
232 Index("my_idx", tt.c.data)
233
234 if testing.requires.schemas.enabled:
235 tt = Table(
236 "test_table",
237 metadata,
238 Column("id", Integer, primary_key=True),
239 Column("data", String(50)),
240 schema=config.test_schema,
241 )
242 Index("my_idx_s", tt.c.data)
243
244 kind = testing.combinations("dialect", "inspector", argnames="kind")
245
246 def _has_index(self, kind, conn):
247 if kind == "dialect":
248 return lambda *a, **k: config.db.dialect.has_index(conn, *a, **k)
249 else:
250 return inspect(conn).has_index
251
252 @kind
253 def test_has_index(self, kind, connection, metadata):
254 meth = self._has_index(kind, connection)
255 assert meth("test_table", "my_idx")
256 assert not meth("test_table", "my_idx_s")
257 assert not meth("nonexistent_table", "my_idx")
258 assert not meth("test_table", "nonexistent_idx")
259
260 assert not meth("test_table", "my_idx_2")
261 assert not meth("test_table_2", "my_idx_3")
262 idx = Index("my_idx_2", self.tables.test_table.c.data2)
263 tbl = Table(
264 "test_table_2",
265 metadata,
266 Column("foo", Integer),
267 Index("my_idx_3", "foo"),
268 )
269 idx.create(connection)
270 tbl.create(connection)
271 try:
272 if kind == "inspector":
273 assert not meth("test_table", "my_idx_2")
274 assert not meth("test_table_2", "my_idx_3")
275 meth.__self__.clear_cache()
276 assert meth("test_table", "my_idx_2") is True
277 assert meth("test_table_2", "my_idx_3") is True
278 finally:
279 tbl.drop(connection)
280 idx.drop(connection)
281
282 @testing.requires.schemas
283 @kind
284 def test_has_index_schema(self, kind, connection):
285 meth = self._has_index(kind, connection)
286 assert meth("test_table", "my_idx_s", schema=config.test_schema)
287 assert not meth("test_table", "my_idx", schema=config.test_schema)
288 assert not meth(
289 "nonexistent_table", "my_idx_s", schema=config.test_schema
290 )
291 assert not meth(
292 "test_table", "nonexistent_idx_s", schema=config.test_schema
293 )
294
295
296class BizarroCharacterFKResolutionTest(fixtures.TestBase):
297 """tests for #10275"""
298
299 __backend__ = True
300
301 @testing.combinations(
302 ("id",), ("(3)",), ("col%p",), ("[brack]",), argnames="columnname"
303 )
304 @testing.variation("use_composite", [True, False])
305 @testing.combinations(
306 ("plain",),
307 ("(2)",),
308 ("per % cent",),
309 ("[brackets]",),
310 argnames="tablename",
311 )
312 def test_fk_ref(
313 self, connection, metadata, use_composite, tablename, columnname
314 ):
315 tt = Table(
316 tablename,
317 metadata,
318 Column(columnname, Integer, key="id", primary_key=True),
319 test_needs_fk=True,
320 )
321 if use_composite:
322 tt.append_column(Column("id2", Integer, primary_key=True))
323
324 if use_composite:
325 Table(
326 "other",
327 metadata,
328 Column("id", Integer, primary_key=True),
329 Column("ref", Integer),
330 Column("ref2", Integer),
331 sa.ForeignKeyConstraint(["ref", "ref2"], [tt.c.id, tt.c.id2]),
332 test_needs_fk=True,
333 )
334 else:
335 Table(
336 "other",
337 metadata,
338 Column("id", Integer, primary_key=True),
339 Column("ref", ForeignKey(tt.c.id)),
340 test_needs_fk=True,
341 )
342
343 metadata.create_all(connection)
344
345 m2 = MetaData()
346
347 o2 = Table("other", m2, autoload_with=connection)
348 t1 = m2.tables[tablename]
349
350 assert o2.c.ref.references(t1.c[0])
351 if use_composite:
352 assert o2.c.ref2.references(t1.c[1])
353
354
355class QuotedNameArgumentTest(fixtures.TablesTest):
356 run_create_tables = "once"
357 __backend__ = True
358
359 @classmethod
360 def define_tables(cls, metadata):
361 Table(
362 "quote ' one",
363 metadata,
364 Column("id", Integer),
365 Column("name", String(50)),
366 Column("data", String(50)),
367 Column("related_id", Integer),
368 sa.PrimaryKeyConstraint("id", name="pk quote ' one"),
369 sa.Index("ix quote ' one", "name"),
370 sa.UniqueConstraint(
371 "data",
372 name="uq quote' one",
373 ),
374 sa.ForeignKeyConstraint(
375 ["id"], ["related.id"], name="fk quote ' one"
376 ),
377 sa.CheckConstraint("name != 'foo'", name="ck quote ' one"),
378 comment=r"""quote ' one comment""",
379 test_needs_fk=True,
380 )
381
382 if testing.requires.symbol_names_w_double_quote.enabled:
383 Table(
384 'quote " two',
385 metadata,
386 Column("id", Integer),
387 Column("name", String(50)),
388 Column("data", String(50)),
389 Column("related_id", Integer),
390 sa.PrimaryKeyConstraint("id", name='pk quote " two'),
391 sa.Index('ix quote " two', "name"),
392 sa.UniqueConstraint(
393 "data",
394 name='uq quote" two',
395 ),
396 sa.ForeignKeyConstraint(
397 ["id"], ["related.id"], name='fk quote " two'
398 ),
399 sa.CheckConstraint("name != 'foo'", name='ck quote " two '),
400 comment=r"""quote " two comment""",
401 test_needs_fk=True,
402 )
403
404 Table(
405 "related",
406 metadata,
407 Column("id", Integer, primary_key=True),
408 Column("related", Integer),
409 test_needs_fk=True,
410 )
411
412 if testing.requires.view_column_reflection.enabled:
413 if testing.requires.symbol_names_w_double_quote.enabled:
414 names = [
415 "quote ' one",
416 'quote " two',
417 ]
418 else:
419 names = [
420 "quote ' one",
421 ]
422 for name in names:
423 query = "CREATE VIEW %s AS SELECT * FROM %s" % (
424 config.db.dialect.identifier_preparer.quote(
425 "view %s" % name
426 ),
427 config.db.dialect.identifier_preparer.quote(name),
428 )
429
430 event.listen(metadata, "after_create", DDL(query))
431 event.listen(
432 metadata,
433 "before_drop",
434 DDL(
435 "DROP VIEW %s"
436 % config.db.dialect.identifier_preparer.quote(
437 "view %s" % name
438 )
439 ),
440 )
441
442 def quote_fixtures(fn):
443 return testing.combinations(
444 ("quote ' one",),
445 ('quote " two', testing.requires.symbol_names_w_double_quote),
446 )(fn)
447
448 @quote_fixtures
449 def test_get_table_options(self, name):
450 insp = inspect(config.db)
451
452 if testing.requires.reflect_table_options.enabled:
453 res = insp.get_table_options(name)
454 is_true(isinstance(res, dict))
455 else:
456 with expect_raises(NotImplementedError):
457 res = insp.get_table_options(name)
458
459 @quote_fixtures
460 @testing.requires.view_column_reflection
461 def test_get_view_definition(self, name):
462 insp = inspect(config.db)
463 assert insp.get_view_definition("view %s" % name)
464
465 @quote_fixtures
466 def test_get_columns(self, name):
467 insp = inspect(config.db)
468 assert insp.get_columns(name)
469
470 @quote_fixtures
471 def test_get_pk_constraint(self, name):
472 insp = inspect(config.db)
473 assert insp.get_pk_constraint(name)
474
475 @quote_fixtures
476 def test_get_foreign_keys(self, name):
477 insp = inspect(config.db)
478 assert insp.get_foreign_keys(name)
479
480 @quote_fixtures
481 def test_get_indexes(self, name):
482 insp = inspect(config.db)
483 assert insp.get_indexes(name)
484
485 @quote_fixtures
486 @testing.requires.unique_constraint_reflection
487 def test_get_unique_constraints(self, name):
488 insp = inspect(config.db)
489 assert insp.get_unique_constraints(name)
490
491 @quote_fixtures
492 @testing.requires.comment_reflection
493 def test_get_table_comment(self, name):
494 insp = inspect(config.db)
495 assert insp.get_table_comment(name)
496
497 @quote_fixtures
498 @testing.requires.check_constraint_reflection
499 def test_get_check_constraints(self, name):
500 insp = inspect(config.db)
501 assert insp.get_check_constraints(name)
502
503
504def _multi_combination(fn):
505 schema = testing.combinations(
506 None,
507 (
508 lambda: config.test_schema,
509 testing.requires.schemas,
510 ),
511 argnames="schema",
512 )
513 scope = testing.combinations(
514 ObjectScope.DEFAULT,
515 ObjectScope.TEMPORARY,
516 ObjectScope.ANY,
517 argnames="scope",
518 )
519 kind = testing.combinations(
520 ObjectKind.TABLE,
521 ObjectKind.VIEW,
522 ObjectKind.MATERIALIZED_VIEW,
523 ObjectKind.ANY,
524 ObjectKind.ANY_VIEW,
525 ObjectKind.TABLE | ObjectKind.VIEW,
526 ObjectKind.TABLE | ObjectKind.MATERIALIZED_VIEW,
527 argnames="kind",
528 )
529 filter_names = testing.combinations(True, False, argnames="use_filter")
530
531 return schema(scope(kind(filter_names(fn))))
532
533
534class ComponentReflectionTest(ComparesTables, OneConnectionTablesTest):
535 run_inserts = run_deletes = None
536
537 __backend__ = True
538
539 @classmethod
540 def define_tables(cls, metadata):
541 cls.define_reflected_tables(metadata, None)
542 if testing.requires.schemas.enabled:
543 cls.define_reflected_tables(metadata, testing.config.test_schema)
544
545 @classmethod
546 def define_reflected_tables(cls, metadata, schema):
547 if schema:
548 schema_prefix = schema + "."
549 else:
550 schema_prefix = ""
551
552 if testing.requires.self_referential_foreign_keys.enabled:
553 parent_id_args = (
554 ForeignKey(
555 "%susers.user_id" % schema_prefix, name="user_id_fk"
556 ),
557 )
558 else:
559 parent_id_args = ()
560 users = Table(
561 "users",
562 metadata,
563 Column("user_id", sa.INT, primary_key=True),
564 Column("test1", sa.CHAR(5), nullable=False),
565 Column("test2", sa.Float(), nullable=False),
566 Column("parent_user_id", sa.Integer, *parent_id_args),
567 sa.CheckConstraint(
568 "test2 > 0",
569 name="zz_test2_gt_zero",
570 comment="users check constraint",
571 ),
572 sa.CheckConstraint("test2 <= 1000"),
573 schema=schema,
574 test_needs_fk=True,
575 )
576
577 Table(
578 "dingalings",
579 metadata,
580 Column("dingaling_id", sa.Integer, primary_key=True),
581 Column(
582 "address_id",
583 sa.Integer,
584 ForeignKey(
585 "%semail_addresses.address_id" % schema_prefix,
586 name="zz_email_add_id_fg",
587 comment="di fk comment",
588 ),
589 ),
590 Column(
591 "id_user",
592 sa.Integer,
593 ForeignKey("%susers.user_id" % schema_prefix),
594 ),
595 Column("data", sa.String(30), unique=True),
596 sa.CheckConstraint(
597 "address_id > 0 AND address_id < 1000",
598 name="address_id_gt_zero",
599 ),
600 sa.UniqueConstraint(
601 "address_id",
602 "dingaling_id",
603 name="zz_dingalings_multiple",
604 comment="di unique comment",
605 ),
606 schema=schema,
607 test_needs_fk=True,
608 )
609 Table(
610 "email_addresses",
611 metadata,
612 Column("address_id", sa.Integer),
613 Column("remote_user_id", sa.Integer, ForeignKey(users.c.user_id)),
614 Column("email_address", sa.String(20), index=True),
615 sa.PrimaryKeyConstraint(
616 "address_id", name="email_ad_pk", comment="ea pk comment"
617 ),
618 schema=schema,
619 test_needs_fk=True,
620 )
621 Table(
622 "comment_test",
623 metadata,
624 Column("id", sa.Integer, primary_key=True, comment="id comment"),
625 Column("data", sa.String(20), comment="data % comment"),
626 Column(
627 "d2",
628 sa.String(20),
629 comment=r"""Comment types type speedily ' " \ '' Fun!""",
630 ),
631 Column("d3", sa.String(42), comment="Comment\nwith\rescapes"),
632 schema=schema,
633 comment=r"""the test % ' " \ table comment""",
634 )
635 Table(
636 "no_constraints",
637 metadata,
638 Column("data", sa.String(20)),
639 schema=schema,
640 comment="no\nconstraints\rhas\fescaped\vcomment",
641 )
642
643 if testing.requires.cross_schema_fk_reflection.enabled:
644 if schema is None:
645 Table(
646 "local_table",
647 metadata,
648 Column("id", sa.Integer, primary_key=True),
649 Column("data", sa.String(20)),
650 Column(
651 "remote_id",
652 ForeignKey(
653 "%s.remote_table_2.id" % testing.config.test_schema
654 ),
655 ),
656 test_needs_fk=True,
657 schema=config.db.dialect.default_schema_name,
658 )
659 else:
660 Table(
661 "remote_table",
662 metadata,
663 Column("id", sa.Integer, primary_key=True),
664 Column(
665 "local_id",
666 ForeignKey(
667 "%s.local_table.id"
668 % config.db.dialect.default_schema_name
669 ),
670 ),
671 Column("data", sa.String(20)),
672 schema=schema,
673 test_needs_fk=True,
674 )
675 Table(
676 "remote_table_2",
677 metadata,
678 Column("id", sa.Integer, primary_key=True),
679 Column("data", sa.String(20)),
680 schema=schema,
681 test_needs_fk=True,
682 )
683
684 if testing.requires.index_reflection.enabled:
685 Index("users_t_idx", users.c.test1, users.c.test2, unique=True)
686 Index(
687 "users_all_idx", users.c.user_id, users.c.test2, users.c.test1
688 )
689
690 if not schema:
691 # test_needs_fk is at the moment to force MySQL InnoDB
692 noncol_idx_test_nopk = Table(
693 "noncol_idx_test_nopk",
694 metadata,
695 Column("q", sa.String(5)),
696 test_needs_fk=True,
697 )
698
699 noncol_idx_test_pk = Table(
700 "noncol_idx_test_pk",
701 metadata,
702 Column("id", sa.Integer, primary_key=True),
703 Column("q", sa.String(5)),
704 test_needs_fk=True,
705 )
706
707 if (
708 testing.requires.indexes_with_ascdesc.enabled
709 and testing.requires.reflect_indexes_with_ascdesc.enabled
710 ):
711 Index("noncol_idx_nopk", noncol_idx_test_nopk.c.q.desc())
712 Index("noncol_idx_pk", noncol_idx_test_pk.c.q.desc())
713
714 if testing.requires.view_column_reflection.enabled:
715 cls.define_views(metadata, schema)
716 if not schema and testing.requires.temp_table_reflection.enabled:
717 cls.define_temp_tables(metadata)
718
719 @classmethod
720 def temp_table_name(cls):
721 return get_temp_table_name(
722 config, config.db, f"user_tmp_{config.ident}"
723 )
724
725 @classmethod
726 def define_temp_tables(cls, metadata):
727 kw = temp_table_keyword_args(config, config.db)
728 table_name = cls.temp_table_name()
729 user_tmp = Table(
730 table_name,
731 metadata,
732 Column("id", sa.INT, primary_key=True),
733 Column("name", sa.VARCHAR(50)),
734 Column("foo", sa.INT),
735 # disambiguate temp table unique constraint names. this is
736 # pretty arbitrary for a generic dialect however we are doing
737 # it to suit SQL Server which will produce name conflicts for
738 # unique constraints created against temp tables in different
739 # databases.
740 # https://www.arbinada.com/en/node/1645
741 sa.UniqueConstraint("name", name=f"user_tmp_uq_{config.ident}"),
742 sa.Index("user_tmp_ix", "foo"),
743 **kw,
744 )
745 if (
746 testing.requires.view_reflection.enabled
747 and testing.requires.temporary_views.enabled
748 ):
749 event.listen(
750 user_tmp,
751 "after_create",
752 DDL(
753 "create temporary view user_tmp_v as "
754 "select * from user_tmp_%s" % config.ident
755 ),
756 )
757 event.listen(user_tmp, "before_drop", DDL("drop view user_tmp_v"))
758
759 @classmethod
760 def define_views(cls, metadata, schema):
761 if testing.requires.materialized_views.enabled:
762 materialized = {"dingalings"}
763 else:
764 materialized = set()
765 for table_name in ("users", "email_addresses", "dingalings"):
766 fullname = table_name
767 if schema:
768 fullname = f"{schema}.{table_name}"
769 view_name = fullname + "_v"
770 prefix = "MATERIALIZED " if table_name in materialized else ""
771 query = (
772 f"CREATE {prefix}VIEW {view_name} AS SELECT * FROM {fullname}"
773 )
774
775 event.listen(metadata, "after_create", DDL(query))
776 if table_name in materialized:
777 index_name = "mat_index"
778 if schema and testing.against("oracle"):
779 index_name = f"{schema}.{index_name}"
780 idx = f"CREATE INDEX {index_name} ON {view_name}(data)"
781 event.listen(metadata, "after_create", DDL(idx))
782 event.listen(
783 metadata, "before_drop", DDL(f"DROP {prefix}VIEW {view_name}")
784 )
785
786 def _resolve_kind(self, kind, tables, views, materialized):
787 res = {}
788 if ObjectKind.TABLE in kind:
789 res.update(tables)
790 if ObjectKind.VIEW in kind:
791 res.update(views)
792 if ObjectKind.MATERIALIZED_VIEW in kind:
793 res.update(materialized)
794 return res
795
796 def _resolve_views(self, views, materialized):
797 if not testing.requires.view_column_reflection.enabled:
798 materialized.clear()
799 views.clear()
800 elif not testing.requires.materialized_views.enabled:
801 views.update(materialized)
802 materialized.clear()
803
804 def _resolve_names(self, schema, scope, filter_names, values):
805 scope_filter = lambda _: True # noqa: E731
806 if scope is ObjectScope.DEFAULT:
807 scope_filter = lambda k: "tmp" not in k[1] # noqa: E731
808 if scope is ObjectScope.TEMPORARY:
809 scope_filter = lambda k: "tmp" in k[1] # noqa: E731
810
811 removed = {
812 None: {"remote_table", "remote_table_2"},
813 testing.config.test_schema: {
814 "local_table",
815 "noncol_idx_test_nopk",
816 "noncol_idx_test_pk",
817 "user_tmp_v",
818 self.temp_table_name(),
819 },
820 }
821 if not testing.requires.cross_schema_fk_reflection.enabled:
822 removed[None].add("local_table")
823 removed[testing.config.test_schema].update(
824 ["remote_table", "remote_table_2"]
825 )
826 if not testing.requires.index_reflection.enabled:
827 removed[None].update(
828 ["noncol_idx_test_nopk", "noncol_idx_test_pk"]
829 )
830 if (
831 not testing.requires.temp_table_reflection.enabled
832 or not testing.requires.temp_table_names.enabled
833 ):
834 removed[None].update(["user_tmp_v", self.temp_table_name()])
835 if not testing.requires.temporary_views.enabled:
836 removed[None].update(["user_tmp_v"])
837
838 res = {
839 k: v
840 for k, v in values.items()
841 if scope_filter(k)
842 and k[1] not in removed[schema]
843 and (not filter_names or k[1] in filter_names)
844 }
845 return res
846
847 def exp_options(
848 self,
849 schema=None,
850 scope=ObjectScope.ANY,
851 kind=ObjectKind.ANY,
852 filter_names=None,
853 ):
854 materialized = {(schema, "dingalings_v"): mock.ANY}
855 views = {
856 (schema, "email_addresses_v"): mock.ANY,
857 (schema, "users_v"): mock.ANY,
858 (schema, "user_tmp_v"): mock.ANY,
859 }
860 self._resolve_views(views, materialized)
861 tables = {
862 (schema, "users"): mock.ANY,
863 (schema, "dingalings"): mock.ANY,
864 (schema, "email_addresses"): mock.ANY,
865 (schema, "comment_test"): mock.ANY,
866 (schema, "no_constraints"): mock.ANY,
867 (schema, "local_table"): mock.ANY,
868 (schema, "remote_table"): mock.ANY,
869 (schema, "remote_table_2"): mock.ANY,
870 (schema, "noncol_idx_test_nopk"): mock.ANY,
871 (schema, "noncol_idx_test_pk"): mock.ANY,
872 (schema, self.temp_table_name()): mock.ANY,
873 }
874 res = self._resolve_kind(kind, tables, views, materialized)
875 res = self._resolve_names(schema, scope, filter_names, res)
876 return res
877
878 def exp_comments(
879 self,
880 schema=None,
881 scope=ObjectScope.ANY,
882 kind=ObjectKind.ANY,
883 filter_names=None,
884 ):
885 empty = {"text": None}
886 materialized = {(schema, "dingalings_v"): empty}
887 views = {
888 (schema, "email_addresses_v"): empty,
889 (schema, "users_v"): empty,
890 (schema, "user_tmp_v"): empty,
891 }
892 self._resolve_views(views, materialized)
893 tables = {
894 (schema, "users"): empty,
895 (schema, "dingalings"): empty,
896 (schema, "email_addresses"): empty,
897 (schema, "comment_test"): {
898 "text": r"""the test % ' " \ table comment"""
899 },
900 (schema, "no_constraints"): {
901 "text": "no\nconstraints\rhas\fescaped\vcomment"
902 },
903 (schema, "local_table"): empty,
904 (schema, "remote_table"): empty,
905 (schema, "remote_table_2"): empty,
906 (schema, "noncol_idx_test_nopk"): empty,
907 (schema, "noncol_idx_test_pk"): empty,
908 (schema, self.temp_table_name()): empty,
909 }
910 res = self._resolve_kind(kind, tables, views, materialized)
911 res = self._resolve_names(schema, scope, filter_names, res)
912 return res
913
914 def exp_columns(
915 self,
916 schema=None,
917 scope=ObjectScope.ANY,
918 kind=ObjectKind.ANY,
919 filter_names=None,
920 ):
921 def col(
922 name, auto=False, default=mock.ANY, comment=None, nullable=True
923 ):
924 res = {
925 "name": name,
926 "autoincrement": auto,
927 "type": mock.ANY,
928 "default": default,
929 "comment": comment,
930 "nullable": nullable,
931 }
932 if auto == "omit":
933 res.pop("autoincrement")
934 return res
935
936 def pk(name, **kw):
937 kw = {"auto": True, "default": mock.ANY, "nullable": False, **kw}
938 return col(name, **kw)
939
940 materialized = {
941 (schema, "dingalings_v"): [
942 col("dingaling_id", auto="omit", nullable=mock.ANY),
943 col("address_id"),
944 col("id_user"),
945 col("data"),
946 ]
947 }
948 views = {
949 (schema, "email_addresses_v"): [
950 col("address_id", auto="omit", nullable=mock.ANY),
951 col("remote_user_id"),
952 col("email_address"),
953 ],
954 (schema, "users_v"): [
955 col("user_id", auto="omit", nullable=mock.ANY),
956 col("test1", nullable=mock.ANY),
957 col("test2", nullable=mock.ANY),
958 col("parent_user_id"),
959 ],
960 (schema, "user_tmp_v"): [
961 col("id", auto="omit", nullable=mock.ANY),
962 col("name"),
963 col("foo"),
964 ],
965 }
966 self._resolve_views(views, materialized)
967 tables = {
968 (schema, "users"): [
969 pk("user_id"),
970 col("test1", nullable=False),
971 col("test2", nullable=False),
972 col("parent_user_id"),
973 ],
974 (schema, "dingalings"): [
975 pk("dingaling_id"),
976 col("address_id"),
977 col("id_user"),
978 col("data"),
979 ],
980 (schema, "email_addresses"): [
981 pk("address_id"),
982 col("remote_user_id"),
983 col("email_address"),
984 ],
985 (schema, "comment_test"): [
986 pk("id", comment="id comment"),
987 col("data", comment="data % comment"),
988 col(
989 "d2",
990 comment=r"""Comment types type speedily ' " \ '' Fun!""",
991 ),
992 col("d3", comment="Comment\nwith\rescapes"),
993 ],
994 (schema, "no_constraints"): [col("data")],
995 (schema, "local_table"): [pk("id"), col("data"), col("remote_id")],
996 (schema, "remote_table"): [pk("id"), col("local_id"), col("data")],
997 (schema, "remote_table_2"): [pk("id"), col("data")],
998 (schema, "noncol_idx_test_nopk"): [col("q")],
999 (schema, "noncol_idx_test_pk"): [pk("id"), col("q")],
1000 (schema, self.temp_table_name()): [
1001 pk("id"),
1002 col("name"),
1003 col("foo"),
1004 ],
1005 }
1006 res = self._resolve_kind(kind, tables, views, materialized)
1007 res = self._resolve_names(schema, scope, filter_names, res)
1008 return res
1009
1010 @property
1011 def _required_column_keys(self):
1012 return {"name", "type", "nullable", "default"}
1013
1014 def exp_pks(
1015 self,
1016 schema=None,
1017 scope=ObjectScope.ANY,
1018 kind=ObjectKind.ANY,
1019 filter_names=None,
1020 ):
1021 def pk(*cols, name=mock.ANY, comment=None):
1022 return {
1023 "constrained_columns": list(cols),
1024 "name": name,
1025 "comment": comment,
1026 }
1027
1028 empty = pk(name=None)
1029 if testing.requires.materialized_views_reflect_pk.enabled:
1030 materialized = {(schema, "dingalings_v"): pk("dingaling_id")}
1031 else:
1032 materialized = {(schema, "dingalings_v"): empty}
1033 views = {
1034 (schema, "email_addresses_v"): empty,
1035 (schema, "users_v"): empty,
1036 (schema, "user_tmp_v"): empty,
1037 }
1038 self._resolve_views(views, materialized)
1039 tables = {
1040 (schema, "users"): pk("user_id"),
1041 (schema, "dingalings"): pk("dingaling_id"),
1042 (schema, "email_addresses"): pk(
1043 "address_id", name="email_ad_pk", comment="ea pk comment"
1044 ),
1045 (schema, "comment_test"): pk("id"),
1046 (schema, "no_constraints"): empty,
1047 (schema, "local_table"): pk("id"),
1048 (schema, "remote_table"): pk("id"),
1049 (schema, "remote_table_2"): pk("id"),
1050 (schema, "noncol_idx_test_nopk"): empty,
1051 (schema, "noncol_idx_test_pk"): pk("id"),
1052 (schema, self.temp_table_name()): pk("id"),
1053 }
1054 if not testing.requires.reflects_pk_names.enabled:
1055 for val in tables.values():
1056 if val["name"] is not None:
1057 val["name"] = mock.ANY
1058 res = self._resolve_kind(kind, tables, views, materialized)
1059 res = self._resolve_names(schema, scope, filter_names, res)
1060 return res
1061
1062 @property
1063 def _required_pk_keys(self):
1064 return {"name", "constrained_columns"}
1065
1066 def exp_fks(
1067 self,
1068 schema=None,
1069 scope=ObjectScope.ANY,
1070 kind=ObjectKind.ANY,
1071 filter_names=None,
1072 ):
1073 class tt:
1074 def __eq__(self, other):
1075 return (
1076 other is None
1077 or config.db.dialect.default_schema_name == other
1078 )
1079
1080 def fk(
1081 cols,
1082 ref_col,
1083 ref_table,
1084 ref_schema=schema,
1085 name=mock.ANY,
1086 comment=None,
1087 ):
1088 return {
1089 "constrained_columns": cols,
1090 "referred_columns": ref_col,
1091 "name": name,
1092 "options": mock.ANY,
1093 "referred_schema": (
1094 ref_schema if ref_schema is not None else tt()
1095 ),
1096 "referred_table": ref_table,
1097 "comment": comment,
1098 }
1099
1100 materialized = {(schema, "dingalings_v"): []}
1101 views = {
1102 (schema, "email_addresses_v"): [],
1103 (schema, "users_v"): [],
1104 (schema, "user_tmp_v"): [],
1105 }
1106 self._resolve_views(views, materialized)
1107 tables = {
1108 (schema, "users"): [
1109 fk(["parent_user_id"], ["user_id"], "users", name="user_id_fk")
1110 ],
1111 (schema, "dingalings"): [
1112 fk(["id_user"], ["user_id"], "users"),
1113 fk(
1114 ["address_id"],
1115 ["address_id"],
1116 "email_addresses",
1117 name="zz_email_add_id_fg",
1118 comment="di fk comment",
1119 ),
1120 ],
1121 (schema, "email_addresses"): [
1122 fk(["remote_user_id"], ["user_id"], "users")
1123 ],
1124 (schema, "comment_test"): [],
1125 (schema, "no_constraints"): [],
1126 (schema, "local_table"): [
1127 fk(
1128 ["remote_id"],
1129 ["id"],
1130 "remote_table_2",
1131 ref_schema=config.test_schema,
1132 )
1133 ],
1134 (schema, "remote_table"): [
1135 fk(["local_id"], ["id"], "local_table", ref_schema=None)
1136 ],
1137 (schema, "remote_table_2"): [],
1138 (schema, "noncol_idx_test_nopk"): [],
1139 (schema, "noncol_idx_test_pk"): [],
1140 (schema, self.temp_table_name()): [],
1141 }
1142 if not testing.requires.self_referential_foreign_keys.enabled:
1143 tables[(schema, "users")].clear()
1144 if not testing.requires.named_constraints.enabled:
1145 for vals in tables.values():
1146 for val in vals:
1147 if val["name"] is not mock.ANY:
1148 val["name"] = mock.ANY
1149
1150 res = self._resolve_kind(kind, tables, views, materialized)
1151 res = self._resolve_names(schema, scope, filter_names, res)
1152 return res
1153
1154 @property
1155 def _required_fk_keys(self):
1156 return {
1157 "name",
1158 "constrained_columns",
1159 "referred_schema",
1160 "referred_table",
1161 "referred_columns",
1162 }
1163
1164 def exp_indexes(
1165 self,
1166 schema=None,
1167 scope=ObjectScope.ANY,
1168 kind=ObjectKind.ANY,
1169 filter_names=None,
1170 ):
1171 def idx(
1172 *cols,
1173 name,
1174 unique=False,
1175 column_sorting=None,
1176 duplicates=False,
1177 fk=False,
1178 ):
1179 fk_req = testing.requires.foreign_keys_reflect_as_index
1180 dup_req = testing.requires.unique_constraints_reflect_as_index
1181 sorting_expression = (
1182 testing.requires.reflect_indexes_with_ascdesc_as_expression
1183 )
1184
1185 if (fk and not fk_req.enabled) or (
1186 duplicates and not dup_req.enabled
1187 ):
1188 return ()
1189 res = {
1190 "unique": unique,
1191 "column_names": list(cols),
1192 "name": name,
1193 "dialect_options": mock.ANY,
1194 "include_columns": [],
1195 }
1196 if column_sorting:
1197 res["column_sorting"] = column_sorting
1198 if sorting_expression.enabled:
1199 res["expressions"] = orig = res["column_names"]
1200 res["column_names"] = [