codekingpro/portable-devtools

1# dialects/oracle/provision.py
2# Copyright (C) 2005-2024 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
9from ... import create_engine
10from ... import exc
11from ... import inspect
12from ...engine import url as sa_url
13from ...testing.provision import configure_follower
14from ...testing.provision import create_db
15from ...testing.provision import drop_all_schema_objects_post_tables
16from ...testing.provision import drop_all_schema_objects_pre_tables
17from ...testing.provision import drop_db
18from ...testing.provision import follower_url_from_main
19from ...testing.provision import log
20from ...testing.provision import post_configure_engine
21from ...testing.provision import run_reap_dbs
22from ...testing.provision import set_default_schema_on_connection
23from ...testing.provision import stop_test_class_outside_fixtures
24from ...testing.provision import temp_table_keyword_args
25from ...testing.provision import update_db_opts
26
27
28@create_db.for_db("oracle")
29def _oracle_create_db(cfg, eng, ident):
30    # NOTE: make sure you've run "ALTER DATABASE default tablespace users" or
31    # similar, so that the default tablespace is not "system"; reflection will
32    # fail otherwise
33    with eng.begin() as conn:
34        conn.exec_driver_sql("create user %s identified by xe" % ident)
35        conn.exec_driver_sql("create user %s_ts1 identified by xe" % ident)
36        conn.exec_driver_sql("create user %s_ts2 identified by xe" % ident)
37        conn.exec_driver_sql("grant dba to %s" % (ident,))
38        conn.exec_driver_sql("grant unlimited tablespace to %s" % ident)
39        conn.exec_driver_sql("grant unlimited tablespace to %s_ts1" % ident)
40        conn.exec_driver_sql("grant unlimited tablespace to %s_ts2" % ident)
41        # these are needed to create materialized views
42        conn.exec_driver_sql("grant create table to %s" % ident)
43        conn.exec_driver_sql("grant create table to %s_ts1" % ident)
44        conn.exec_driver_sql("grant create table to %s_ts2" % ident)
45
46
47@configure_follower.for_db("oracle")
48def _oracle_configure_follower(config, ident):
49    config.test_schema = "%s_ts1" % ident
50    config.test_schema_2 = "%s_ts2" % ident
51
52
53def _ora_drop_ignore(conn, dbname):
54    try:
55        conn.exec_driver_sql("drop user %s cascade" % dbname)
56        log.info("Reaped db: %s", dbname)
57        return True
58    except exc.DatabaseError as err:
59        log.warning("couldn't drop db: %s", err)
60        return False
61
62
63@drop_all_schema_objects_pre_tables.for_db("oracle")
64def _ora_drop_all_schema_objects_pre_tables(cfg, eng):
65    _purge_recyclebin(eng)
66    _purge_recyclebin(eng, cfg.test_schema)
67
68
69@drop_all_schema_objects_post_tables.for_db("oracle")
70def _ora_drop_all_schema_objects_post_tables(cfg, eng):
71    with eng.begin() as conn:
72        for syn in conn.dialect._get_synonyms(conn, None, None, None):
73            conn.exec_driver_sql(f"drop synonym {syn['synonym_name']}")
74
75        for syn in conn.dialect._get_synonyms(
76            conn, cfg.test_schema, None, None
77        ):
78            conn.exec_driver_sql(
79                f"drop synonym {cfg.test_schema}.{syn['synonym_name']}"
80            )
81
82        for tmp_table in inspect(conn).get_temp_table_names():
83            conn.exec_driver_sql(f"drop table {tmp_table}")
84
85
86@drop_db.for_db("oracle")
87def _oracle_drop_db(cfg, eng, ident):
88    with eng.begin() as conn:
89        # cx_Oracle seems to occasionally leak open connections when a large
90        # suite it run, even if we confirm we have zero references to
91        # connection objects.
92        # while there is a "kill session" command in Oracle,
93        # it unfortunately does not release the connection sufficiently.
94        _ora_drop_ignore(conn, ident)
95        _ora_drop_ignore(conn, "%s_ts1" % ident)
96        _ora_drop_ignore(conn, "%s_ts2" % ident)
97
98
99@stop_test_class_outside_fixtures.for_db("oracle")
100def _ora_stop_test_class_outside_fixtures(config, db, cls):
101    try:
102        _purge_recyclebin(db)
103    except exc.DatabaseError as err:
104        log.warning("purge recyclebin command failed: %s", err)
105
106    # clear statement cache on all connections that were used
107    # https://github.com/oracle/python-cx_Oracle/issues/519
108
109    for cx_oracle_conn in _all_conns:
110        try:
111            sc = cx_oracle_conn.stmtcachesize
112        except db.dialect.dbapi.InterfaceError:
113            # connection closed
114            pass
115        else:
116            cx_oracle_conn.stmtcachesize = 0
117            cx_oracle_conn.stmtcachesize = sc
118    _all_conns.clear()
119
120
121def _purge_recyclebin(eng, schema=None):
122    with eng.begin() as conn:
123        if schema is None:
124            # run magic command to get rid of identity sequences
125            # https://floo.bar/2019/11/29/drop-the-underlying-sequence-of-an-identity-column/  # noqa: E501
126            conn.exec_driver_sql("purge recyclebin")
127        else:
128            # per user: https://community.oracle.com/tech/developers/discussion/2255402/how-to-clear-dba-recyclebin-for-a-particular-user  # noqa: E501
129            for owner, object_name, type_ in conn.exec_driver_sql(
130                "select owner, object_name,type from "
131                "dba_recyclebin where owner=:schema and type='TABLE'",
132                {"schema": conn.dialect.denormalize_name(schema)},
133            ).all():
134                conn.exec_driver_sql(f'purge {type_} {owner}."{object_name}"')
135
136
137_all_conns = set()
138
139
140@post_configure_engine.for_db("oracle")
141def _oracle_post_configure_engine(url, engine, follower_ident):
142    from sqlalchemy import event
143
144    @event.listens_for(engine, "checkout")
145    def checkout(dbapi_con, con_record, con_proxy):
146        _all_conns.add(dbapi_con)
147
148    @event.listens_for(engine, "checkin")
149    def checkin(dbapi_connection, connection_record):
150        # work around cx_Oracle issue:
151        # https://github.com/oracle/python-cx_Oracle/issues/530
152        # invalidate oracle connections that had 2pc set up
153        if "cx_oracle_xid" in connection_record.info:
154            connection_record.invalidate()
155
156
157@run_reap_dbs.for_db("oracle")
158def _reap_oracle_dbs(url, idents):
159    log.info("db reaper connecting to %r", url)
160    eng = create_engine(url)
161    with eng.begin() as conn:
162        log.info("identifiers in file: %s", ", ".join(idents))
163
164        to_reap = conn.exec_driver_sql(
165            "select u.username from all_users u where username "
166            "like 'TEST_%' and not exists (select username "
167            "from v$session where username=u.username)"
168        )
169        all_names = {username.lower() for (username,) in to_reap}
170        to_drop = set()
171        for name in all_names:
172            if name.endswith("_ts1") or name.endswith("_ts2"):
173                continue
174            elif name in idents:
175                to_drop.add(name)
176                if "%s_ts1" % name in all_names:
177                    to_drop.add("%s_ts1" % name)
178                if "%s_ts2" % name in all_names:
179                    to_drop.add("%s_ts2" % name)
180
181        dropped = total = 0
182        for total, username in enumerate(to_drop, 1):
183            if _ora_drop_ignore(conn, username):
184                dropped += 1
185        log.info(
186            "Dropped %d out of %d stale databases detected", dropped, total
187        )
188
189
190@follower_url_from_main.for_db("oracle")
191def _oracle_follower_url_from_main(url, ident):
192    url = sa_url.make_url(url)
193    return url.set(username=ident, password="xe")
194
195
196@temp_table_keyword_args.for_db("oracle")
197def _oracle_temp_table_keyword_args(cfg, eng):
198    return {
199        "prefixes": ["GLOBAL TEMPORARY"],
200        "oracle_on_commit": "PRESERVE ROWS",
201    }
202
203
204@set_default_schema_on_connection.for_db("oracle")
205def _oracle_set_default_schema_on_connection(
206    cfg, dbapi_connection, schema_name
207):
208    cursor = dbapi_connection.cursor()
209    cursor.execute("ALTER SESSION SET CURRENT_SCHEMA=%s" % schema_name)
210    cursor.close()
211
212
213@update_db_opts.for_db("oracle")
214def _update_db_opts(db_url, db_opts, options):
215    """Set database options (db_opts) for a test database that we created."""
216    if (
217        options.oracledb_thick_mode
218        and sa_url.make_url(db_url).get_driver_name() == "oracledb"
219    ):
220        db_opts["thick_mode"] = True
221