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

diff --git a/test/dialect/postgresql/test_compiler.py b/test/dialect/postgresql/test_compiler.py
index d1b753d54feed0d68465c34ee703791084b529c2..d6a71b53b8de139d71d396f16d11b4c8f70973e2 100644 (file)
--- 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 c45b4bc9b2bd50ef530c150513c3185ce62f9c07..7ed794b61f4c9f4119d0424eff1addebc73af32d 100644 (file)
--- 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