Merge pull request #281 raise exception if aquire session from stopped pool

Author: rekby

tests/aio/test_session_pool.py

@@ -62,14 +62,14 @@ async def test_close_basic_logic_case_1(driver):
     waiter = asyncio.ensure_future(pool.acquire())
 
     await pool.stop()
-    waiter_sess = waiter.result()
-    assert not waiter_sess.initialized()
-    after_stop = await pool.acquire()
-    assert not after_stop.initialized()
+
+    with pytest.raises(ValueError):
+        waiter.result()
+
+    with pytest.raises(ValueError):
+        await pool.acquire()
 
     await pool.release(s)
-    await pool.release(after_stop)
-    await pool.release(waiter_sess)
     assert pool._active_count == 0
 
 
@@ -106,9 +106,9 @@ async def test_close_basic_logic_case_2(driver):
 
     assert pool._active_count == 0
 
-    sess = await pool.acquire()
+    with pytest.raises(ValueError):
+        await pool.acquire()
 
-    assert not sess.initialized()
     await pool.stop()
 
 

tests/aio/test_table.py

@@ -0,0 +1,12 @@
+import pytest
+import ydb.aio
+
+
+@pytest.mark.asyncio
+class TestSessionPool:
+    async def test_checkout_from_stopped_pool(self, driver):
+        pool = ydb.aio.SessionPool(driver, 1)
+        await pool.stop()
+
+        with pytest.raises(ValueError):
+            await pool.acquire()

tests/session_pool.py

@@ -0,0 +1,40 @@
+import pytest
+
+import ydb
+
+
+def test_close_basic_logic_case_1(driver_sync):
+    pool = ydb.SessionPool(driver_sync, 1)
+    s = pool.acquire()
+
+    pool.stop()
+
+    with pytest.raises(ValueError):
+        pool.acquire()
+
+    pool.release(s)
+    assert pool._pool_impl._active_count == 0
+
+
+def test_close_basic_logic_case_2(driver_sync):
+    pool = ydb.SessionPool(driver_sync, 10)
+    acquired = []
+
+    for _ in range(10):
+        acquired.append(pool.acquire())
+
+    for _ in range(3):
+        pool.release(acquired.pop(-1))
+
+    pool.stop()
+    assert pool._pool_impl._active_count == 7
+
+    while acquired:
+        pool.release(acquired.pop(-1))
+
+    assert pool._pool_impl._active_count == 0
+
+    with pytest.raises(ValueError):
+        pool.acquire()
+
+    pool.stop()

tests/table/table_test.py

@@ -1,6 +1,16 @@
+import pytest
 import ydb
 
 
+class TestSessionPool:
+    def test_checkout_from_stopped_pool(self, driver_sync):
+        pool = ydb.SessionPool(driver_sync, 1)
+        pool.stop()
+
+        with pytest.raises(ValueError):
+            pool.acquire()
+
+
 class TestTable:
     def test_create_table_with_not_null_primary_key_by_api(self, driver_sync, database):
         table_path = database + "/test_table"

ydb/_sp_impl.py

@@ -335,6 +335,10 @@ def _on_keep_alive(self, session, f):
             self._destroy(session, "keep-alive-error")
 
     def acquire(self, blocking=True, timeout=None):
+        if self._should_stop.is_set():
+            self._logger.error("Take session from closed session pool")
+            raise ValueError("Take session from closed session pool.")
+
         waiter = self.subscribe()
         has_result = False
         if blocking:

ydb/aio/table.py

@@ -341,8 +341,8 @@ async def _get_session_from_queue(self, timeout: float):
     async def acquire(self, timeout: float = None, retry_timeout: float = None, retry_num: int = None) -> ydb.ISession:
 
         if self._should_stop.is_set():
-            self._logger.debug("Acquired not inited session")
-            return self._create()
+            self._logger.error("Take session from closed session pool")
+            raise ValueError("Take session from closed session pool.")
 
         if retry_timeout is None:
             retry_timeout = timeout