# testing/suite/test_reflection.py

# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors

# <see AUTHORS file>

#

# This module is part of SQLAlchemy and is released under

# the MIT License: https://www.opensource.org/licenses/mit-license.php

# mypy: ignore-errors

import contextlib

import operator

import re

import sqlalchemy as sa

from .. import config

from .. import engines

from .. import eq_

from .. import eq_regex

from .. import expect_raises

from .. import expect_raises_message

from .. import expect_warnings

from .. import fixtures

from .. import is_

from ..provision import get_temp_table_name

from ..provision import temp_table_keyword_args

from ..schema import Column

from ..schema import Table

from ... import Boolean

from ... import DateTime

from ... import event

from ... import ForeignKey

from ... import func

from ... import Identity

from ... import inspect

from ... import Integer

from ... import MetaData

from ... import String

from ... import testing

from ... import types as sql_types

from ...engine import Inspector

from ...engine import ObjectKind

from ...engine import ObjectScope

from ...exc import NoSuchTableError

from ...exc import UnreflectableTableError

from ...schema import DDL

from ...schema import Index

from ...sql.elements import quoted_name

from ...sql.schema import BLANK_SCHEMA

from ...testing import ComparesIndexes

from ...testing import ComparesTables

from ...testing import is_false

from ...testing import is_none

from ...testing import is_true

from ...testing import mock

metadata, users = None, None

class OneConnectionTablesTest(fixtures.TablesTest):

@classmethod

def setup_bind(cls):

# TODO: when temp tables are subject to server reset,

# this will also have to disable that server reset from

# happening

if config.requirements.independent_connections.enabled:

from sqlalchemy import pool

return engines.testing_engine(

options=dict(poolclass=pool.StaticPool, scope="class"),

)

else:

return config.db

class HasTableTest(OneConnectionTablesTest):

__sparse_driver_backend__ = True

run_deletes = None

@classmethod

def define_tables(cls, metadata):

Table(

"test_table",

metadata,

Column("id", Integer, primary_key=True),

Column("data", String(50)),

)

if testing.requires.schemas.enabled:

Table(

"test_table_s",

metadata,

Column("id", Integer, primary_key=True),

Column("data", String(50)),

schema=config.test_schema,

)

if testing.requires.view_reflection:

cls.define_views(metadata)

if testing.requires.has_temp_table.enabled:

cls.define_temp_tables(metadata)

@classmethod

def define_views(cls, metadata):

test_table = metadata.tables["test_table"]

sa.CreateView(

sa.select(test_table.c.id, test_table.c.data),

"vv",

metadata=metadata,

)

if testing.requires.schemas.enabled:

test_table_s = metadata.tables[

f"{config.test_schema}.test_table_s"

]

sa.CreateView(

sa.select(test_table_s.c.id, test_table_s.c.data),

"vv",

metadata=metadata,

schema=config.test_schema,

)

@classmethod

def temp_table_name(cls):

return get_temp_table_name(

config, config.db, f"user_tmp_{config.ident}"

)

@classmethod

def define_temp_tables(cls, metadata):

kw = temp_table_keyword_args(config, config.db)

table_name = cls.temp_table_name()

user_tmp = Table(

table_name,

metadata,

Column("id", sa.INT, primary_key=True),

Column("name", sa.VARCHAR(50)),

**kw,

)

if (

testing.requires.view_reflection.enabled

and testing.requires.temporary_views.enabled

):

event.listen(

user_tmp,

"after_create",

DDL(

"create temporary view user_tmp_v as "

"select * from user_tmp_%s" % config.ident

),

)

event.listen(user_tmp, "before_drop", DDL("drop view user_tmp_v"))

def test_has_table(self):

with config.db.begin() as conn:

is_true(config.db.dialect.has_table(conn, "test_table"))

is_false(config.db.dialect.has_table(conn, "test_table_s"))

is_false(config.db.dialect.has_table(conn, "nonexistent_table"))

