From 017fd9ae0645eaf2a0fbdd067d10c721505b018c Mon Sep 17 00:00:00 2001 From: David Baumgold Date: Fri, 11 Feb 2022 12:30:24 -0500 Subject: Domain type Added a new Postgresql :class:`_postgresql.DOMAIN` datatype, which follows the same CREATE TYPE / DROP TYPE behaviors as that of PostgreSQL :class:`_postgresql.ENUM`. Much thanks to David Baumgold for the efforts on this. Fixes: #7316 Closes: #7317 Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/7317 Pull-request-sha: bc9a82f010e6ca2f70a6e8a7620b748e483c26c3 Change-Id: Id8d7e48843a896de17d20cc466b115b3cc065132 --- test/dialect/postgresql/test_reflection.py | 126 ++++++++++++++++++++++++++++- 1 file changed, 123 insertions(+), 3 deletions(-) (limited to 'test/dialect/postgresql/test_reflection.py') diff --git a/test/dialect/postgresql/test_reflection.py b/test/dialect/postgresql/test_reflection.py index 21b4149bc..99bc14d78 100644 --- a/test/dialect/postgresql/test_reflection.py +++ b/test/dialect/postgresql/test_reflection.py @@ -410,6 +410,9 @@ class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults): "CREATE DOMAIN nullable_domain AS TEXT CHECK " "(VALUE IN('FOO', 'BAR'))", "CREATE DOMAIN not_nullable_domain AS TEXT NOT NULL", + "CREATE DOMAIN my_int AS int CONSTRAINT b_my_int_one CHECK " + "(VALUE > 1) CONSTRAINT a_my_int_two CHECK (VALUE < 42) " + "CHECK(VALUE != 22)", ]: try: con.exec_driver_sql(ddl) @@ -468,6 +471,7 @@ class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults): con.exec_driver_sql("DROP TABLE nullable_domain_test") con.exec_driver_sql("DROP DOMAIN nullable_domain") con.exec_driver_sql("DROP DOMAIN not_nullable_domain") + con.exec_driver_sql("DROP DOMAIN my_int") def test_table_is_reflected(self, connection): metadata = MetaData() @@ -579,6 +583,122 @@ class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults): finally: base.PGDialect.ischema_names = ischema_names + @property + def all_domains(self): + return { + "public": [ + { + "visible": True, + "name": "arraydomain", + "schema": "public", + "nullable": True, + "type": "integer[]", + "default": None, + "constraints": [], + }, + { + "visible": True, + "name": "enumdomain", + "schema": "public", + "nullable": True, + "type": "testtype", + "default": None, + "constraints": [], + }, + { + "visible": True, + "name": "my_int", + "schema": "public", + "nullable": True, + "type": "integer", + "default": None, + "constraints": [ + {"check": "VALUE < 42", "name": "a_my_int_two"}, + {"check": "VALUE > 1", "name": "b_my_int_one"}, + # autogenerated name by pg + {"check": "VALUE <> 22", "name": "my_int_check"}, + ], + }, + { + "visible": True, + "name": "not_nullable_domain", + "schema": "public", + "nullable": False, + "type": "text", + "default": None, + "constraints": [], + }, + { + "visible": True, + "name": "nullable_domain", + "schema": "public", + "nullable": True, + "type": "text", + "default": None, + "constraints": [ + { + "check": "VALUE = ANY (ARRAY['FOO'::text, " + "'BAR'::text])", + # autogenerated name by pg + "name": "nullable_domain_check", + } + ], + }, + { + "visible": True, + "name": "testdomain", + "schema": "public", + "nullable": False, + "type": "integer", + "default": "42", + "constraints": [], + }, + ], + "test_schema": [ + { + "visible": False, + "name": "testdomain", + "schema": "test_schema", + "nullable": True, + "type": "integer", + "default": "0", + "constraints": [], + } + ], + "SomeSchema": [ + { + "visible": False, + "name": "Quoted.Domain", + "schema": "SomeSchema", + "nullable": True, + "type": "integer", + "default": "0", + "constraints": [], + } + ], + } + + def test_inspect_domains(self, connection): + inspector = inspect(connection) + eq_(inspector.get_domains(), self.all_domains["public"]) + + def test_inspect_domains_schema(self, connection): + inspector = inspect(connection) + eq_( + inspector.get_domains("test_schema"), + self.all_domains["test_schema"], + ) + eq_( + inspector.get_domains("SomeSchema"), self.all_domains["SomeSchema"] + ) + + def test_inspect_domains_star(self, connection): + inspector = inspect(connection) + all_ = [d for dl in self.all_domains.values() for d in dl] + all_ += inspector.get_domains("information_schema") + exp = sorted(all_, key=lambda d: (d["schema"], d["name"])) + eq_(inspector.get_domains("*"), exp) + class ReflectionTest( ReflectionFixtures, AssertsCompiledSQL, fixtures.TestBase @@ -1800,10 +1920,10 @@ class ReflectionTest( eq_( check_constraints, { - "cc1": "(a > 1) AND (a < 5)", - "cc2": "(a = 1) OR ((a > 2) AND (a < 5))", + "cc1": "a > 1 AND a < 5", + "cc2": "a = 1 OR a > 2 AND a < 5", "cc3": "is_positive(a)", - "cc4": "(b)::text <> 'hi\nim a name \nyup\n'::text", + "cc4": "b::text <> 'hi\nim a name \nyup\n'::text", }, ) -- cgit v1.2.1