From c4b68ebfa6e5c80b679306612b63c22cd9f13a1d Mon Sep 17 00:00:00 2001 From: KapilDagur Date: Sat, 13 Sep 2025 00:47:51 +0530 Subject: [PATCH] test(postgresql): add schema resolution tests for ENUM/DOMAIN with builtin names Add comprehensive unit tests to verify schema resolution for ENUM and DOMAIN types when their names conflict with PostgreSQL built-in types. Covers scenarios: * raises CompileError when no schema/default is provided * works with explicit schema specified * inherits schema from table definition * inherits schema from metadata Tests are parameterized via @testing.variation and applied to both ENUM and DOMAIN. Also includes compiler tests for naming conflicts to ensure consistent error handling and valid SQL generation. Closes: #12761 --- test/dialect/postgresql/test_compiler.py | 109 +++++++++++++++++++++++ test/dialect/postgresql/test_types.py | 106 ++++++++++++++++++++++ 2 files changed, 215 insertions(+) diff --git a/test/dialect/postgresql/test_compiler.py b/test/dialect/postgresql/test_compiler.py index d1b753d54f..d6a71b53b8 100644 --- a/test/dialect/postgresql/test_compiler.py +++ b/test/dialect/postgresql/test_compiler.py @@ -68,6 +68,7 @@ from sqlalchemy.sql import operators from sqlalchemy.sql import table from sqlalchemy.sql import util as sql_util from sqlalchemy.sql.functions import GenericFunction +from sqlalchemy.testing import config from sqlalchemy.testing import expect_raises from sqlalchemy.testing import expect_raises_message from sqlalchemy.testing import fixtures @@ -356,6 +357,60 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): render_schema_translate=True, ) + def test_enum_with_builtin_name_schema_is_qualified(self): + for sch in (config.test_schema, config.test_schema_2): + e = Enum("active", "inactive", name="text", schema=sch) + self.assert_compile( + postgresql.CreateEnumType(e), + f"CREATE TYPE {sch}.text AS ENUM ('active', 'inactive')", + ) + + t = Table("user_status", MetaData(), Column("status", e)) + self.assert_compile( + schema.CreateTable(t), + f"CREATE TABLE user_status (status {sch}.text)", + ) + + def test_drop_enum_with_builtin_name_is_qualified(self): + for sch in (config.test_schema, config.test_schema_2): + e = Enum("a", "b", name="text", schema=sch) + self.assert_compile( + postgresql.DropEnumType(e), + f"DROP TYPE {sch}.text", + ) + + def test_cast_enum_with_builtin_name(self): + for sch in (config.test_schema, config.test_schema_2): + e = Enum("opt1", "opt2", name="text", schema=sch) + stmt = select(cast(column("foo"), e)) + self.assert_compile( + stmt, + f"SELECT CAST(foo AS {sch}.text) AS foo", + ) + + def test_compileerror_enum_without_schema_builtin_name(self): + e = Enum("yes", "no", name="text") + dialect = postgresql.dialect() + dialect.default_schema_name = None + + with expect_raises_message( + exc.CompileError, + "ENUM with name 'text' requires an explicit schema " + "when no default_schema_name is configured", + ): + e.compile(dialect=dialect) + + def test_schema_translate_for_builtin_name_enum(self): + for sch in (config.test_schema, config.test_schema_2): + e = Enum("hot", "cold", name="text", schema=sch) + schema_translate_map = {sch: "translated_schema"} + self.assert_compile( + postgresql.CreateEnumType(e), + "CREATE TYPE translated_schema.text AS ENUM ('hot', 'cold')", + schema_translate_map=schema_translate_map, + render_schema_translate=True, + ) + def test_domain(self): self.assert_compile( postgresql.CreateDomainType( @@ -429,6 +484,60 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): render_schema_translate=True, ) + def test_domain_with_builtin_name_schema_is_qualified(self): + for sch in (config.test_schema, config.test_schema_2): + d = DOMAIN("text", Integer, schema=sch) + self.assert_compile( + postgresql.CreateDomainType(d), + f"CREATE DOMAIN {sch}.text AS INTEGER ", + ) + + t = Table("doc", MetaData(), Column("body", d)) + self.assert_compile( + schema.CreateTable(t), + f"CREATE TABLE doc (body {sch}.text)", + ) + + def test_drop_domain_with_builtin_name_is_qualified(self): + for sch in (config.test_schema, config.test_schema_2): + d = DOMAIN("text", Integer, schema=sch) + self.assert_compile( + postgresql.DropDomainType(d), + f"DROP DOMAIN {sch}.text", + ) + + def test_cast_domain_with_builtin_name(self): + for sch in (config.test_schema, config.test_schema_2): + d = DOMAIN("text", Integer, schema=sch) + stmt = select(cast(column("bar"), d)) + self.assert_compile( + stmt, + f"SELECT CAST(bar AS {sch}.text) AS bar", + ) + + def test_compileerror_domain_without_schema_builtin_name(self): + d = DOMAIN("text", Integer) + dialect = postgresql.dialect() + dialect.default_schema_name = None + + with expect_raises_message( + exc.CompileError, + "DOMAIN with name 'text' requires an explicit schema " + "when no default_schema_name is configured", + ): + d.compile(dialect=dialect) + + def test_schema_translate_for_builtin_name_domain(self): + for sch in (config.test_schema, config.test_schema_2): + d = DOMAIN("text", Integer, schema=sch) + schema_translate_map = {sch: "translated_schema"} + self.assert_compile( + postgresql.CreateDomainType(d), + "CREATE DOMAIN translated_schema.text AS INTEGER ", + schema_translate_map=schema_translate_map, + render_schema_translate=True, + ) + def test_create_table_with_schema_type_schema_translate(self): e1 = Enum("x", "y", "z", name="somename") e2 = Enum("x", "y", "z", name="somename", schema="someschema") diff --git a/test/dialect/postgresql/test_types.py b/test/dialect/postgresql/test_types.py index c45b4bc9b2..7ed794b61f 100644 --- a/test/dialect/postgresql/test_types.py +++ b/test/dialect/postgresql/test_types.py @@ -84,6 +84,7 @@ from sqlalchemy.orm import Session from sqlalchemy.sql import bindparam from sqlalchemy.sql import operators from sqlalchemy.sql import sqltypes +from sqlalchemy.testing import config from sqlalchemy.testing import expect_raises from sqlalchemy.testing import expect_raises_message from sqlalchemy.testing import fixtures @@ -1241,6 +1242,111 @@ class NamedTypeTest( e["name"] for e in inspect(connection).get_enums() ] + def _setup_type(self, type_kind, name, **kw): + """Create ENUM or DOMAIN with given name and args.""" + if type_kind.enum: + return ENUM("a", "b", "c", name=name, **kw) + else: + return DOMAIN(name=name, data_type=Integer, **kw) + + def _method_for(self, type_kind): + """Return inspector method name for given type.""" + return "get_enums" if type_kind.enum else "get_domains" + + def _type_names_for_schema(self, connection, method_name, schema): + """Fetch type names from inspector for a schema.""" + insp = inspect(connection) + method = getattr(insp, method_name) + return [t["name"] for t in method(schema=schema)] + + @testing.variation("type_kind", ["enum", "domain"]) + def test_builtin_name_conflict_compiletime_failure( + self, connection, type_kind + ): + """Case 1: Fails if builtin name used without schema/default.""" + dialect = postgresql.dialect() + dialect.default_schema_name = None + builtin_name = "text" + + t = self._setup_type(type_kind, builtin_name) + + assert_raises_message( + exc.CompileError, + f"{t.__class__.__name__.upper()} with " + f"name '{builtin_name}' requires an explicit schema " + "when no default_schema_name is configured", + t.compile, + dialect=dialect, + ) + + @testing.variation("type_kind", ["enum", "domain"]) + def test_builtin_name_conflict_explicit_schema( + self, connection, type_kind + ): + """Case 2: Works with explicit schema.""" + explicit_schema = config.test_schema + builtin_name = "text" + t = self._setup_type(type_kind, builtin_name, schema=explicit_schema) + + metadata = MetaData() + Table("t", metadata, Column("c", t)) + metadata.create_all(connection) + + eq_( + self._type_names_for_schema( + connection, self._method_for(type_kind), explicit_schema + ), + [builtin_name], + ) + metadata.drop_all(connection) + + @testing.variation("type_kind", ["enum", "domain"]) + def test_builtin_name_conflict_inherit_schema_from_table( + self, connection, type_kind + ): + """Case 3: Inherits schema from table.""" + explicit_schema = config.test_schema + inherited_name = ( + "status_enum_inherit" if type_kind.enum else "age_domain_inherit" + ) + + t = self._setup_type(type_kind, inherited_name, inherit_schema=True) + + metadata = MetaData() + Table("t", metadata, Column("c", t), schema=explicit_schema) + metadata.create_all(connection) + + eq_( + self._type_names_for_schema( + connection, self._method_for(type_kind), explicit_schema + ), + [inherited_name], + ) + metadata.drop_all(connection) + + @testing.variation("type_kind", ["enum", "domain"]) + def test_builtin_name_conflict_inherit_schema_from_metadata( + self, connection, type_kind + ): + """Case 4: Inherits schema from metadata.""" + explicit_schema = config.test_schema + meta_inherit_name = ( + "priority_enum_meta" if type_kind.enum else "score_domain_meta" + ) + + metadata = MetaData(schema=explicit_schema) + t = self._setup_type(type_kind, meta_inherit_name) + Table("t", metadata, Column("c", t)) + metadata.create_all(connection) + + eq_( + self._type_names_for_schema( + connection, self._method_for(type_kind), explicit_schema + ), + [meta_inherit_name], + ) + metadata.drop_all(connection) + class DomainTest( AssertsCompiledSQL, fixtures.TestBase, AssertsExecutionResults -- 2.47.3