From 0963b1cf1e8536fd6f5c72b16a1a6ac1ccaabed4 Mon Sep 17 00:00:00 2001
From: HenryNebula <22852427+HenryNebula@users.noreply.github.com>
Date: Wed, 29 Apr 2026 09:31:46 -0400
Subject: [PATCH] Verify: multiple sequential and simultaneous connections work
 (issue #97)

Add HSQLDB integration tests verifying that multiple connect/close
cycles and multiple simultaneous connections work correctly. The
upstream Py4J-era bug (baztian/jaydebeapi#56) about getConnection
overload mismatches does not affect the JPype-based implementation.

Closes #97

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
---
 test/test_integration.py | 38 ++++++++++++++++++++++++++++++++++++++
 1 file changed, 38 insertions(+)

diff --git a/test/test_integration.py b/test/test_integration.py
index c9242f0d..ff4caf1d 100644
--- a/test/test_integration.py
+++ b/test/test_integration.py
@@ -1000,6 +1000,44 @@ def test_rollback_with_autocommit_disabled(self):
         self.conn.rollback()
 
 
+class HsqldbMultipleConnectionsTest(unittest.TestCase):
+    """Test that multiple sequential and simultaneous connections work (issue #97)."""
+
+    def _connect(self, db_name):
+        driver = 'org.hsqldb.jdbcDriver'
+        url = f'jdbc:hsqldb:mem:{db_name}'
+        return jaydebeapiarrow.connect(driver, url, ['SA', ''])
+
+    def test_sequential_connections(self):
+        """Connect, query, close, then connect again — each cycle should succeed."""
+        for i in range(3):
+            conn = self._connect(f'seq_test_{i}')
+            cursor = conn.cursor()
+            cursor.execute("SELECT 1 FROM (VALUES(0))")
+            rows = cursor.fetchall()
+            self.assertEqual(len(rows), 1)
+            self.assertEqual(rows[0][0], 1)
+            cursor.close()
+            conn.close()
+
+    def test_multiple_simultaneous_connections(self):
+        """Multiple open connections at the same time should work independently."""
+        connections = []
+        for i in range(3):
+            conn = self._connect(f'sim_test_{i}')
+            connections.append(conn)
+
+        for conn in connections:
+            cursor = conn.cursor()
+            cursor.execute("SELECT 1 FROM (VALUES(0))")
+            rows = cursor.fetchall()
+            self.assertEqual(len(rows), 1)
+            cursor.close()
+
+        for conn in connections:
+            conn.close()
+
+
 class PostgresTest(IntegrationTestBase, unittest.TestCase):
 
     def connect(self):