wehub-resource-sync
356 lines
12 KiB
Python
356 lines
12 KiB
Python
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 --<category> 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.<category>` anywhere in a module's source (module-level
|
|
# `pytestmark = ...` or per-test `@pytest.mark.<category>` 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} <name>). 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 --<category>):", ""]
|
|
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 "<unknown name>"
|
|
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)
|