Update rollback & ping & bugfix

Author: GoodManWEN

.github/workflows/python-package.yml

@@ -47,7 +47,7 @@ jobs:
         run: |  # sleep to make sure Oracle server is fully loaded
           sleep 60            
           python -m pip install --upgrade pip
-          python -m pip install flake8 pytest pytest-asyncio
+          python -m pip install flake8 pytest pytest-asyncio async_timeout
           pip install -r requirements.txt
       - name: Lint with flake8
         run: |

cx_Oracle_async/connections.py

@@ -52,11 +52,17 @@ async def gettype(self , *args , **kwargs):
         return await self._loop.run_in_executor(self._thread_pool , self._conn.gettype , *args , **kwargs)
 
     async def commit(self):
-        await self._loop.run_in_executor(self._thread_pool , self._conn.commit)
+        return await self._loop.run_in_executor(self._thread_pool , self._conn.commit)
 
     async def release(self):
         self._pool_wrapper._unoccupied(self._conn)
         return await self._loop.run_in_executor(self._thread_pool , self._pool.release , self._conn)
 
     async def cancel(self):
-        await self._loop.run_in_executor(self._thread_pool , self._conn.cancel)
+        return await self._loop.run_in_executor(self._thread_pool , self._conn.cancel)
+
+    async def ping(self):
+        return await self._loop.run_in_executor(self._thread_pool , self._conn.ping)
+
+    async def rollback(self):
+        return await self._loop.run_in_executor(self._thread_pool , self._conn.rollback)

cx_Oracle_async/pools.py

@@ -49,7 +49,7 @@ def _acquire(self):
         self._occupied.update((_conn , ))
         return AsyncConnectionWrapper(_conn , self._loop , self._thread_pool , self._pool , self)
 
-    def _unoccupied(self , obj):
+    def _unoccupied(self , obj: Connection):
         self._occupied.remove(obj)
 
     async def release(self , conn: Connection):
@@ -58,10 +58,17 @@ async def release(self , conn: Connection):
     async def drop(self , conn: Connection):
         return await self._loop.run_in_executor(self._thread_pool , self._pool.drop , conn)
 
-    async def close(self , force = False):
-        if force:
-            while self._occupied:
-                _conn = self._occupied.pop()
+    async def close(self , force: bool = False , interrupt: bool = False):
+        '''
+        WARNING: option `interrupt` will force cancel all running connections before close
+        the pool. This may cause fetching thread no response forever in some legacy version
+        of oracle database such as 11 or lower.
+
+        Do make sure this option works fine with your working enviornment.
+        '''
+        while self._occupied:
+            _conn = self._occupied.pop()
+            if interrupt:
                 await self._loop.run_in_executor(self._thread_pool , _conn.cancel)
 
         return await self._loop.run_in_executor(self._thread_pool , self._pool.close , force)

tests/test_behavior.py

@@ -24,7 +24,7 @@ async def test_new_table():
             ret = await cursor.fetchone()
             assert ret 
             if ret[0] > 0:
