fix(tests): run FFI dispose test in a subprocess and add test timeouts (#931)

Author: cloudwebrtc

.github/workflows/tests.yml

@@ -31,6 +31,7 @@ jobs:
   test:
     name: Test (${{ inputs.os }}, Python ${{ inputs.python-version }})
     runs-on: ${{ inputs.os }}
+    timeout-minutes: 45
     steps:
       - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
         with:
@@ -100,7 +101,7 @@ jobs:
           uv venv .test-venv
           source .test-venv/bin/activate
           uv pip install rtc-wheel/*.whl ./livekit-api ./livekit-protocol
-          uv pip install pytest pytest-asyncio numpy matplotlib
+          uv pip install pytest pytest-asyncio pytest-timeout numpy matplotlib
 
       - name: Create venv and install dependencies (macOS)
         if: runner.os == 'macOS'
@@ -109,15 +110,15 @@ jobs:
           source .test-venv/bin/activate
           uv pip install "${{ steps.select-wheel-macos.outputs.wheel }}"
           uv pip install ./livekit-api ./livekit-protocol
-          uv pip install pytest pytest-asyncio numpy matplotlib
+          uv pip install pytest pytest-asyncio pytest-timeout numpy matplotlib
 
       - name: Create venv and install dependencies (Windows)
         if: runner.os == 'Windows'
         run: |
           uv venv .test-venv
           $wheel = (Get-ChildItem rtc-wheel\*.whl)[0].FullName
           uv pip install --python .test-venv $wheel .\livekit-api .\livekit-protocol
-          uv pip install --python .test-venv pytest pytest-asyncio numpy matplotlib
+          uv pip install --python .test-venv pytest pytest-asyncio pytest-timeout numpy matplotlib
         shell: pwsh
 
       - name: Run tests (Unix)
@@ -128,14 +129,14 @@ jobs:
           LIVEKIT_API_SECRET: ${{ secrets.LIVEKIT_API_SECRET }}
         run: |
           source .test-venv/bin/activate
-          pytest tests/ livekit-rtc/tests/
+          pytest --timeout=300 tests/ livekit-rtc/tests/
 
       - name: Run tests (Windows)
         if: runner.os == 'Windows'
         env:
           LIVEKIT_URL: ${{ secrets.LIVEKIT_URL }}
           LIVEKIT_API_KEY: ${{ secrets.LIVEKIT_API_KEY }}
           LIVEKIT_API_SECRET: ${{ secrets.LIVEKIT_API_SECRET }}
-        run: .test-venv\Scripts\python.exe -m pytest tests/ livekit-rtc/tests/
+        run: .test-venv\Scripts\python.exe -m pytest --timeout=300 tests/ livekit-rtc/tests/
         shell: pwsh
 

livekit-rtc/tests/test_dc.py

@@ -308,10 +308,13 @@ async def test_publish_data_makes_no_ctypes_array_type() -> None:
             lambda: len(collector.packets) >= len(payloads),
             message="receiver did not get all payloads",
         )
+        # Check `type(o)` instead of `isinstance(o, type)`: gc can hand back a
+        # dead `weakref.proxy`, and isinstance() dereferences it and raises
+        # ReferenceError (seen on Windows ARM64).
         array_types = [
             o
             for o in gc.get_objects()
-            if isinstance(o, type)
+            if issubclass(type(o), type)
             and issubclass(o, ctypes.Array)
             and getattr(o, "_length_", None) in lengths
         ]

livekit-rtc/tests/test_ffi_handle_lifecycle.py

@@ -14,26 +14,12 @@
 
 from __future__ import annotations
 
-import gc
 import subprocess
 import sys
 
 import pytest
 
 from livekit.rtc._ffi_client import FfiClient, FfiHandle
-from livekit.rtc._proto import audio_frame_pb2, ffi_pb2
-
-
-def _allocate_real_handle() -> int:
-    req = ffi_pb2.FfiRequest()
-    req.new_audio_source.type = audio_frame_pb2.AUDIO_SOURCE_NATIVE
-    req.new_audio_source.sample_rate = 48000
-    req.new_audio_source.num_channels = 1
-    req.new_audio_source.queue_size_ms = 100
-    resp = FfiClient.instance.request(req)
-    handle_id = resp.new_audio_source.source.handle.id
-    assert handle_id != 0
-    return int(handle_id)
 
 
 def test_active_runtime_invalid_handle_asserts() -> None:
@@ -42,26 +28,57 @@ def test_active_runtime_invalid_handle_asserts() -> None:
         FfiHandle(999999).dispose()
 
 
-def test_live_handles_marked_consumed_and_no_assert_after_dispose() -> None:
-    handle_id = _allocate_real_handle()
-    handle = FfiHandle(handle_id)
-    assert not handle.disposed
+# Runs in a fresh interpreter: FfiClient._dispose() tears down the process-wide
+# FFI runtime, and every later test in the same pytest process would then wait
+# forever for FFI callbacks that never arrive.
+_DISPOSE_MARKS_LIVE_HANDLES_CONSUMED = """
+import gc
+
+from livekit.rtc._ffi_client import FfiClient, FfiHandle
+from livekit.rtc._proto import audio_frame_pb2, ffi_pb2
+
+req = ffi_pb2.FfiRequest()
+req.new_audio_source.type = audio_frame_pb2.AUDIO_SOURCE_NATIVE
+req.new_audio_source.sample_rate = 48000
+req.new_audio_source.num_channels = 1
+req.new_audio_source.queue_size_ms = 100
+resp = FfiClient.instance.request(req)
+handle_id = resp.new_audio_source.source.handle.id
+assert handle_id != 0
+
+handle = FfiHandle(handle_id)
+assert not handle.disposed
 
-    # Global dispose marks all live handles consumed and sets _ffi_disposed
-    FfiClient.instance._dispose()
-    assert handle.disposed
+# Global dispose marks all live handles consumed and sets _ffi_disposed
+FfiClient.instance._dispose()
+assert handle.disposed
 
-    # Explicit dispose or __del__ after global dispose should be a no-op and not assert
-    handle.dispose()
-    handle.__del__()
+# Explicit dispose or __del__ after global dispose should be a no-op and not assert
+handle.dispose()
+handle.__del__()
+
+# Newly instantiated handle after global dispose also safely no-ops without assert
+late_handle = FfiHandle(12345)
+late_handle.dispose()
+
+del handle
+del late_handle
+gc.collect()
+print("ok")
+"""
 
-    # Newly instantiated handle after global dispose also safely no-ops without assert
-    late_handle = FfiHandle(12345)
-    late_handle.dispose()
 
-    del handle
-    del late_handle
-    gc.collect()
+def test_live_handles_marked_consumed_and_no_assert_after_dispose() -> None:
+    result = subprocess.run(
+        [sys.executable, "-c", _DISPOSE_MARKS_LIVE_HANDLES_CONSUMED],
+        capture_output=True,
+        text=True,
+        timeout=60,
+    )
+    assert result.returncode == 0, result.stderr
+    assert result.stdout.strip() == "ok"
+    assert "AssertionError" not in result.stderr
+    assert "Exception ignored" not in result.stderr
 
 
 def test_subprocess_exit_with_undisposed_handle_has_no_assertion_error() -> None:
@@ -85,6 +102,7 @@ def test_subprocess_exit_with_undisposed_handle_has_no_assertion_error() -> None
         [sys.executable, "-c", code],
         capture_output=True,
         text=True,
+        timeout=60,
     )
     assert result.returncode == 0
     assert "AssertionError" not in result.stderr