@@ -11,118 +11,23 @@
 
 from __future__ import annotations
 
-import asyncio
 import os
-import uuid
-from typing import 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():
-    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, **kwargs) -> 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,29 +94,28 @@ 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, (
         f"expected p3 reconnected, got {p3.connection_state}"
     )
 
     # 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()