diff options
Diffstat (limited to 'lib/sqlalchemy/testing/provision.py')
| -rw-r--r-- | lib/sqlalchemy/testing/provision.py | 70 |
1 files changed, 50 insertions, 20 deletions
diff --git a/lib/sqlalchemy/testing/provision.py b/lib/sqlalchemy/testing/provision.py index d38437732..498d92a77 100644 --- a/lib/sqlalchemy/testing/provision.py +++ b/lib/sqlalchemy/testing/provision.py @@ -230,7 +230,39 @@ def drop_all_schema_objects(cfg, eng): drop_all_schema_objects_pre_tables(cfg, eng) + drop_views(cfg, eng) + + if config.requirements.materialized_views.enabled: + drop_materialized_views(cfg, eng) + inspector = inspect(eng) + + consider_schemas = (None,) + if config.requirements.schemas.enabled_for_config(cfg): + consider_schemas += (cfg.test_schema, cfg.test_schema_2) + util.drop_all_tables(eng, inspector, consider_schemas=consider_schemas) + + drop_all_schema_objects_post_tables(cfg, eng) + + if config.requirements.sequences.enabled_for_config(cfg): + with eng.begin() as conn: + for seq in inspector.get_sequence_names(): + conn.execute(ddl.DropSequence(schema.Sequence(seq))) + if config.requirements.schemas.enabled_for_config(cfg): + for schema_name in [cfg.test_schema, cfg.test_schema_2]: + for seq in inspector.get_sequence_names( + schema=schema_name + ): + conn.execute( + ddl.DropSequence( + schema.Sequence(seq, schema=schema_name) + ) + ) + + +def drop_views(cfg, eng): + inspector = inspect(eng) + try: view_names = inspector.get_view_names() except NotImplementedError: @@ -244,7 +276,7 @@ def drop_all_schema_objects(cfg, eng): if config.requirements.schemas.enabled_for_config(cfg): try: - view_names = inspector.get_view_names(schema="test_schema") + view_names = inspector.get_view_names(schema=cfg.test_schema) except NotImplementedError: pass else: @@ -255,32 +287,30 @@ def drop_all_schema_objects(cfg, eng): schema.Table( vname, schema.MetaData(), - schema="test_schema", + schema=cfg.test_schema, ) ) ) - util.drop_all_tables(eng, inspector) - if config.requirements.schemas.enabled_for_config(cfg): - util.drop_all_tables(eng, inspector, schema=cfg.test_schema) - util.drop_all_tables(eng, inspector, schema=cfg.test_schema_2) - drop_all_schema_objects_post_tables(cfg, eng) +def drop_materialized_views(cfg, eng): + inspector = inspect(eng) - if config.requirements.sequences.enabled_for_config(cfg): + mview_names = inspector.get_materialized_view_names() + + with eng.begin() as conn: + for vname in mview_names: + conn.exec_driver_sql(f"DROP MATERIALIZED VIEW {vname}") + + if config.requirements.schemas.enabled_for_config(cfg): + mview_names = inspector.get_materialized_view_names( + schema=cfg.test_schema + ) with eng.begin() as conn: - for seq in inspector.get_sequence_names(): - conn.execute(ddl.DropSequence(schema.Sequence(seq))) - if config.requirements.schemas.enabled_for_config(cfg): - for schema_name in [cfg.test_schema, cfg.test_schema_2]: - for seq in inspector.get_sequence_names( - schema=schema_name - ): - conn.execute( - ddl.DropSequence( - schema.Sequence(seq, schema=schema_name) - ) - ) + for vname in mview_names: + conn.exec_driver_sql( + f"DROP MATERIALIZED VIEW {cfg.test_schema}.{vname}" + ) @register.init |