ci: rework interrupt tests to use a barrier and longer waits

Author: paultiq

tests/fast/api/test_connection_interrupt.py

@@ -1,6 +1,7 @@
 import platform
 import threading
 import time
+from concurrent.futures import ThreadPoolExecutor
 
 import duckdb
 import pytest
@@ -11,20 +12,28 @@ class TestConnectionInterrupt(object):
         condition=platform.system() == "Emscripten",
         reason="threads not allowed on Emscripten",
     )
+    @pytest.mark.timeout(10)
     def test_connection_interrupt(self):
         conn = duckdb.connect()
+        barrier = threading.Barrier(2)
 
-        def interrupt():
-            # Wait for query to start running before interrupting
-            time.sleep(1)
+        def execute_query():
+            barrier.wait()
+            return conn.execute('select * from range(1000000) t1, range(1000000) t2').fetchall()
+
+        def interrupt_query():
+            barrier.wait()
+            time.sleep(2)
             conn.interrupt()
 
-        thread = threading.Thread(target=interrupt)
-        thread.start()
-        with pytest.raises(duckdb.InterruptException):
-            conn.execute('select * from range(100000) t1,range(100000) t2').fetchall()
+        with ThreadPoolExecutor() as executor:
+            query_future = executor.submit(execute_query)
+            interrupt_future = executor.submit(interrupt_query)
+
+            interrupt_future.result()
 
-        thread.join()
+            with pytest.raises((duckdb.InterruptException, duckdb.InvalidInputException)):
+                query_future.result()
 
     def test_interrupt_closed_connection(self):
         conn = duckdb.connect()

tests/fast/api/test_query_interrupt.py

@@ -1,35 +1,32 @@
-import duckdb
-import time
-import pytest
-
 import platform
 import threading
+import time
 import _thread as thread
 
-
-def send_keyboard_interrupt():
-    # Wait a little, so we're sure the 'execute' has started
-    time.sleep(1)
-    # Send an interrupt to the main thread
-    thread.interrupt_main()
+import duckdb
+import pytest
 
 
 class TestQueryInterruption(object):
     @pytest.mark.xfail(
         condition=platform.system() == "Emscripten",
         reason="Emscripten builds cannot use threads",
     )
+    @pytest.mark.timeout(10)
     def test_query_interruption(self):
         con = duckdb.connect()
-        thread = threading.Thread(target=send_keyboard_interrupt)
-        # Start the thread
-        thread.start()
-        try:
-            con.execute('select * from range(100000) t1,range(100000) t2').fetchall()
-        except RuntimeError:
-            # If this is not reached, we could not cancel the query before it completed
-            # indicating that the query interruption functionality is broken
-            assert True
-        except KeyboardInterrupt:
-            pytest.fail()
-        thread.join()
+        barrier = threading.Barrier(2)
+
+        def send_keyboard_interrupt():
+            barrier.wait()
+            time.sleep(2)
+            thread.interrupt_main()
+
+        interrupt_thread = threading.Thread(target=send_keyboard_interrupt)
+        interrupt_thread.start()
+
+        with pytest.raises((KeyboardInterrupt, RuntimeError)):
+            barrier.wait()
+            con.execute('select * from range(1000000) t1,range(1000000) t2').fetchall()
+
+        interrupt_thread.join()