def test_has_table_cache(self, metadata):

insp = inspect(config.db)

is_true(insp.has_table("test_table"))

nt = Table("new_table", metadata, Column("col", Integer))

is_false(insp.has_table("new_table"))

nt.create(config.db)

try:

is_false(insp.has_table("new_table"))

insp.clear_cache()

is_true(insp.has_table("new_table"))

finally:

nt.drop(config.db)

@testing.requires.schemas

def test_has_table_schema(self):

with config.db.begin() as conn:

is_false(

config.db.dialect.has_table(

conn, "test_table", schema=config.test_schema

)

)

is_true(

config.db.dialect.has_table(

conn, "test_table_s", schema=config.test_schema

)

)

is_false(

config.db.dialect.has_table(

conn, "nonexistent_table", schema=config.test_schema

)

)

@testing.requires.schemas

def test_has_table_nonexistent_schema(self):

with config.db.begin() as conn:

is_false(

config.db.dialect.has_table(

conn, "test_table", schema="nonexistent_schema"

)

)

@testing.requires.views

def test_has_table_view(self, connection):

insp = inspect(connection)

is_true(insp.has_table("vv"))

@testing.requires.has_temp_table

def test_has_table_temp_table(self, connection):

insp = inspect(connection)

temp_table_name = self.temp_table_name()

is_true(insp.has_table(temp_table_name))

@testing.requires.has_temp_table

@testing.requires.view_reflection

@testing.requires.temporary_views

def test_has_table_temp_view(self, connection):

insp = inspect(connection)

is_true(insp.has_table("user_tmp_v"))

@testing.requires.views

@testing.requires.schemas

def test_has_table_view_schema(self, connection):

insp = inspect(connection)

is_true(insp.has_table("vv", config.test_schema))

class HasIndexTest(fixtures.TablesTest):

__sparse_driver_backend__ = True

__requires__ = ("index_reflection",)

@classmethod

def define_tables(cls, metadata):

tt = Table(

"test_table",

metadata,

Column("id", Integer, primary_key=True),

Column("data", String(50)),

Column("data2", String(50)),

)

Index("my_idx", tt.c.data)

if testing.requires.schemas.enabled:

tt = Table(

"test_table",

metadata,

Column("id", Integer, primary_key=True),

Column("data", String(50)),

schema=config.test_schema,

)

Index("my_idx_s", tt.c.data)

kind = testing.combinations("dialect", "inspector", argnames="kind")

def _has_index(self, kind, conn):

if kind == "dialect":

return lambda *a, **k: config.db.dialect.has_index(conn, *a, **k)

else:

return inspect(conn).has_index

@kind

def test_has_index(self, kind, connection, metadata):

meth = self._has_index(kind, connection)

assert meth("test_table", "my_idx")

assert not meth("test_table", "my_idx_s")

assert not meth("nonexistent_table", "my_idx")

assert not meth("test_table", "nonexistent_idx")

assert not meth("test_table", "my_idx_2")

assert not meth("test_table_2", "my_idx_3")

idx = Index("my_idx_2", self.tables.test_table.c.data2)

tbl = Table(

"test_table_2",

metadata,

Column("foo", Integer),

Index("my_idx_3", "foo"),

)

idx.create(connection)

tbl.create(connection)

try:

if kind == "inspector":

assert not meth("test_table", "my_idx_2")

assert not meth("test_table_2", "my_idx_3")

meth.__self__.clear_cache()

assert meth("test_table", "my_idx_2") is True

assert meth("test_table_2", "my_idx_3") is True

finally:

tbl.drop(connection)

idx.drop(connection)

@testing.requires.schemas

@kind

def test_has_index_schema(self, kind, connection):

meth = self._has_index(kind, connection)

assert meth("test_table", "my_idx_s", schema=config.test_schema)

assert not meth("test_table", "my_idx", schema=config.test_schema)

