tests: speed up unit tests by running them concurrently

pyproject.toml

@@ -86,9 +86,11 @@ dev = [
     "python-dotenv>=1.0.1",
     "mypy",
     "scikit-learn",
-    "pytest",
+    "pytest>=9.0.3,<9.1",
     "ruff",
     "pytest-asyncio>=0.25.3",
+    "pytest-asyncio-concurrent==0.5.2",
+    "async-solipsism>=0.9",
     "jiwer>=3.1.0",
     "scipy>=1.13.1",
     "tiktoken>=0.9.0",
@@ -125,7 +127,13 @@ convention = "google"
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
 asyncio_default_fixture_loop_scope = "function"
-addopts = ["--import-mode=importlib", "--ignore=examples"]
+filterwarnings = [
+    # virtual-time tests override the event_loop_policy fixture (tests/virtual_time.py) to run on
+    # an async-solipsism loop; pytest-asyncio deprecates the override but offers no per-test
+    # alternative that returns a default loop for unmarked tests without erroring.
+    "ignore:Overriding the \"event_loop_policy\" fixture is deprecated:Warning",
+]
+addopts = ["--import-mode=importlib", "--ignore=examples", "-p", "tests.concurrency"]
 pythonpath = ["."]
 testpaths = ["tests"]
 markers = [

tests/conftest.py

@@ -3,14 +3,22 @@
 import logging
 import re
 from collections.abc import Iterator
+from functools import cache
 from pathlib import Path
+from typing import NamedTuple
 
 import pytest
 
 from livekit.agents import DEFAULT_API_CONNECT_OPTIONS, utils
 from livekit.agents.cli import log
 
+from . import concurrency
 from .toxic_proxy import Toxiproxy
+from .virtual_time import (  # noqa: F401  (re-exported so pytest discovers the fixtures)
+    _virtual_wall_clock,
+    event_loop_policy,
+    register_marker as _register_virtual_time_marker,
+)
 
 TEST_CONNECT_OPTIONS = dataclasses.replace(DEFAULT_API_CONNECT_OPTIONS, retry_interval=0.0)
 
@@ -71,15 +79,25 @@ def pytest_addoption(parser: pytest.Parser) -> None:
     )
 
 
-def _read_source(path: Path) -> str:
-    try:
-        return path.read_text(encoding="utf-8")
-    except OSError:
-        return ""
+class _ModuleFacts(NamedTuple):
+    categories: frozenset[str]
+    has_tests: bool
 
 
-def _module_categories(source: str) -> set[str]:
-    return set(_CATEGORY_RE.findall(source))
+@cache
+def _module_facts(path: Path) -> _ModuleFacts:
+    """
+    Retrieve information about the module without importing it.
+    Cache to avoid redundant filesystem reads.
+    """
+    try:
+        source = path.read_text(encoding="utf-8")
+    except OSError:
+        return _ModuleFacts(frozenset(), False)
+    return _ModuleFacts(
+        frozenset(_CATEGORY_RE.findall(source)),
+        _HAS_TESTS_RE.search(source) is not None,
+    )
 
 
 def _plural(n: int, noun: str) -> str:
@@ -100,31 +118,33 @@ def _uncategorized_modules(config: pytest.Config) -> list[Path]:
     """Test files that contain tests but declare no category marker."""
     offenders: list[Path] = []
     for path in _iter_test_files(config):
-        source = _read_source(path)
-        if not _HAS_TESTS_RE.search(source):
+        facts = _module_facts(path)
+        if not facts.has_tests:
             continue  # empty / fully-commented module: nothing to categorize
-        if not _module_categories(source):
+        if not facts.categories:
             offenders.append(path)
     return offenders
 
 
 def pytest_configure(config: pytest.Config) -> None:
     """Handle `--list-categories` before any collection/import happens."""
+    _module_facts.cache_clear()
+    _register_virtual_time_marker(config)
+
     if not config.getoption("--list-categories"):
         return
 
     rootdir = Path(str(config.rootdir))
     by_category: dict[str, list[str]] = {c: [] for c in CATEGORIES}
     uncategorized: list[str] = []
     for path in _iter_test_files(config):
-        source = _read_source(path)
-        if not _HAS_TESTS_RE.search(source):
+        facts = _module_facts(path)
+        if not facts.has_tests:
             continue
         rel = str(path.relative_to(rootdir))
-        cats = _module_categories(source)
-        if not cats:
+        if not facts.categories:
             uncategorized.append(rel)
-        for cat in cats:
+        for cat in facts.categories:
             by_category[cat].append(rel)
 
     lines = ["", "Test categories (select with --<category>):", ""]
@@ -178,16 +198,9 @@ def pytest_ignore_collect(collection_path: Path, config: pytest.Config) -> bool
         return None
     if not collection_path.name.startswith("test_"):
         return None
-
-    try:
-        source = collection_path.read_text(encoding="utf-8")
-    except OSError:
+    if _module_facts(collection_path).categories.intersection(selected):
         return None
 
-    # ignore the module only if it carries none of the selected categories' markers;
-    # otherwise return None (defer) so --ignore and other plugins still apply.
-    if any(f"pytest.mark.{category}" in source for category in selected):
-        return None
     return True
 
 
@@ -242,7 +255,7 @@ def _logging_baseline():
     return [(logger, logger.level, logger.handlers[:], logger.propagate) for logger in loggers]
 
 
-@pytest.fixture(autouse=True)
+@pytest.fixture(scope="module", autouse=True)
 def _restore_logging(_logging_baseline):
     """Revert global logging a test mutated (e.g. via cli.log.setup_logging)."""
     yield
@@ -307,7 +320,22 @@ def format_task(task) -> str:
 
 
 @pytest.fixture(autouse=True)
-async def fail_on_leaked_tasks():
+async def fail_on_leaked_tasks(request):
+    if concurrency.is_concurrent_member(request.node):
+        # Concurrent group members share one event loop, so we can't diff asyncio.all_tasks()
+        # (it would flag other still-running tests). Instead ask which still-pending tasks were
+        # created by *this* test -- tagged via contextvars, see tests/concurrency.py.
+        yield
+        leaked_tasks = [
+            task
+            for task in concurrency.owned_pending_tasks(request.node.nodeid)
+            if not _is_ignorable_task(task)
+        ]
+        if leaked_tasks:
+            tasks_msg = "\n\n".join(format_task(task) for task in leaked_tasks)
+            pytest.fail("Test leaked tasks:\n\n" + tasks_msg)
+        return
+
     tasks_before = set(asyncio.all_tasks())
 
     yield

tests/test_agent_session.py

@@ -37,7 +37,7 @@
 
 from .fake_session import FakeActions, create_session, run_session
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.virtual_time, pytest.mark.no_concurrent]
 
 
 class MyAgent(Agent):

