col.default_on_null,
(
SELECT id.generation_type || ',' || id.IDENTITY_OPTIONS
- FROM ALL_TAB_IDENTITY_COLS id
+ FROM ALL_TAB_IDENTITY_COLS%(dblink)s id
WHERE col.table_name = id.table_name
AND col.column_name = id.column_name
AND col.owner = id.owner
- ) AS identity_options"""
+ ) AS identity_options""" % {
+ "dblink": dblink
+ }
else:
identity_cols = "NULL as default_on_null, NULL as identity_options"
from ...testing.provision import log
from ...testing.provision import run_reap_dbs
from ...testing.provision import set_default_schema_on_connection
+from ...testing.provision import stop_test_class
from ...testing.provision import temp_table_keyword_args
from ...testing.provision import update_db_opts
pass
+@stop_test_class.for_db("oracle")
+def stop_test_class(config, db, cls):
+ """run magic command to get rid of identity sequences
+
+ # https://floo.bar/2019/11/29/drop-the-underlying-sequence-of-an-identity-column/
+
+ """
+
+ with db.begin() as conn:
+ conn.exec_driver_sql("purge recyclebin")
+
+
@run_reap_dbs.for_db("oracle")
def _reap_oracle_dbs(url, idents):
log.info("db reaper connecting to %r", url)
drop_all_tables(self.metadata, self.bind)
# no need to run deletes if tables are recreated on setup
- if self.run_define_tables != "each" and self.run_deletes == "each":
+ if (
+ self.run_define_tables != "each"
+ and self.run_create_tables != "each"
+ and self.run_deletes == "each"
+ ):
with self.bind.begin() as conn:
for table in reversed(
[
exclusions = None
warnings = None
profiling = None
+provision = None
assertions = None
requirements = None
config = None
fn(options, file_config)
# late imports, has to happen after config.
- global util, fixtures, engines, exclusions, assertions
+ global util, fixtures, engines, exclusions, assertions, provision
global warnings, profiling, config, testing
from sqlalchemy import testing # noqa
from sqlalchemy.testing import fixtures, engines, exclusions # noqa
from sqlalchemy.testing import assertions, warnings, profiling # noqa
- from sqlalchemy.testing import config # noqa
+ from sqlalchemy.testing import config, provision # noqa
from sqlalchemy import util # noqa
warnings.setup_filters()
def stop_test_class(cls):
# from sqlalchemy import inspect
# assert not inspect(testing.db).get_table_names()
+
+ provision.stop_test_class(config, config.db, cls)
engines.testing_reaper._stop_test_ctx()
try:
if not options.low_connections:
)
+@register.init
+def stop_test_class(config, db, testcls):
+ pass
+
+
@register.init
def get_temp_table_name(cfg, eng, base_name):
"""Specify table name for creating a temporary Table.
@testing.provide_metadata
def test_table_round_trip(self, connection):
- oracle.RESERVED_WORDS.remove("UNION")
+ oracle.RESERVED_WORDS.discard("UNION")
metadata = self.metadata
table = Table(