IGNITE-27373 Improve tests

Author: isapego

modules/platforms/python/dbapi/tests/conftest.py

@@ -25,20 +25,26 @@
 
 TEST_PAGE_SIZE = 32
 
+TEST_CONNECT_KWARGS = {
+    "address": server_addresses_basic,
+    "page_size": TEST_PAGE_SIZE,
+    "heartbeat_interval": 2,
+}
+
 @pytest.fixture()
 def table_name(request):
     return f"{request.node.originalname}_{int(time.monotonic_ns())}"
 
 
 @pytest.fixture()
 def connection():
-    conn = pyignite_dbapi.connect(address=server_addresses_basic, page_size=TEST_PAGE_SIZE, heartbeat_interval=2)
+    conn = pyignite_dbapi.connect(**TEST_CONNECT_KWARGS)
     yield conn
     conn.close()
 
 @pytest.fixture()
 def service_connection():
-    conn = pyignite_dbapi.connect(address=server_addresses_basic, page_size=TEST_PAGE_SIZE, heartbeat_interval=2)
+    conn = pyignite_dbapi.connect(**TEST_CONNECT_KWARGS)
     yield conn
     conn.close()
 

modules/platforms/python/dbapi/tests/test_concurrency.py

@@ -19,10 +19,9 @@
 import pytest
 
 import pyignite_dbapi
-from tests.util import server_addresses_basic
+from tests.conftest import TEST_CONNECT_KWARGS
 
 
-CONNECT_KWARGS = {"address": server_addresses_basic}
 NUM_THREADS = 50
 
 
@@ -77,7 +76,7 @@ def task(_):
 
 def test_concurrent_connect_use_close(module_level_threadsafety):
     def task(_):
-        c = pyignite_dbapi.connect(**CONNECT_KWARGS)
+        c = pyignite_dbapi.connect(**TEST_CONNECT_KWARGS)
         with c.cursor() as cur:
             cur.execute("SELECT 1")
             assert cur.fetchone() is not None
@@ -118,7 +117,7 @@ def test_concurrent_commit_and_rollback(table, module_level_threadsafety):
     lock = threading.Lock()
 
     def task(thread_id):
-        with pyignite_dbapi.connect(**CONNECT_KWARGS) as conn:
+        with pyignite_dbapi.connect(**TEST_CONNECT_KWARGS) as conn:
             conn.autocommit = False
             with conn.cursor() as cur:
                 cur.execute(f"INSERT INTO {table} (id, data) VALUES (?, ?)", (thread_id, "x"))
@@ -133,7 +132,7 @@ def task(thread_id):
 
     time.sleep(3.0)
 
-    with pyignite_dbapi.connect(**CONNECT_KWARGS) as conn:
+    with pyignite_dbapi.connect(**TEST_CONNECT_KWARGS) as conn:
         with conn.cursor() as cur:
             cur.execute(f"SELECT id FROM {table} ORDER BY id")
             found_ids = {row[0] for row in cur.fetchall()}