summaryrefslogtreecommitdiff
path: root/lib/sqlalchemy/dialects
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2020-12-21 10:22:43 -0500
committerMike Bayer <mike_mp@zzzcomputing.com>2021-01-03 13:22:29 -0500
commitfd3c063dd68b289814af724689165418de5e4408 (patch)
tree3a13c1cd3bd58b8b5b88bc3294e491aca63ecf0b /lib/sqlalchemy/dialects
parentdd41a5e61a30a2d05ee09f583fdfde1f1c204807 (diff)
downloadsqlalchemy-fd3c063dd68b289814af724689165418de5e4408.tar.gz
remove metadata.bind use from test suite
importantly this means we can remove bound metadata from the fixtures that are used by Alembic's test suite. hopefully this is the last one that has to happen to allow Alembic to be fully 1.4/2.0. Start moving from @testing.provide_metadata to a pytest metadata fixture. This does not seem to have any negative effects even though TablesTest uses a "self.metadata" attribute. Change-Id: Iae6ab95938a7e92b6d42086aec534af27b5577d3
Diffstat (limited to 'lib/sqlalchemy/dialects')
-rw-r--r--lib/sqlalchemy/dialects/postgresql/provision.py24
1 files changed, 24 insertions, 0 deletions
diff --git a/lib/sqlalchemy/dialects/postgresql/provision.py b/lib/sqlalchemy/dialects/postgresql/provision.py
index 575316c61..d345cdfdf 100644
--- a/lib/sqlalchemy/dialects/postgresql/provision.py
+++ b/lib/sqlalchemy/dialects/postgresql/provision.py
@@ -1,8 +1,11 @@
import time
from ... import exc
+from ... import inspect
from ... import text
from ...testing.provision import create_db
+from ...testing.provision import drop_all_schema_objects_post_tables
+from ...testing.provision import drop_all_schema_objects_pre_tables
from ...testing.provision import drop_db
from ...testing.provision import log
from ...testing.provision import set_default_schema_on_connection
@@ -78,3 +81,24 @@ def _postgresql_set_default_schema_on_connection(
cursor.execute("SET SESSION search_path='%s'" % schema_name)
cursor.close()
dbapi_connection.autocommit = existing_autocommit
+
+
+@drop_all_schema_objects_pre_tables.for_db("postgresql")
+def drop_all_schema_objects_pre_tables(cfg, eng):
+ with eng.connect().execution_options(isolation_level="AUTOCOMMIT") as conn:
+ for xid in conn.execute("select gid from pg_prepared_xacts").scalars():
+ conn.execute("ROLLBACK PREPARED '%s'" % xid)
+
+
+@drop_all_schema_objects_post_tables.for_db("postgresql")
+def drop_all_schema_objects_post_tables(cfg, eng):
+ from sqlalchemy.dialects import postgresql
+
+ inspector = inspect(eng)
+ with eng.begin() as conn:
+ for enum in inspector.get_enums("*"):
+ conn.execute(
+ postgresql.DropEnumType(
+ postgresql.ENUM(name=enum["name"], schema=enum["schema"])
+ )
+ )