Skip to content

chore: Add shared helper functions used across the livekit-rtc test suite. #684

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cloudwebrtc merged 2 commits into from
May 22, 2026
Merged

chore: Add shared helper functions used across the livekit-rtc test suite. #684

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
191b453
chore: Add shared helper functions used across the livekit-rtc test s…
cloudwebrtc May 22, 2026
73db5ba
fix types check errors.
cloudwebrtc May 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions livekit-rtc/tests/__init__.py → livekit-rtc/tests/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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))
38 changes: 7 additions & 31 deletions livekit-rtc/tests/test_audio.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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,
Expand Down Expand Up @@ -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")
Expand All @@ -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
Expand Down
38 changes: 6 additions & 32 deletions livekit-rtc/tests/test_change_video_quality.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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],
*,
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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",
Expand Down
56 changes: 8 additions & 48 deletions livekit-rtc/tests/test_e2ee_per_participant.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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))
Expand All @@ -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()
Expand Down
56 changes: 8 additions & 48 deletions livekit-rtc/tests/test_e2ee_shared_key.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,64 +18,24 @@

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"
WRONG_KEY = b"wrongkey"
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()
Expand Down
39 changes: 7 additions & 32 deletions livekit-rtc/tests/test_video.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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)
Expand Down Expand Up @@ -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")
Expand All @@ -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
Expand Down
Loading
Loading