tests/test_aio.py

@@ -4,7 +4,7 @@
 
 from livekit.agents.utils import aio
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.concurrent]
 
 
 async def test_channel():

tests/test_aio_itertools.py

@@ -4,7 +4,7 @@
 
 from livekit.agents.utils.aio.itertools import Tee
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.virtual_time, pytest.mark.no_concurrent]
 
 
 async def _async_iter(items):

tests/test_amd_classifier.py

@@ -21,7 +21,7 @@
 
 from .fake_llm import FakeLLM, FakeLLMResponse
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.virtual_time, pytest.mark.no_concurrent]
 
 
 def _make_classifier(

tests/test_audio_decoder.py

@@ -17,6 +17,10 @@
 
 pytestmark = pytest.mark.unit
 
+# Decodes audio on background threads / executors with blocking waits; it deadlocks when forced
+# to share one event loop with other tests.
+pytestmark = pytest.mark.no_concurrent
+
 TEST_AUDIO_FILEPATH = os.path.join(os.path.dirname(__file__), "change-sophie.opus")
 
 

tests/test_audio_emitter.py

@@ -11,7 +11,7 @@
 from livekit.agents import tts, utils
 from livekit.agents.types import USERDATA_TIMED_TRANSCRIPT, TimedString
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.concurrent]
 
 
 def _make_pcm(sample_rate: int, num_channels: int, duration_ms: int) -> bytes:

tests/test_audio_recognition_aclose.py

@@ -15,7 +15,7 @@
 
 from livekit.agents.voice.audio_recognition import AudioRecognition
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.virtual_time, pytest.mark.no_concurrent]
 
 
 class TestAudioRecognitionAclose:

tests/test_audio_recognition_handoff.py

@@ -10,7 +10,7 @@
 from livekit.agents.voice.agent import ModelSettings
 from livekit.agents.voice.agent_activity import AgentActivity
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.concurrent]
 
 
 def _make_activity(agent: Agent, stt: object) -> MagicMock:

