Tweaks to ensure that test suite can run cleanly against cloud

databases.

Author: anthony-tuininga

tests/create_schema.py

@@ -47,4 +47,7 @@
                         main_password=test_env.get_main_password(),
                         proxy_user=test_env.get_proxy_user(),
                         proxy_password=test_env.get_proxy_password())
+if test_env.is_on_oracle_cloud(conn):
+    test_env.run_sql_script(conn, "create_schema_cloud",
+                            main_user=test_env.get_main_user())
 print("Done.")

tests/sql/create_schema_cloud.sql

@@ -0,0 +1,36 @@
+/*-----------------------------------------------------------------------------
+ * Copyright (c) 2022, Oracle and/or its affiliates.
+ *
+ * This software is dual-licensed to you under the Universal Permissive License
+ * (UPL) 1.0 as shown at https://oss.oracle.com/licenses/upl and Apache License
+ * 2.0 as shown at http://www.apache.org/licenses/LICENSE-2.0. You may choose
+ * either license.*
+ *
+ * If you elect to accept the software under the Apache License, Version 2.0,
+ * the following applies:
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *---------------------------------------------------------------------------*/
+
+/*-----------------------------------------------------------------------------
+ * create_schema_cloud.sql
+ *
+ * Performs the actual work of creating and populating the schemas with the
+ * database objects used by the python-oracledb test suite. This script is
+ * specific to Oracle Cloud and is only executed when being run against the
+ * Oracle Cloud, after the main script (create_schema.sql) is run. It is
+ * executed by the Python script create_schema.py.
+ *---------------------------------------------------------------------------*/
+
+grant select on v$session to &main_user
+/

tests/test_2700_aq.py

@@ -469,6 +469,8 @@ def test_2719_recipients_list(self):
 
     def test_2720_aq_notification(self):
         "2720 - verify msgid of aq message which spawned notification "
+        if self.is_on_oracle_cloud(self.connection):
+            self.skipTest("AQ notification not supported on the cloud")
         queue = self.get_and_clear_queue(self.book_queue_name,
                                          self.book_type_name)
         condition = threading.Condition()

tests/test_env.py

@@ -104,8 +104,11 @@ def get_admin_connection():
     admin_user = get_value("ADMIN_USER", "Administrative user", "admin")
     admin_password = get_value("ADMIN_PASSWORD", f"Password for {admin_user}",
                                password=True)
-    return oracledb.connect(dsn=get_connect_string(),
-                            params=get_connect_params(),
+    params = get_connect_params()
+    if admin_user and admin_user.upper() == "SYS":
+        params = params.copy()
+        params.set(mode=oracledb.AUTH_MODE_SYSDBA)
+    return oracledb.connect(dsn=get_connect_string(), params=params,
                             user=admin_user, password=admin_password)
 
 def get_charset_ratios():
@@ -217,6 +220,17 @@ def get_external_user():
     if not get_is_thin():
         return get_value("EXTERNAL_USER", "External User")
 
+def is_on_oracle_cloud(connection):
+    server = get_server_version()
+    if server < (18, 0):
+        return False
+    cursor = connection.cursor()
+    cursor.execute("""
+            select sys_context('userenv', 'cloud_service')
+            from dual""")
+    service_name, = cursor.fetchone()
+    return service_name is not None
+
 def run_sql_script(conn, script_name, **kwargs):
     statement_parts = []
     cursor = conn.cursor()
@@ -377,17 +391,9 @@ def get_soda_database(self, minclient=(18, 3), minserver=(18, 0),
         return self.connection.getSodaDatabase()
 
     def is_on_oracle_cloud(self, connection=None):
-        server = get_server_version()
-        if server < (18, 0):
-            return False
         if connection is None:
             connection = self.connection
-        cursor = connection.cursor()
-        cursor.execute("""
-                select sys_context('userenv', 'cloud_service')
-                from dual""")
-        service_name, = cursor.fetchone()
-        return service_name is not None
+        return is_on_oracle_cloud(connection)
 
     def setUp(self):
         if self.requires_connection: