diff --git a/livekit-rtc/tests/__init__.py b/livekit-rtc/tests/conftest.py similarity index 54% rename from livekit-rtc/tests/__init__.py rename to livekit-rtc/tests/conftest.py index c75e2bef..03812308 100644 --- a/livekit-rtc/tests/__init__.py +++ b/livekit-rtc/tests/conftest.py @@ -11,3 +11,17 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. + +"""Put this tests directory on sys.path so test modules can `from utils import ...`. + +Both `livekit-rtc/tests/` and the repo-root `tests/` are collected by pytest +together; if either has an `__init__.py` they collide on the `tests` namespace +package name. So we keep this directory non-package and use absolute imports. +""" + +import sys +from pathlib import Path + +_TESTS_DIR = Path(__file__).resolve().parent +if str(_TESTS_DIR) not in sys.path: + sys.path.insert(0, str(_TESTS_DIR)) diff --git a/livekit-rtc/tests/test_audio.py b/livekit-rtc/tests/test_audio.py index baf860ad..4ae3ef6b 100644 --- a/livekit-rtc/tests/test_audio.py +++ b/livekit-rtc/tests/test_audio.py @@ -14,20 +14,23 @@ """End-to-end audio publish/subscribe tests.""" +from __future__ import annotations + import asyncio import ctypes import math import os -import uuid import wave from pathlib import Path import numpy as np import pytest -from livekit import api, rtc +from livekit import rtc from livekit.rtc.audio_frame import AudioFrame +from utils import create_token, skip_if_no_credentials, unique_room_name # type: ignore[import-not-found] + SAMPLE_RATE = 48000 NUM_CHANNELS = 1 @@ -36,33 +39,6 @@ AMPLITUDE = 0.5 -def skip_if_no_credentials(): - 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]}" - - def _generate_sine_wave( frequency: int, sample_rate: int, @@ -132,7 +108,7 @@ def _band_energies( class TestAudioStreamPublishSubscribe: """End-to-end: publish a sine sweep into a room and verify spectrum on the subscriber.""" - async def test_audio_stream_publish_subscribe(self): + async def test_audio_stream_publish_subscribe(self) -> None: """Publish 5 seconds of 100/300/500/700/1000 Hz tones and FFT-verify received audio.""" url = os.environ["LIVEKIT_URL"] room_name = unique_room_name("test-audio-sweep") @@ -151,7 +127,7 @@ def on_track_subscribed( track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant, - ): + ) -> None: nonlocal subscribed_track if track.kind == rtc.TrackKind.KIND_AUDIO: subscribed_track = track diff --git a/livekit-rtc/tests/test_change_video_quality.py b/livekit-rtc/tests/test_change_video_quality.py index 12cd0088..debe0587 100644 --- a/livekit-rtc/tests/test_change_video_quality.py +++ b/livekit-rtc/tests/test_change_video_quality.py @@ -42,16 +42,17 @@ import os import sys import time -import uuid -from typing import Callable, Optional, Tuple +from typing import Any, Callable, Optional, Tuple import numpy as np import pytest -from livekit import api, rtc +from livekit import rtc from livekit.rtc._proto.track_publication_pb2 import VideoQuality from livekit.rtc.room import EventTypes +from utils import create_token, skip_if_no_credentials, unique_room_name # type: ignore[import-not-found] + WAIT_TIMEOUT = 30.0 WAIT_INTERVAL = 0.1 @@ -73,33 +74,6 @@ ] -def skip_if_no_credentials(): - 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], *, @@ -149,7 +123,7 @@ def _expect_event( loop = asyncio.get_running_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): @@ -248,7 +222,7 @@ async def _wait_for_layer( _IS_MACOS = sys.platform == "darwin" -@skip_if_no_credentials() +@skip_if_no_credentials() # type: ignore[untyped-decorator] @pytest.mark.asyncio @pytest.mark.parametrize( "video_codec, codec_name, mode", diff --git a/livekit-rtc/tests/test_e2ee_per_participant.py b/livekit-rtc/tests/test_e2ee_per_participant.py index 6c90e9bb..8062662e 100644 --- a/livekit-rtc/tests/test_e2ee_per_participant.py +++ b/livekit-rtc/tests/test_e2ee_per_participant.py @@ -18,12 +18,17 @@ import asyncio import os -import uuid -from typing import Any, Callable, TypeVar import pytest -from livekit import api, rtc +from livekit import rtc + +from utils import ( # type: ignore[import-not-found] + assert_eventually, + create_token, + skip_if_no_credentials, + unique_room_name, +) # Per-participant keys (publisher identity → list of (key_bytes, key_index)) @@ -37,51 +42,6 @@ WIDTH, HEIGHT = 320, 180 FRAME_RATE = 15 -T = TypeVar("T") - - -def skip_if_no_credentials() -> Any: - 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 assert_eventually( - condition: Callable[[], T], - timeout: float = 15.0, - interval: float = 0.1, - message: str = "Condition not met within timeout", -) -> T: - deadline = asyncio.get_event_loop().time() + timeout - last_result = None - while asyncio.get_event_loop().time() < deadline: - last_result = condition() - if last_result: - return last_result - await asyncio.sleep(interval) - raise AssertionError(f"{message} (last result: {last_result})") - def make_per_participant_e2ee_options() -> rtc.E2EEOptions: options = rtc.E2EEOptions() diff --git a/livekit-rtc/tests/test_e2ee_shared_key.py b/livekit-rtc/tests/test_e2ee_shared_key.py index a14c86bc..4334bb78 100644 --- a/livekit-rtc/tests/test_e2ee_shared_key.py +++ b/livekit-rtc/tests/test_e2ee_shared_key.py @@ -18,12 +18,17 @@ import asyncio import os -import uuid -from typing import Any, Callable, TypeVar import pytest -from livekit import api, rtc +from livekit import rtc + +from utils import ( # type: ignore[import-not-found] + assert_eventually, + create_token, + skip_if_no_credentials, + unique_room_name, +) SHARED_KEY = b"12345678" @@ -31,51 +36,6 @@ WIDTH, HEIGHT = 320, 180 FRAME_RATE = 15 -T = TypeVar("T") - - -def skip_if_no_credentials() -> Any: - 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 assert_eventually( - condition: Callable[[], T], - timeout: float = 10.0, - interval: float = 0.1, - message: str = "Condition not met within timeout", -) -> T: - deadline = asyncio.get_event_loop().time() + timeout - last_result = None - while asyncio.get_event_loop().time() < deadline: - last_result = condition() - if last_result: - return last_result - await asyncio.sleep(interval) - raise AssertionError(f"{message} (last result: {last_result})") - def make_e2ee_options() -> rtc.E2EEOptions: options = rtc.E2EEOptions() diff --git a/livekit-rtc/tests/test_video.py b/livekit-rtc/tests/test_video.py index 5ae0ba1b..857b84f8 100644 --- a/livekit-rtc/tests/test_video.py +++ b/livekit-rtc/tests/test_video.py @@ -14,17 +14,19 @@ """End-to-end video publish/subscribe tests.""" +from __future__ import annotations + import asyncio import os import struct -import uuid import zlib from pathlib import Path import numpy as np -import pytest -from livekit import api, rtc +from livekit import rtc + +from utils import create_token, skip_if_no_credentials, unique_room_name # type: ignore[import-not-found] VIDEO_WIDTH = 640 @@ -41,33 +43,6 @@ ] -def skip_if_no_credentials(): - 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]}" - - def _solid_color_rgba_frame(width: int, height: int, rgb: tuple[int, int, int]) -> rtc.VideoFrame: """Build a solid-color 640x480 RGBA `VideoFrame` for the given RGB triple.""" pixels = np.empty((height, width, 4), dtype=np.uint8) @@ -129,7 +104,7 @@ def _chunk(tag: bytes, data: bytes) -> bytes: class TestVideoStreamPublishSubscribe: """End-to-end: publish a 640x480 color-cycle video and verify colors on the subscriber.""" - async def test_video_stream_publish_subscribe(self): + async def test_video_stream_publish_subscribe(self) -> None: """Publish red/green/blue/white/black (1s each, 15fps) and verify color sequence.""" url = os.environ["LIVEKIT_URL"] room_name = unique_room_name("test-video-colors") @@ -148,7 +123,7 @@ def on_track_subscribed( track: rtc.Track, publication: rtc.RemoteTrackPublication, participant: rtc.RemoteParticipant, - ): + ) -> None: nonlocal subscribed_track if track.kind == rtc.TrackKind.KIND_VIDEO: subscribed_track = track diff --git a/livekit-rtc/tests/utils.py b/livekit-rtc/tests/utils.py new file mode 100644 index 00000000..07f51c84 --- /dev/null +++ b/livekit-rtc/tests/utils.py @@ -0,0 +1,76 @@ +# Copyright 2026 LiveKit, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Shared helpers used across the livekit-rtc test suite.""" + +from __future__ import annotations + +import asyncio +import os +import uuid +from typing import Callable, TypeVar + +import pytest + +from livekit import api + +T = TypeVar("T") + +_REQUIRED_ENV_VARS = ("LIVEKIT_URL", "LIVEKIT_API_KEY", "LIVEKIT_API_SECRET") + + +def skip_if_no_credentials() -> pytest.MarkDecorator: + """pytest mark that skips when any LiveKit credentials env var is missing.""" + missing = [var for var in _REQUIRED_ENV_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: + """Build a room-join JWT for the given identity/room.""" + 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: + """Suffix the base with a short random token so concurrent runs don't collide.""" + return f"{base}-{uuid.uuid4().hex[:8]}" + + +async def assert_eventually( + condition: Callable[[], T], + timeout: float = 15.0, + interval: float = 0.1, + message: str = "Condition not met within timeout", +) -> T: + """Poll `condition()` until it returns a truthy value or `timeout` elapses.""" + deadline = asyncio.get_event_loop().time() + timeout + last_result: T | None = None + while asyncio.get_event_loop().time() < deadline: + last_result = condition() + if last_result: + return last_result + await asyncio.sleep(interval) + raise AssertionError(f"{message} (last result: {last_result})")