assert not meth(

"nonexistent_table", "my_idx_s", schema=config.test_schema

)

assert not meth(

"test_table", "nonexistent_idx_s", schema=config.test_schema

)

class BizarroCharacterTest(fixtures.TestBase):

__sparse_driver_backend__ = True

def column_names():

return testing.combinations(

("plainname",),

("(3)",),

("col%p",),

("[brack]",),

argnames="columnname",

)

def table_names():

return testing.combinations(

("plain",),

("(2)",),

("per % cent",),

("[brackets]",),

argnames="tablename",

)

@testing.variation("use_composite", [True, False])

@column_names()

@table_names()

@testing.requires.foreign_key_constraint_reflection

def test_fk_ref(

self, connection, metadata, use_composite, tablename, columnname

):

"""tests for #10275"""

tt = Table(

tablename,

metadata,

Column(columnname, Integer, key="id", primary_key=True),

test_needs_fk=True,

)

if use_composite:

tt.append_column(Column("id2", Integer, primary_key=True))

if use_composite:

Table(

"other",

metadata,

Column("id", Integer, primary_key=True),

Column("ref", Integer),

Column("ref2", Integer),

sa.ForeignKeyConstraint(["ref", "ref2"], [tt.c.id, tt.c.id2]),

test_needs_fk=True,

)

else:

Table(

"other",

metadata,

Column("id", Integer, primary_key=True),

Column("ref", ForeignKey(tt.c.id)),

test_needs_fk=True,

)

metadata.create_all(connection)

m2 = MetaData()

o2 = Table("other", m2, autoload_with=connection)

t1 = m2.tables[tablename]

assert o2.c.ref.references(t1.c[0])

if use_composite:

assert o2.c.ref2.references(t1.c[1])

@column_names()

@table_names()

@testing.requires.identity_columns

def test_reflect_identity(

self, tablename, columnname, connection, metadata

):

Table(

tablename,

metadata,

Column(columnname, Integer, Identity(), primary_key=True),

)

metadata.create_all(connection)

insp = inspect(connection)

eq_(insp.get_columns(tablename)[0]["identity"]["start"], 1)

@column_names()

@table_names()

@testing.requires.comment_reflection

def test_reflect_comments(

self, tablename, columnname, connection, metadata

):

Table(

tablename,

metadata,

Column("id", Integer, primary_key=True),

Column(columnname, Integer, comment="some comment"),

)

metadata.create_all(connection)

insp = inspect(connection)

eq_(insp.get_columns(tablename)[1]["comment"], "some comment")

class TempTableElementsTest(fixtures.TestBase):

__sparse_driver_backend__ = True

__requires__ = ("temp_table_reflection",)

@testing.fixture

def tablename(self):

return get_temp_table_name(

config, config.db, f"ident_tmp_{config.ident}"

)

@testing.requires.identity_columns

def test_reflect_identity(self, tablename, connection, metadata):

Table(

tablename,

metadata,

Column("id", Integer, Identity(), primary_key=True),

)

metadata.create_all(connection)

insp = inspect(connection)

eq_(insp.get_columns(tablename)[0]["identity"]["start"], 1)

@testing.requires.temp_table_comment_reflection

def test_reflect_comments(self, tablename, connection, metadata):

Table(

tablename,

metadata,

Column("id", Integer, primary_key=True),

Column("foobar", Integer, comment="some comment"),

)

metadata.create_all(connection)

insp = inspect(connection)

eq_(insp.get_columns(tablename)[1]["comment"], "some comment")

class QuotedNameArgumentTest(fixtures.TablesTest):

run_create_tables = "once"

__sparse_driver_backend__ = True

@classmethod

def define_tables(cls, metadata):