tests/test_chat_ctx.py

@@ -14,7 +14,7 @@
 
 from .fake_llm import FakeLLM, FakeLLMResponse
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.concurrent]
 
 
 def ai_function1(a: int, b: str = "default") -> None:

tests/test_connection_pool.py

@@ -4,7 +4,7 @@
 
 from livekit.agents.utils import ConnectionPool
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.virtual_time, pytest.mark.no_concurrent]
 
 
 class DummyConnection:

tests/test_debounce.py

@@ -4,7 +4,7 @@
 
 from livekit.agents.utils.aio.debounce import Debounced, debounced
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.virtual_time, pytest.mark.no_concurrent]
 
 
 class TestDebounce:

tests/test_drain_timeout.py

@@ -27,7 +27,7 @@
 from livekit.agents.utils import aio
 from livekit.agents.worker import AgentServer
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.virtual_time, pytest.mark.no_concurrent]
 
 _CLI_ARGS = CliArgs(log_level="ERROR", url=None, api_key=None, api_secret=None)
 

tests/test_filler.py

@@ -9,7 +9,7 @@
 from livekit.agents.voice.events import AgentStateChangedEvent, UserStateChangedEvent
 from livekit.agents.voice.filler_scheduler import _FillerScheduler
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.virtual_time, pytest.mark.no_concurrent]
 
 
 class _FakeSpeechHandle:

tests/test_http_context_helper.py

@@ -12,7 +12,7 @@
 from livekit.agents import inference
 from livekit.agents.utils import http_context
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.concurrent]
 
 
 async def test_open_yields_working_session_and_closes_on_exit() -> None:

tests/test_interruption/test_interruption_failover.py

@@ -28,7 +28,7 @@
 )
 from livekit.agents.types import APIConnectOptions
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.concurrent]
 
 MAX_RETRY = 2
 CONN_OPTIONS = APIConnectOptions(max_retry=MAX_RETRY, retry_interval=0.0, timeout=1.0)

tests/test_ipc.py

@@ -23,6 +23,10 @@
 
 pytestmark = pytest.mark.unit
 
+# Drives real subprocesses / IPC and installs process-global signal handlers; it deadlocks and
+# raises _ExitCli when forced to share one event loop with other tests.
+pytestmark = pytest.mark.no_concurrent
+
 
 @dataclass
 class EmptyMessage:

tests/test_langgraph.py

@@ -16,7 +16,7 @@
 from livekit.agents.llm import ChatContext
 from livekit.plugins.langchain import LLMAdapter
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.concurrent]
 
 # --- State definitions ---
 

tests/test_nested_agent_task.py

@@ -9,7 +9,7 @@
 
 from .fake_llm import FakeLLM, FakeLLMResponse
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.virtual_time, pytest.mark.no_concurrent]
 
 
 class InnerTask(AgentTask):

tests/test_openai_realtime_chat_ctx.py

@@ -7,7 +7,7 @@
 from livekit.agents.llm import remote_chat_context
 from livekit.plugins.openai.realtime.realtime_model import RealtimeSession
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.concurrent]
 
 
 def _create_session() -> RealtimeSession:

tests/test_recording.py

@@ -27,7 +27,7 @@
 from .fake_tts import FakeTTS
 from .fake_vad import FakeVAD
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.concurrent]
 
 # ---------------------------------------------------------------------------
 # Shared helpers

tests/test_room.py

@@ -32,7 +32,7 @@
     wait_for_event,
 )
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.concurrent]
 
 TIMEOUT = 5.0
 

tests/test_room_io.py

@@ -16,7 +16,7 @@
 from livekit.agents.voice.room_io.room_io import RoomIO
 from livekit.agents.voice.room_io.types import NoiseCancellationParams
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.virtual_time, pytest.mark.no_concurrent]
 
 # -- helpers ------------------------------------------------------------------
 

tests/test_schema_gemini.py

@@ -6,7 +6,7 @@
 
 from livekit.plugins.google import utils
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.concurrent]
 
 #  Gemini Schema Tests
 

tests/test_session_host.py

@@ -32,7 +32,7 @@
 )
 from livekit.protocol.agent_pb import agent_session as agent_pb
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.virtual_time, pytest.mark.no_concurrent]
 
 # ---------------------------------------------------------------------------
 # In-memory transport for testing