From 65e06e0e0c7435f8d07c53d935bb7471b987861f Mon Sep 17 00:00:00 2001 From: cloudwebrtc Date: Fri, 30 Oct 2026 09:12:15 +0800 Subject: [PATCH 1/2] fix(tests): add missing type annotations in test_basic.py mypy (Check Types) failed on main after #641 with no-untyped-def, no-untyped-call and untyped-decorator errors. Annotate skip_if_no_credentials() like tests/utils.py does, and type the event callback's *args/**kwargs. Co-Authored-By: Claude Opus 5.5 --- livekit-rtc/tests/test_basic.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/livekit-rtc/tests/test_basic.py b/livekit-rtc/tests/test_basic.py index 8a67f7f0..a52b1db9 100644 --- a/livekit-rtc/tests/test_basic.py +++ b/livekit-rtc/tests/test_basic.py @@ -14,7 +14,7 @@ import asyncio import os import uuid -from typing import Callable, Optional +from typing import Any, Callable, Optional import pytest @@ -26,7 +26,7 @@ WAIT_INTERVAL = 0.1 -def skip_if_no_credentials(): +def skip_if_no_credentials() -> pytest.MarkDecorator: required_vars = ["LIVEKIT_URL", "LIVEKIT_API_KEY", "LIVEKIT_API_SECRET"] missing = [var for var in required_vars if not os.getenv(var)] return pytest.mark.skipif( @@ -108,7 +108,7 @@ def _expect_event( loop = asyncio.get_event_loop() fut: asyncio.Future = loop.create_future() - def _on_event(*args, **kwargs) -> None: + def _on_event(*args: Any, **kwargs: Any) -> None: if fut.done(): return if predicate is None or predicate(*args, **kwargs): From 1e591db9765f5fe4b211affdaff5db7ad1a568ec Mon Sep 17 00:00:00 2001 From: cloudwebrtc Date: Fri, 30 Oct 2026 09:16:50 +0800 Subject: [PATCH 2/2] refactor(tests): reuse utils helpers in test_basic.py Replace the local copies of skip_if_no_credentials, create_token, unique_room_name and the wait/connect/event helpers with the shared ones in tests/utils.py. expect_room_event now unregisters its handler once the event fires. Co-Authored-By: Claude Opus 5.5 --- livekit-rtc/tests/test_basic.py | 155 ++++++-------------------------- 1 file changed, 29 insertions(+), 126 deletions(-) diff --git a/livekit-rtc/tests/test_basic.py b/livekit-rtc/tests/test_basic.py index a52b1db9..98e8f4c4 100644 --- a/livekit-rtc/tests/test_basic.py +++ b/livekit-rtc/tests/test_basic.py @@ -11,118 +11,23 @@ from __future__ import annotations -import asyncio import os -import uuid -from typing import Any, Callable, Optional import pytest -from livekit import api, rtc -from livekit.rtc.room import EventTypes +from livekit import rtc - -WAIT_TIMEOUT = 20.0 -WAIT_INTERVAL = 0.1 - - -def skip_if_no_credentials() -> pytest.MarkDecorator: - required_vars = ["LIVEKIT_URL", "LIVEKIT_API_KEY", "LIVEKIT_API_SECRET"] - missing = [var for var in required_vars if not os.getenv(var)] - return pytest.mark.skipif( - bool(missing), reason=f"Missing environment variables: {', '.join(missing)}" - ) - - -def create_token(identity: str, room_name: str) -> str: - return ( - api.AccessToken() - .with_identity(identity) - .with_name(identity) - .with_grants( - api.VideoGrants( - room_join=True, - room=room_name, - ) - ) - .to_jwt() - ) - - -def unique_room_name(base: str) -> str: - return f"{base}-{uuid.uuid4().hex[:8]}" - - -async def _wait_until( - predicate: Callable[[], bool], - *, - timeout: float = WAIT_TIMEOUT, - interval: float = WAIT_INTERVAL, - message: str = "condition not met", -) -> None: - loop = asyncio.get_event_loop() - deadline = loop.time() + timeout - while loop.time() < deadline: - if predicate(): - return - await asyncio.sleep(interval) - raise AssertionError(f"timeout waiting: {message}") - - -async def _connect(room: rtc.Room, identity: str, room_name: str) -> str: - """Mints a token, connects `room`, and returns the token (for reconnect).""" - token = create_token(identity, room_name) - url = os.environ["LIVEKIT_URL"] - await room.connect(url, token) - return token - - -async def _ensure_all_connected(rooms: list[rtc.Room]) -> None: - await _wait_until( - lambda: all(r.connection_state == rtc.ConnectionState.CONN_CONNECTED for r in rooms), - message="not all rooms reached CONN_CONNECTED", - ) - - -async def _ensure_track_subscribed(room: rtc.Room, track_sid: str) -> None: - def _has_subscribed() -> bool: - for participant in room.remote_participants.values(): - pub = participant.track_publications.get(track_sid) - if pub is not None and pub.subscribed: - return True - return False - - await _wait_until( - _has_subscribed, - message=f"room did not subscribe to track {track_sid}", - ) - - -def _expect_event( - room: rtc.Room, - event: EventTypes, - predicate: Optional[Callable[..., bool]] = None, -) -> asyncio.Future: - """Returns a future that resolves when `event` (optionally matching - `predicate`) is fired on `room`.""" - loop = asyncio.get_event_loop() - fut: asyncio.Future = loop.create_future() - - def _on_event(*args: Any, **kwargs: Any) -> None: - if fut.done(): - return - if predicate is None or predicate(*args, **kwargs): - fut.set_result(args) - - room.on(event, _on_event) - return fut - - -async def _await_event(fut: asyncio.Future, timeout: float = WAIT_TIMEOUT) -> None: - try: - await asyncio.wait_for(fut, timeout=timeout) - except asyncio.TimeoutError as e: - raise AssertionError("timed out waiting for event") from e +from utils import ( # type: ignore[import-not-found] + await_event, + connect_room, + create_token, + ensure_rooms_all_connected, + ensure_track_subscribed, + expect_room_event, + skip_if_no_credentials, + unique_room_name, + wait_until, +) async def _publish_video(room: rtc.Room, track_name: str) -> rtc.LocalTrackPublication: @@ -139,40 +44,39 @@ async def _publish_audio(room: rtc.Room, track_name: str) -> rtc.LocalTrackPubli return await room.local_participant.publish_track(track, options) -@skip_if_no_credentials() +@skip_if_no_credentials() # type: ignore[untyped-decorator] @pytest.mark.asyncio async def test_connection_basics() -> None: room_name = unique_room_name("py-basics") - p1, p2 = rtc.Room(), rtc.Room() - await _connect(p1, "p1", room_name) - await _connect(p2, "p2", room_name) - await _ensure_all_connected([p1, p2]) + p1 = await connect_room("p1", room_name) + p2 = await connect_room("p2", room_name) + await ensure_rooms_all_connected([p1, p2]) # p2 should observe p1 leaving - p2_saw_p1_left = _expect_event( + p2_saw_p1_left = expect_room_event( p2, "participant_disconnected", predicate=lambda p: p.identity == "p1", ) await p1.disconnect() - await _await_event(p2_saw_p1_left) + await await_event(p2_saw_p1_left) - await _wait_until( + await wait_until( lambda: p1.connection_state == rtc.ConnectionState.CONN_DISCONNECTED, message="p1 did not reach CONN_DISCONNECTED", ) await p2.disconnect() - await _wait_until( + await wait_until( lambda: p2.connection_state == rtc.ConnectionState.CONN_DISCONNECTED, message="p2 did not reach CONN_DISCONNECTED", ) # p3: connect, disconnect, reconnect, disconnect cycle - p3 = rtc.Room() - p3_token = await _connect(p3, "p3", room_name) + p3 = await connect_room("p3", room_name) p3_url = os.environ["LIVEKIT_URL"] + p3_token = create_token("p3", room_name) await p3.disconnect() assert p3.connection_state == rtc.ConnectionState.CONN_DISCONNECTED, ( @@ -190,8 +94,7 @@ async def test_connection_basics() -> None: ) # p4 joins, then p3 reconnects to publish to p4 - p4 = rtc.Room() - await _connect(p4, "p4", room_name) + p4 = await connect_room("p4", room_name) await p3.connect(p3_url, p3_token) assert p3.connection_state == rtc.ConnectionState.CONN_CONNECTED, ( @@ -199,20 +102,20 @@ async def test_connection_basics() -> None: ) # publish camera from p3, expect p4 to see track_published - video_published = _expect_event(p4, "track_published") + video_published = expect_room_event(p4, "track_published") video_pub = await _publish_video(p3, "p3-camera") - await _await_event(video_published) - await _ensure_track_subscribed(p4, video_pub.sid) + await await_event(video_published) + await ensure_track_subscribed(p4, video_pub.sid) # publish microphone from p3, expect p4 to see another track_published - audio_published = _expect_event( + audio_published = expect_room_event( p4, "track_published", predicate=lambda pub, _p: pub.sid != video_pub.sid, ) audio_pub = await _publish_audio(p3, "p3-mic") - await _await_event(audio_published) - await _ensure_track_subscribed(p4, audio_pub.sid) + await await_event(audio_published) + await ensure_track_subscribed(p4, audio_pub.sid) await p3.disconnect() await p4.disconnect()