Table(

"quote ' one",

metadata,

Column("id", Integer),

Column("name", String(50)),

Column("data", String(50)),

Column("related_id", Integer),

sa.PrimaryKeyConstraint("id", name="pk quote ' one"),

sa.Index("ix quote ' one", "name"),

sa.UniqueConstraint(

"data",

name="uq quote' one",

),

sa.ForeignKeyConstraint(

["id"], ["related.id"], name="fk quote ' one"

),

sa.CheckConstraint("name != 'foo'", name="ck quote ' one"),

comment=r"""quote ' one comment""",

test_needs_fk=True,

)

if testing.requires.symbol_names_w_double_quote.enabled:

Table(

'quote " two',

metadata,

Column("id", Integer),

Column("name", String(50)),

Column("data", String(50)),

Column("related_id", Integer),

sa.PrimaryKeyConstraint("id", name='pk quote " two'),

sa.Index('ix quote " two', "name"),

sa.UniqueConstraint(

"data",

name='uq quote" two',

),

sa.ForeignKeyConstraint(

["id"], ["related.id"], name='fk quote " two'

),

sa.CheckConstraint("name != 'foo'", name='ck quote " two '),

comment=r"""quote " two comment""",

test_needs_fk=True,

)

Table(

"related",

metadata,

Column("id", Integer, primary_key=True),

Column("related", Integer),

test_needs_fk=True,

)

if testing.requires.view_column_reflection.enabled:

if testing.requires.symbol_names_w_double_quote.enabled:

names = [

"quote ' one",

'quote " two',

]

else:

names = [

"quote ' one",

]

for name in names:

query = "CREATE VIEW %s AS SELECT * FROM %s" % (

config.db.dialect.identifier_preparer.quote(

"view %s" % name

),

config.db.dialect.identifier_preparer.quote(name),

)

event.listen(metadata, "after_create", DDL(query))

event.listen(

metadata,

"before_drop",

DDL(

"DROP VIEW %s"

% config.db.dialect.identifier_preparer.quote(

"view %s" % name

)

),

)

def quote_fixtures(fn):

return testing.combinations(

("quote ' one",),

('quote " two', testing.requires.symbol_names_w_double_quote),

)(fn)

@quote_fixtures

def test_get_table_options(self, name):

insp = inspect(config.db)

if testing.requires.reflect_table_options.enabled:

res = insp.get_table_options(name)

is_true(isinstance(res, dict))

else:

with expect_raises(NotImplementedError):

insp.get_table_options(name)

@quote_fixtures

@testing.requires.view_column_reflection

def test_get_view_definition(self, name):

insp = inspect(config.db)

assert insp.get_view_definition("view %s" % name)

@quote_fixtures

def test_get_columns(self, name):

insp = inspect(config.db)

assert insp.get_columns(name)

@quote_fixtures

def test_get_pk_constraint(self, name):

insp = inspect(config.db)

assert insp.get_pk_constraint(name)

@quote_fixtures

@testing.requires.foreign_key_constraint_reflection

def test_get_foreign_keys(self, name):

insp = inspect(config.db)

assert insp.get_foreign_keys(name)

@quote_fixtures

@testing.requires.index_reflection

def test_get_indexes(self, name):

insp = inspect(config.db)

assert insp.get_indexes(name)

@quote_fixtures

@testing.requires.unique_constraint_reflection

def test_get_unique_constraints(self, name):

insp = inspect(config.db)

assert insp.get_unique_constraints(name)

@quote_fixtures

@testing.requires.comment_reflection

def test_get_table_comment(self, name):

insp = inspect(config.db)

assert insp.get_table_comment(name)

@quote_fixtures

@testing.requires.check_constraint_reflection

def test_get_check_constraints(self, name):

insp = inspect(config.db)

assert insp.get_check_constraints(name)

def _multi_combination(fn):

schema = testing.combinations(

None,

(

lambda: config.test_schema,

testing.requires.schemas,

),

argnames="schema",

)

scope = testing.combinations(

ObjectScope.DEFAULT,

ObjectScope.TEMPORARY,

ObjectScope.ANY,

argnames="scope",

)

