import asyncio import dataclasses 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, add_realtime_option as _add_realtime_option, event_loop_policy, register_marker as _register_virtual_time_marker, ) TEST_CONNECT_OPTIONS = dataclasses.replace(DEFAULT_API_CONNECT_OPTIONS, retry_interval=0.0) # Category names are the pytest markers declared in pyproject.toml — the single # source of truth. They're read from the file rather than via config.getini() # because the -- options are registered in pytest_addoption, which runs # before pytest has parsed the ini file. _PYPROJECT = Path(__file__).resolve().parents[1] / "pyproject.toml" def _load_categories() -> tuple[str, ...]: block = re.search(r"(?ms)^markers\s*=\s*\[(.*?)^\s*\]", _PYPROJECT.read_text(encoding="utf-8")) names = re.findall(r'^\s*"(\w+)', block.group(1), re.MULTILINE) if block else [] if not names: raise RuntimeError(f"no pytest markers found in {_PYPROJECT}") return tuple(dict.fromkeys(names)) CATEGORIES = _load_categories() # matches `pytest.mark.` anywhere in a module's source (module-level # `pytestmark = ...` or per-test `@pytest.mark.` decorators). _CATEGORY_RE = re.compile(r"pytest\.mark\.(" + "|".join(CATEGORIES) + r")\b") # a module "has tests" if it declares any (non-commented) test function or Test class. _HAS_TESTS_RE = re.compile(r"^\s*(?:async\s+)?def test|^\s*class Test", re.MULTILINE) _CATEGORY_HINT = ( "Every test module must declare its category with a module-level marker, one of:\n\n" + "\n".join(f" pytestmark = pytest.mark.{c}" for c in CATEGORIES) + "\n\nRun pytest with the --allow-uncategorized option to temporarily disable this rule." ) def pytest_addoption(parser: pytest.Parser) -> None: _add_realtime_option(parser) group = parser.getgroup("categories", "test category selection") for category in CATEGORIES: group.addoption( f"--{category}", nargs="?", const=True, default=None, metavar="PROVIDER", help=f"select only `{category}` tests; optionally narrow to a single " f"provider/target (e.g. --{category} ). Repeatable categories are unioned.", ) group.addoption( "--list-categories", action="store_true", default=False, help="list every test module grouped by category, then exit (no tests run).", ) group.addoption( "--allow-uncategorized", action="store_true", default=False, help="allow test modules without a category marker, temporarily disabling " "the categorization rule (CI keeps it on by default).", ) class _ModuleFacts(NamedTuple): categories: frozenset[str] has_tests: bool @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: return f"{n} {noun}" if n == 1 else f"{n} {noun}s" def _iter_test_files(config: pytest.Config) -> Iterator[Path]: rootdir = Path(str(config.rootdir)) for testpath in config.getini("testpaths") or ["tests"]: base = rootdir / testpath if base.is_dir(): yield from sorted(base.rglob("test_*.py")) elif base.is_file() and base.name.startswith("test_"): yield base 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): facts = _module_facts(path) if not facts.has_tests: continue # empty / fully-commented module: nothing to categorize 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): facts = _module_facts(path) if not facts.has_tests: continue rel = str(path.relative_to(rootdir)) if not facts.categories: uncategorized.append(rel) for cat in facts.categories: by_category[cat].append(rel) lines = ["", "Test categories (select with --):", ""] for category in CATEGORIES: modules = by_category[category] lines.append(f" {category:<8} {_plural(len(modules), 'module')}") lines.extend(f" - {rel}" for rel in modules) if uncategorized: blocked = _plural(len(uncategorized), "module") lines.append(f"\n UNCATEGORIZED {blocked} — run is blocked:") lines.extend(f" - {rel}" for rel in uncategorized) pytest.exit("\n".join(lines), returncode=0) def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None: """Enforce category markers, then apply provider-argument filtering.""" if not config.getoption("--allow-uncategorized"): offenders = _uncategorized_modules(config) if offenders: rootdir = Path(str(config.rootdir)) listed = "\n".join(f" - {p.relative_to(rootdir)}" for p in offenders) raise pytest.UsageError( f"Found {_plural(len(offenders), 'test module')} without a " f"category marker:\n\n{listed}\n\n{_CATEGORY_HINT}" ) _filter_by_provider(config, items) def _selected_categories(config: pytest.Config) -> dict[str, object]: # category -> True (whole category) or a provider string (single target) selected: dict[str, object] = {} for category in CATEGORIES: value = config.getoption(f"--{category}") if value is not None: selected[category] = value return selected def pytest_ignore_collect(collection_path: Path, config: pytest.Config) -> bool | None: """Skip non-selected modules *before* importing them. Marker selection alone can't do this: pytest must import a module to read its markers, and some modules fail to import without optional/cloud deps. So we detect the category marker by a plain substring scan of the source instead. """ selected = _selected_categories(config) if not selected: return None if collection_path.is_dir() or collection_path.suffix != ".py": return None if not collection_path.name.startswith("test_"): return None if _module_facts(collection_path).categories.intersection(selected): return None return True def _filter_by_provider(config: pytest.Config, items: list[pytest.Item]) -> None: """Apply provider-argument filtering once modules are imported. `--plugin anthropic` keeps only items whose `plugin` marker carries "anthropic" in its args; a bare `--plugin` keeps the whole category. """ selected = _selected_categories(config) if not selected: return kept, deselected = [], [] for item in items: keep = False for category, target in selected.items(): for marker in item.iter_markers(category): if target is True or target in marker.args: keep = True break if keep: break (kept if keep else deselected).append(item) if deselected: config.hook.pytest_deselected(items=deselected) items[:] = kept @pytest.fixture async def job_process(): utils.http_context._new_session_ctx() yield await utils.http_context._close_http_ctx() @pytest.fixture(autouse=True) def configure_test(): log._silence_noisy_loggers() @pytest.fixture(scope="session", autouse=True) def _logging_baseline(): """Pristine logging state, snapshotted once before any test runs.""" loggers = [logging.getLogger()] # root loggers += [ logging.getLogger(name) for name in logging.root.manager.loggerDict if name.startswith("livekit") ] return [(logger, logger.level, logger.handlers[:], logger.propagate) for logger in loggers] @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 for logger, level, handlers, propagate in _logging_baseline: logger.setLevel(level) logger.handlers[:] = handlers logger.propagate = propagate @pytest.fixture def toxiproxy(): toxiproxy = Toxiproxy() yield toxiproxy if toxiproxy.running(): toxiproxy.destroy_all() @pytest.fixture() def logger(): logger = logging.getLogger("livekit.tests") logger.setLevel(logging.DEBUG) return logger def _is_ignorable_task(task) -> bool: try: coro = task.get_coro() # async_generator_athrow tasks are created by Python's GC when finalizing # async generators (e.g. httpx/httpcore streaming generators) coro_name = getattr(coro, "__qualname__", "") or type(coro).__name__ if "async_generator_athrow" in coro_name: return True mod = getattr(coro, "__module__", "") if "pytest" in mod or "pytest_asyncio" in mod: return True for frame in task.get_stack(): if "pytest" in frame.f_code.co_filename or "pytest_asyncio" in frame.f_code.co_filename: return True except Exception: pass return False def format_task(task) -> str: try: name = task.get_name() if hasattr(task, "get_name") else "" coro = task.get_coro() coro_name = getattr(coro, "__qualname__", None) or type(coro).__name__ frame = getattr(coro, "cr_frame", None) if frame: location = f"{frame.f_code.co_filename}:{frame.f_lineno}" else: location = "no frame available" return ( f"Task Name : {name}\n" f"Coroutine : {coro_name}\n" f"Location : {location}\n" f"State : {'pending' if not task.done() else 'done'}" ) except Exception: return repr(task) @pytest.fixture(autouse=True) 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 tasks_after = set(asyncio.all_tasks()) leaked_tasks = [ task for task in tasks_after - tasks_before if not task.done() and 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)