codekingpro/portable-devtools

1# dialects/oracle/provision.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: ignore-errors
8
9import time
10
11from ... import create_engine
12from ... import exc
13from ... import inspect
14from ...engine import url as sa_url
15from ...testing.provision import configure_follower
16from ...testing.provision import create_db
17from ...testing.provision import drop_all_schema_objects_post_tables
18from ...testing.provision import drop_all_schema_objects_pre_tables
19from ...testing.provision import drop_db
20from ...testing.provision import follower_url_from_main
21from ...testing.provision import generate_driver_url
22from ...testing.provision import is_preferred_driver
23from ...testing.provision import log
24from ...testing.provision import post_configure_engine
25from ...testing.provision import post_configure_testing_engine
26from ...testing.provision import run_reap_dbs
27from ...testing.provision import set_default_schema_on_connection
28from ...testing.provision import stop_test_class_outside_fixtures
29from ...testing.provision import temp_table_keyword_args
30from ...testing.provision import update_db_opts
31from ...testing.warnings import warn_test_suite
32
33
34@generate_driver_url.for_db("oracle")
35def _oracle_generate_driver_url(url, driver, query_str):
36
37    backend = url.get_backend_name()
38
39    new_url = url.set(
40        drivername="%s+%s" % (backend, driver),
41    )
42
43    # use oracledb's retry feature, which is essential for oracle 23c
44    # which otherwise frequently rejects connections under load
45    # for cx_oracle we have a connect event instead
46    if driver in ("oracledb", "oracledb_async"):
47        # oracledb is even nice enough to convert from string to int
48        # for these opts, apparently
49        new_url = new_url.update_query_pairs(
50            [("retry_count", "5"), ("retry_delay", "2")]
51        )
52    else:
53        # remove these params for cx_oracle if we received an
54        # already-modified URL
55        new_url = new_url.difference_update_query(
56            ["retry_count", "retry_delay"]
57        )
58
59    try:
60        new_url.get_dialect()
61    except exc.NoSuchModuleError:
62        return None
63    else:
64        return new_url
65
66
67@create_db.for_db("oracle")
68def _oracle_create_db(cfg, eng, ident):
69    # NOTE: make sure you've run "ALTER DATABASE default tablespace users" or
70    # similar, so that the default tablespace is not "system"; reflection will
71    # fail otherwise
72    with eng.begin() as conn:
73        conn.exec_driver_sql("create user %s identified by xe" % ident)
74        conn.exec_driver_sql("create user %s_ts1 identified by xe" % ident)
75        conn.exec_driver_sql("create user %s_ts2 identified by xe" % ident)
76        conn.exec_driver_sql("grant dba to %s" % (ident,))
77        conn.exec_driver_sql("grant unlimited tablespace to %s" % ident)
78        conn.exec_driver_sql("grant unlimited tablespace to %s_ts1" % ident)
79        conn.exec_driver_sql("grant unlimited tablespace to %s_ts2" % ident)
80        # these are needed to create materialized views
81        conn.exec_driver_sql("grant create table to %s" % ident)
82        conn.exec_driver_sql("grant create table to %s_ts1" % ident)
83        conn.exec_driver_sql("grant create table to %s_ts2" % ident)
84
85
86@configure_follower.for_db("oracle")
87def _oracle_configure_follower(config, ident):
88    config.test_schema = "%s_ts1" % ident
89    config.test_schema_2 = "%s_ts2" % ident
90
91
92def _ora_drop_ignore(conn, dbname):
93    try:
94        conn.exec_driver_sql("drop user %s cascade" % dbname)
95        log.info("Reaped db: %s", dbname)
96        return True
97    except exc.DatabaseError as err:
98        log.warning("couldn't drop db: %s", err)
99        return False
100
101
102@drop_all_schema_objects_pre_tables.for_db("oracle")
103def _ora_drop_all_schema_objects_pre_tables(cfg, eng):
104    _purge_recyclebin(eng)
105    _purge_recyclebin(eng, cfg.test_schema)
106
107
108@drop_all_schema_objects_post_tables.for_db("oracle")
109def _ora_drop_all_schema_objects_post_tables(cfg, eng):
110    with eng.begin() as conn:
111        for syn in conn.dialect._get_synonyms(conn, None, None, None):
112            conn.exec_driver_sql(f"drop synonym {syn['synonym_name']}")
113
114        for syn in conn.dialect._get_synonyms(
115            conn, cfg.test_schema, None, None
116        ):
117            conn.exec_driver_sql(
118                f"drop synonym {cfg.test_schema}.{syn['synonym_name']}"
119            )
120
121        for tmp_table in inspect(conn).get_temp_table_names():
122            conn.exec_driver_sql(f"drop table {tmp_table}")
123
124
125@drop_db.for_db("oracle")
126def _oracle_drop_db(cfg, eng, ident):
127    with eng.begin() as conn:
128        # cx_Oracle seems to occasionally leak open connections when a large
129        # suite it run, even if we confirm we have zero references to
130        # connection objects.
131        # while there is a "kill session" command in Oracle Database,
132        # it unfortunately does not release the connection sufficiently.
133        _ora_drop_ignore(conn, ident)
134        _ora_drop_ignore(conn, "%s_ts1" % ident)
135        _ora_drop_ignore(conn, "%s_ts2" % ident)
136
137
138@stop_test_class_outside_fixtures.for_db("oracle")
139def _ora_stop_test_class_outside_fixtures(config, db, cls):
140    try:
141        _purge_recyclebin(db)
142    except exc.DatabaseError as err:
143        log.warning("purge recyclebin command failed: %s", err)
144
145
146def _purge_recyclebin(eng, schema=None):
147    with eng.begin() as conn:
148        if schema is None:
149            # run magic command to get rid of identity sequences
150            # https://floo.bar/2019/11/29/drop-the-underlying-sequence-of-an-identity-column/  # noqa: E501
151            conn.exec_driver_sql("purge recyclebin")
152        else:
153            # per user: https://community.oracle.com/tech/developers/discussion/2255402/how-to-clear-dba-recyclebin-for-a-particular-user  # noqa: E501
154            for owner, object_name, type_ in conn.exec_driver_sql(
155                "select owner, object_name,type from "
156                "dba_recyclebin where owner=:schema and type='TABLE'",
157                {"schema": conn.dialect.denormalize_name(schema)},
158            ).all():
159                conn.exec_driver_sql(f'purge {type_} {owner}."{object_name}"')
160
161
162@is_preferred_driver.for_db("oracle")
163def _oracle_is_preferred_driver(cfg, engine):
164    """establish oracledb as the preferred driver to use for tests, even
165    though cx_Oracle is still the "default" driver"""
166
167    return engine.dialect.driver == "oracledb" and not engine.dialect.is_async
168
169
170def _connect_with_retry(dialect, conn_rec, cargs, cparams):
171    assert dialect.driver == "cx_oracle"
172
173    def _is_couldnt_connect(err):
174        return "DPY-6005" in str(err) or "ORA-12516" in str(err)
175
176    err_ = None
177    for _ in range(5):
178        try:
179            return dialect.loaded_dbapi.connect(*cargs, **cparams)
180        except (
181            dialect.loaded_dbapi.DatabaseError,
182            dialect.loaded_dbapi.OperationalError,
183        ) as err:
184            err_ = err
185            if _is_couldnt_connect(err):
186                warn_test_suite("Oracle database reconnecting...")
187                time.sleep(2)
188                continue
189            else:
190                raise
191    if err_ is not None:
192        raise Exception("connect failed after five attempts") from err_
193
194
195@post_configure_testing_engine.for_db("oracle")
196def _oracle_post_configure_testing_engine(url, engine, options, scope):
197    from ... import event
198
199    if engine.dialect.driver == "cx_oracle":
200        event.listen(engine, "do_connect", _connect_with_retry)
201
202
203@post_configure_engine.for_db("oracle")
204def _oracle_post_configure_engine(url, engine, follower_ident):
205
206    from ... import event
207
208    @event.listens_for(engine, "checkin")
209    def checkin(dbapi_connection, connection_record):
210        # this was meant to work around this issue:
211        # https://github.com/oracle/python-cx_Oracle/issues/530
212        # invalidate oracle connections that had 2pc set up
213        # however things are too complex with some of the 2pc tests,
214        # so just block cx_oracle from being used in 2pc tests (use oracledb
215        # instead)
216        # if "cx_oracle_xid" in connection_record.info:
217        #    connection_record.invalidate()
218
219        # clear statement cache on all connections that were used
220        # https://github.com/oracle/python-cx_Oracle/issues/519
221        # TODO: oracledb claims to have this feature built in somehow,
222        # see if that's in use and/or if it needs to be enabled
223        # (or if this doesn't even apply to the newer oracle's we're using)
224        try:
225            sc = dbapi_connection.stmtcachesize
226        except:
227            # connection closed
228            pass
229        else:
230            dbapi_connection.stmtcachesize = 0
231            dbapi_connection.stmtcachesize = sc
232
233
234@run_reap_dbs.for_db("oracle")
235def _reap_oracle_dbs(url, idents):
236    log.info("db reaper connecting to %r", url)
237    eng = create_engine(url)
238    with eng.begin() as conn:
239        log.info("identifiers in file: %s", ", ".join(idents))
240
241        to_reap = conn.exec_driver_sql(
242            "select u.username from all_users u where username "
243            "like 'TEST_%' and not exists (select username "
244            "from v$session where username=u.username)"
245        )
246        all_names = {username.lower() for (username,) in to_reap}
247        to_drop = set()
248        for name in all_names:
249            if name.endswith("_ts1") or name.endswith("_ts2"):
250                continue
251            elif name in idents:
252                to_drop.add(name)
253                if "%s_ts1" % name in all_names:
254                    to_drop.add("%s_ts1" % name)
255                if "%s_ts2" % name in all_names:
256                    to_drop.add("%s_ts2" % name)
257
258        dropped = total = 0
259        for total, username in enumerate(to_drop, 1):
260            if _ora_drop_ignore(conn, username):
261                dropped += 1
262        log.info(
263            "Dropped %d out of %d stale databases detected", dropped, total
264        )
265
266
267@follower_url_from_main.for_db("oracle")
268def _oracle_follower_url_from_main(url, ident):
269    url = sa_url.make_url(url)
270    return url.set(username=ident, password="xe")
271
272
273@temp_table_keyword_args.for_db("oracle")
274def _oracle_temp_table_keyword_args(cfg, eng):
275    return {
276        "prefixes": ["GLOBAL TEMPORARY"],
277        "oracle_on_commit": "PRESERVE ROWS",
278    }
279
280
281@set_default_schema_on_connection.for_db("oracle")
282def _oracle_set_default_schema_on_connection(
283    cfg, dbapi_connection, schema_name
284):
285    cursor = dbapi_connection.cursor()
286    cursor.execute("ALTER SESSION SET CURRENT_SCHEMA=%s" % schema_name)
287    cursor.close()
288
289
290@update_db_opts.for_db("oracle")
291def _update_db_opts(db_url, db_opts, options):
292    """Set database options (db_opts) for a test database that we created."""
293    if (
294        options.oracledb_thick_mode
295        and sa_url.make_url(db_url).get_driver_name() == "oracledb"
296    ):
297        db_opts["thick_mode"] = True
298