kind = testing.combinations(

ObjectKind.TABLE,

ObjectKind.VIEW,

ObjectKind.MATERIALIZED_VIEW,

ObjectKind.ANY,

ObjectKind.ANY_VIEW,

ObjectKind.TABLE | ObjectKind.VIEW,

ObjectKind.TABLE | ObjectKind.MATERIALIZED_VIEW,

argnames="kind",

)

filter_names = testing.combinations(True, False, argnames="use_filter")

return schema(scope(kind(filter_names(fn))))

class ComponentReflectionTest(ComparesTables, OneConnectionTablesTest):

run_inserts = run_deletes = None

__sparse_driver_backend__ = True

@classmethod

def define_tables(cls, metadata):

cls.define_reflected_tables(metadata, None)

if testing.requires.schemas.enabled:

cls.define_reflected_tables(metadata, testing.config.test_schema)

@classmethod

def define_reflected_tables(cls, metadata, schema):

if schema:

schema_prefix = schema + "."

else:

schema_prefix = ""

if testing.requires.self_referential_foreign_keys.enabled:

parent_id_args = (

ForeignKey(

"%susers.user_id" % schema_prefix, name="user_id_fk"

),

)

else:

parent_id_args = ()

users = Table(

"users",

metadata,

Column("user_id", sa.INT, primary_key=True),

Column("test1", sa.CHAR(5), nullable=False),

Column("test2", sa.Float(), nullable=False),

Column("parent_user_id", sa.Integer, *parent_id_args),

sa.CheckConstraint(

"test2 > 0",

name="zz_test2_gt_zero",

comment="users check constraint",

),

sa.CheckConstraint("test2 <= 1000"),

schema=schema,

test_needs_fk=True,

)

Table(

"dingalings",

metadata,

Column("dingaling_id", sa.Integer, primary_key=True),

Column(

"address_id",

sa.Integer,

ForeignKey(

"%semail_addresses.address_id" % schema_prefix,

name="zz_email_add_id_fg",

comment="di fk comment",

),

),

Column(

"id_user",

sa.Integer,

ForeignKey("%susers.user_id" % schema_prefix),

),

Column("data", sa.String(30), unique=True),

sa.CheckConstraint(

"address_id > 0 AND address_id < 1000",

name="address_id_gt_zero",

),

sa.UniqueConstraint(

"address_id",

"dingaling_id",

name="zz_dingalings_multiple",

comment="di unique comment",

),

schema=schema,

test_needs_fk=True,

)

Table(

"email_addresses",

metadata,

Column("address_id", sa.Integer),

Column("remote_user_id", sa.Integer, ForeignKey(users.c.user_id)),

Column("email_address", sa.String(20), index=True),

sa.PrimaryKeyConstraint(

"address_id", name="email_ad_pk", comment="ea pk comment"

),

schema=schema,

test_needs_fk=True,

)

Table(

"comment_test",

metadata,

Column("id", sa.Integer, primary_key=True, comment="id comment"),

Column("data", sa.String(20), comment="data % comment"),

Column(

"d2",

sa.String(20),

comment=r"""Comment types type speedily ' " \ '' Fun!""",

),

Column("d3", sa.String(42), comment="Comment\nwith\rescapes"),

schema=schema,

comment=r"""the test % ' " \ table comment""",

)

Table(

"no_constraints",

metadata,

Column("data", sa.String(20)),

schema=schema,

comment="no\nconstraints\rhas\fescaped\vcomment",

)

if testing.requires.cross_schema_fk_reflection.enabled:

if schema is None:

Table(

"local_table",

metadata,

Column("id", sa.Integer, primary_key=True),

Column("data", sa.String(20)),

Column(

"remote_id",

ForeignKey(

"%s.remote_table_2.id" % testing.config.test_schema

),

),

test_needs_fk=True,

schema=config.db.dialect.default_schema_name,

)

else:

Table(

"remote_table",

metadata,

Column("id", sa.Integer, primary_key=True),

Column(

"local_id",

ForeignKey(

"%s.local_table.id"

% config.db.dialect.default_schema_name

),

),

Column("data", sa.String(20)),

schema=schema,

test_needs_fk=True,

)

Table(

"remote_table_2",

metadata,

Column("id", sa.Integer, primary_key=True),

Column("data", sa.String(20)),

schema=schema,

test_needs_fk=True,

)

if testing.requires.index_reflection.enabled:

Index("users_t_idx", users.c.test1, users.c.test2, unique=True)

Index(

"users_all_idx", users.c.user_id, users.c.test2, users.c.test1

)

if not schema:

# test_needs_fk is at the moment to force MySQL InnoDB

noncol_idx_test_nopk = Table(

"noncol_idx_test_nopk",

metadata,

Column("q", sa.String(5)),

test_needs_fk=True,

)

noncol_idx_test_pk = Table(

"noncol_idx_test_pk",

metadata,

Column("id", sa.Integer, primary_key=True),

Column("q", sa.String(5)),

test_needs_fk=True,

)

if (

testing.requires.indexes_with_ascdesc.enabled

and testing.requires.reflect_indexes_with_ascdesc.enabled

):

Index("noncol_idx_nopk", noncol_idx_test_nopk.c.q.desc())

Index("noncol_idx_pk", noncol_idx_test_pk.c.q.desc())

if testing.requires.view_column_reflection.enabled:

cls.define_views(metadata, schema)

if not schema and testing.requires.temp_table_reflection.enabled:

cls.define_temp_tables(metadata)

@classmethod

def temp_table_name(cls):

return get_temp_table_name(

config, config.db, f"user_tmp_{config.ident}"

)

@classmethod

def define_temp_tables(cls, metadata):

kw = temp_table_keyword_args(config, config.db)

table_name = cls.temp_table_name()

user_tmp = Table(

table_name,

metadata,

Column("id", sa.INT, primary_key=True),

Column("name", sa.VARCHAR(50)),

Column("foo", sa.INT),

# disambiguate temp table unique constraint names. this is

# pretty arbitrary for a generic dialect however we are doing

# it to suit SQL Server which will produce name conflicts for

# unique constraints created against temp tables in different

# databases.

# https://www.arbinada.com/en/node/1645

sa.UniqueConstraint("name", name=f"user_tmp_uq_{config.ident}"),

sa.Index("user_tmp_ix", "foo"),

**kw,

)

if (

testing.requires.view_reflection.enabled

and testing.requires.temporary_views.enabled

):

event.listen(

user_tmp,

"after_create",

DDL(

"create temporary view user_tmp_v as "

"select * from user_tmp_%s" % config.ident

),

)

event.listen(user_tmp, "before_drop", DDL("drop view user_tmp_v"))

@classmethod

def define_views(cls, metadata, schema):

if testing.requires.materialized_views.enabled:

materialized = {"dingalings"}

else:

materialized = set()

for table_name in ("users", "email_addresses", "dingalings"):

fullname = table_name

if schema:

fullname = f"{schema}.{table_name}"

view_name = fullname + "_v"

prefix = "MATERIALIZED " if table_name in materialized else ""

query = (

f"CREATE {prefix}VIEW {view_name} AS SELECT * FROM {fullname}"

)

event.listen(metadata, "after_create", DDL(query))

if table_name in materialized:

index_name = "mat_index"

if schema and testing.against("oracle"):

index_name = f"{schema}.{index_name}"

idx = f"CREATE INDEX {index_name} ON {view_name}(data)"

event.listen(metadata, "after_create", DDL(idx))

event.listen(

metadata, "before_drop", DDL(f"DROP {prefix}VIEW {view_name}")

)

def _resolve_kind(self, kind, tables, views, materialized):

res = {}

if ObjectKind.TABLE in kind:

res.update(tables)

if ObjectKind.VIEW in kind:

res.update(views)

if ObjectKind.MATERIALIZED_VIEW in kind:

res.update(materialized)

return res

def _resolve_views(self, views, materialized):

if not testing.requires.view_column_reflection.enabled:

materialized.clear()

views.clear()

elif not testing.requires.materialized_views.enabled:

views.update(materialized)

materialized.clear()

def _resolve_names(self, schema, scope, filter_names, values):

scope_filter = lambda _: True # noqa: E731

if scope is ObjectScope.DEFAULT:

scope_filter = lambda k: "tmp" not in k[1] # noqa: E731

if scope is ObjectScope.TEMPORARY:

scope_filter = lambda k: "tmp" in k[1] # noqa: E731

removed = {

None: {"remote_table", "remote_table_2"},

testing.config.test_schema: {

"local_table",

"noncol_idx_test_nopk",

"noncol_idx_test_pk",

"user_tmp_v",

self.temp_table_name(),

},

}

if not testing.requires.cross_schema_fk_reflection.enabled:

removed[None].add("local_table")

removed[testing.config.test_schema].update(

["remote_table", "remote_table_2"]

)

if not testing.requires.index_reflection.enabled:

removed[None].update(

["noncol_idx_test_nopk", "noncol_idx_test_pk"]

)

if (

not testing.requires.temp_table_reflection.enabled

or not testing.requires.temp_table_names.enabled

):

removed[None].update(["user_tmp_v", self.temp_table_name()])

if not testing.requires.temporary_views.enabled:

removed[None].update(["user_tmp_v"])

res = {

k: v

for k, v in values.items()

if scope_filter(k)

and k[1] not in removed[schema]

and (not filter_names or k[1] in filter_names)

}

return res

def exp_options(

self,

schema=None,

scope=ObjectScope.ANY,

kind=ObjectKind.ANY,

filter_names=None,

):

materialized = {(schema, "dingalings_v"): mock.ANY}

views = {

(schema, "email_addresses_v"): mock.ANY,

(schema, "users_v"): mock.ANY,

(schema, "user_tmp_v"): mock.ANY,

}

self._resolve_views(views, materialized)

tables = {

(schema, "users"): mock.ANY,

(schema, "dingalings"): mock.ANY,

(schema, "email_addresses"): mock.ANY,

(schema, "comment_test"): mock.ANY,

(schema, "no_constraints"): mock.ANY,

(schema, "local_table"): mock.ANY,

(schema, "remote_table"): mock.ANY,

(schema, "remote_table_2"): mock.ANY,

(schema, "noncol_idx_test_nopk"): mock.ANY,

(schema, "noncol_idx_test_pk"): mock.ANY,

(schema, self.temp_table_name()): mock.ANY,

}

res = self._resolve_kind(kind, tables, views, materialized)

res = self._resolve_names(schema, scope, filter_names, res)

return res

def exp_comments(

self,

schema=None,

scope=ObjectScope.ANY,

kind=ObjectKind.ANY,

filter_names=None,

):

empty = {"text": None}

materialized = {(schema, "dingalings_v"): empty}

views = {

(schema, "email_addresses_v"): empty,

(schema, "users_v"): empty,

(schema, "user_tmp_v"): empty,

}

self._resolve_views(views, materialized)

tables = {

(schema, "users"): empty,

(schema, "dingalings"): empty,

(schema, "email_addresses"): empty,

(schema, "comment_test"): {

"text": r"""the test % ' " \ table comment"""

},

(schema, "no_constraints"): {

"text": "no\nconstraints\rhas\fescaped\vcomment"

},

(schema, "local_table"): empty,

(schema, "remote_table"): empty,

(schema, "remote_table_2"): empty,

(schema, "noncol_idx_test_nopk"): empty,

(schema, "noncol_idx_test_pk"): empty,

(schema, self.temp_table_name()): empty,

}