-                await cursor.execute("DTOP TABLE DEPT")
+                await cursor.execute("DROP TABLE DEPT")
 
             sql = f"""
                 CREATE TABLE DEPT
@@ -78,4 +78,4 @@ async def test_usage():
 
     cursor1 = await conn1.cursor()
     async with conn1.cursor() as cursor2:
-        assert type(cursor1) is type(cursor2) 
+        assert type(cursor1) is type(cursor2)

tests/test_drop.py

@@ -3,27 +3,73 @@
 import pytest
 import asyncio
 import time
+from async_timeout import timeout
 from cx_Oracle_async import *
 import cx_Oracle
+import threading
 
 async def create_long_query(oracle_pool):
     async with oracle_pool.acquire() as conn:
         cursor = await conn.cursor()
         try:
-            await cursor.execute("BEGIN  DBMS_LOCK.SLEEP(:a); END;",(20,))
+            await cursor.execute("BEGIN  DBMS_LOCK.SLEEP(:a); END;",(10,))
         except Exception as e:
             assert isinstance(e , cx_Oracle.OperationalError)
 
+def create_long_query_sync(oracle_pool):
+    '''
+    use sync function in order to avoid pytest loop never stop bug.
+    '''
+    try:
+        conn = oracle_pool._pool.acquire()
+        cursor = conn.cursor()
+        cursor.execute("BEGIN  DBMS_LOCK.SLEEP(:a); END;",(10,))
+    except:
+        ...
+
 @pytest.mark.asyncio
 async def test_force_close():
     loop = asyncio.get_running_loop()
     dsn  = makedsn('localhost','1521',sid='xe')
     INAQ = 0.5
-    oracle_pool = await create_pool(user='system',password='oracle',dsn=dsn,max=4)
-    loop = asyncio.get_running_loop()
+    oracle_pool = await create_pool(user='system',password='oracle',dsn=dsn)
     loop.create_task(create_long_query(oracle_pool))
     st_time = time.time()
     await asyncio.sleep(2)
-    await oracle_pool.close(force = True)
+    await oracle_pool.close(force = True , interrupt = False)
+    ed_time = time.time()
+    assert (10 - INAQ) <= (ed_time - st_time) <= (10 + INAQ)
+
+    # test occupy
+    oracle_pool = await create_pool(user='system',password='oracle',dsn=dsn,max=4)
+    conn = await oracle_pool.acquire()
+    conn = await oracle_pool.acquire()
+    assert len(oracle_pool._occupied) == 2
+    conn = await oracle_pool.acquire()
+    assert len(oracle_pool._occupied) == 3
+    st_time = time.time()
+    await asyncio.sleep(2)
+    async with timeout(2):
+        # no running task , return immediately
+        await oracle_pool.close(force = True , interrupt = False)
+    ed_time = time.time()
+    assert (2 - INAQ) <= (ed_time - st_time) <= (2 + INAQ)
+
+    # test interrupt
+    oracle_pool = await create_pool(user='system',password='oracle',dsn=dsn,max=4)
+    st_time = time.time()
+    t = threading.Thread(target = create_long_query_sync , args = (oracle_pool,))
+    t.setDaemon(True)
+    t.start()
+    await asyncio.sleep(2)
+    exception_flag = False
+    try:
+        async with timeout(2):
+            # no response forever
+            await oracle_pool.close(force = True , interrupt = True)
+    except Exception as e:
+        exception_flag = True
+        assert isinstance(e , asyncio.TimeoutError)
     ed_time = time.time()
-    assert (ed_time - st_time) <= (20 - INAQ)
+    assert exception_flag
+    assert (4 - INAQ) <= (ed_time - st_time) <= (10 - INAQ)

tests/test_ping.py

@@ -0,0 +1,27 @@
+import os , sys
+sys.path.append(os.getcwd())
+import pytest
+import asyncio
+from cx_Oracle_async import *
+import cx_Oracle
+
+@pytest.mark.asyncio
+async def test_ping():
+    dsn = makedsn(
+        host = 'localhost', 
+        port = '1521', 
+        sid = 'xe'
+    )
+    async with create_pool(user = 'system',password = 'oracle',dsn = dsn) as oracle_pool:
+        conn = await oracle_pool.acquire()
+        r = await conn.ping()
+        assert r == None
+
+        await conn.release()
+        exception_flag = False
+        try:
+            await conn.ping()
+        except Exception as e:
+            exception_flag = True
+            assert isinstance(e , cx_Oracle.InterfaceError)
+        assert exception_flag

tests/test_rollback.py

@@ -0,0 +1,44 @@
+import os , sys
+sys.path.append(os.getcwd())
+import pytest
+import asyncio
+from cx_Oracle_async import *
+import cx_Oracle
+
+@pytest.mark.asyncio
+async def test_ping():
+    dsn = makedsn(
+        host = 'localhost', 
+        port = '1521', 
+        sid = 'xe'
+    )
+    async with create_pool(user = 'system',password = 'oracle',dsn = dsn) as oracle_pool:
+        async with oracle_pool.acquire() as conn:
+            async with conn.cursor() as cursor:
+                # check if dept exesits
+                await cursor.execute("SELECT COUNT(*) FROM USER_TABLES WHERE TABLE_NAME = UPPER(:a)" , ('DEPT' , ))
+                ret = await cursor.fetchone()
+                assert ret 
+                if ret[0] > 0:
+                    await cursor.execute("DROP TABLE DEPT")
+
+                sql = f"""
+                    CREATE TABLE DEPT
+                       (DEPTNO NUMBER(2) CONSTRAINT PK_DEPT PRIMARY KEY,
+                        DNAME VARCHAR2(14),
+                        LOC VARCHAR2(13) 
+                    )
+                """
+                await cursor.execute(sql)
+
+                await cursor.execute(f"INSERT INTO DEPT(DEPTNO) VALUES (:a)" , (10 , ))
+                await conn.rollback()
+                await cursor.execute(f"INSERT INTO DEPT(DEPTNO) VALUES (:a)" , (12 , ))
+                await conn.commit()
+
+        async with oracle_pool.acquire() as conn:
+            async with conn.cursor() as cursor:
+                await cursor.execute(f"SELECT * FROM DEPT")
+                r = await cursor.fetchall()
+                assert len(r) == 1
+                assert r[0][0] == 12