From e9024db5d207463f7ce608e002fed0304a6e4caa Mon Sep 17 00:00:00 2001 From: David Zhao Date: Sun, 5 Jul 2026 16:01:50 +0200 Subject: [PATCH 1/2] consistent error class ServerError, test suite ServerError is now the error base class. we've also kept TwirpError as an alias. also added full api smoke test suite --- livekit-api/livekit/api/__init__.py | 4 +- livekit-api/livekit/api/_dial_timeout.py | 5 +- livekit-api/livekit/api/_service.py | 12 +- livekit-api/livekit/api/connector_service.py | 11 +- livekit-api/livekit/api/livekit_api.py | 64 ++- livekit-api/livekit/api/sip_service.py | 57 +- livekit-api/livekit/api/twirp_client.py | 66 ++- tests/api/test_failover.py | 42 +- tests/api/test_livekitapi.py | 565 +++++++++++++++++++ 9 files changed, 752 insertions(+), 74 deletions(-) create mode 100644 tests/api/test_livekitapi.py diff --git a/livekit-api/livekit/api/__init__.py b/livekit-api/livekit/api/__init__.py index fb115f6f..d60fb9bb 100644 --- a/livekit-api/livekit/api/__init__.py +++ b/livekit-api/livekit/api/__init__.py @@ -36,7 +36,7 @@ from livekit.protocol.connector_whatsapp import * from livekit.protocol.connector_twilio import * -from .twirp_client import TwirpError, TwirpErrorCode +from .twirp_client import ServerError, SipCallError, TwirpError, TwirpErrorCode from .livekit_api import LiveKitAPI from .access_token import ( InferenceGrants, @@ -64,6 +64,8 @@ "AccessToken", "TokenVerifier", "WebhookReceiver", + "ServerError", "TwirpError", "TwirpErrorCode", + "SipCallError", ] diff --git a/livekit-api/livekit/api/_dial_timeout.py b/livekit-api/livekit/api/_dial_timeout.py index a703b204..493afdb2 100644 --- a/livekit-api/livekit/api/_dial_timeout.py +++ b/livekit-api/livekit/api/_dial_timeout.py @@ -2,15 +2,12 @@ from typing import Optional, Union -from livekit.protocol.connector_whatsapp import AcceptWhatsAppCallRequest from livekit.protocol.sip import CreateSIPParticipantRequest, TransferSIPParticipantRequest -# Requests that carry wait_until_answered / ringing_timeout and share the -# phone-dialing timeout behavior. +# Requests that carry ringing_timeout and share the phone-dialing timeout behavior. DialRequest = Union[ CreateSIPParticipantRequest, TransferSIPParticipantRequest, - AcceptWhatsAppCallRequest, ] """@private""" diff --git a/livekit-api/livekit/api/_service.py b/livekit-api/livekit/api/_service.py index a198c388..6e3394aa 100644 --- a/livekit-api/livekit/api/_service.py +++ b/livekit-api/livekit/api/_service.py @@ -20,10 +20,17 @@ def __init__( self._client = TwirpClient(session, host, "livekit", failover=failover) self.api_key = api_key self.api_secret = api_secret + # A pre-signed token set by LiveKitAPI for token auth; sent verbatim, + # skipping per-call signing. Per-service constructors stay key/secret-only. + self._token: str | None = None def _auth_header( self, grants: VideoGrants | None, sip: SIPGrants | None = None ) -> dict[str, str]: + # A pre-signed token is sent verbatim; the caller is responsible for its grants. + if self._token: + return {AUTHORIZATION: "Bearer {}".format(self._token)} + tok = AccessToken(self.api_key, self.api_secret) if grants: tok.with_grants(grants) @@ -31,7 +38,4 @@ def _auth_header( tok.with_sip_grants(sip) token = tok.to_jwt() - - headers = {} - headers[AUTHORIZATION] = "Bearer {}".format(token) - return headers + return {AUTHORIZATION: "Bearer {}".format(token)} diff --git a/livekit-api/livekit/api/connector_service.py b/livekit-api/livekit/api/connector_service.py index eef00821..4f5c09c9 100644 --- a/livekit-api/livekit/api/connector_service.py +++ b/livekit-api/livekit/api/connector_service.py @@ -119,18 +119,17 @@ async def accept_whatsapp_call( Args: request: AcceptWhatsAppCallRequest containing call parameters and SDP timeout: Optional request timeout in seconds. When the request waits - for an answer (wait_until_answered), it defaults to the standard - ring window; set it above the ringing_timeout passed to - dial_whatsapp_call (the two calls are separate, so the SDK can't - derive it). + for the inbound party to join (wait_until_answered), it defaults + to the standard ring window. Returns: AcceptWhatsAppCallResponse with the room name """ client_timeout: Optional[aiohttp.ClientTimeout] = None if request.wait_until_answered: - # Accept can block until the call is answered, so default to the - # standard ring window; the caller overrides via timeout. + # Waiting for the inbound party to join can block, so default the + # request timeout to the standard ring window; the caller overrides + # via timeout. client_timeout = aiohttp.ClientTimeout( total=timeout if timeout else DEFAULT_RINGING_TIMEOUT ) diff --git a/livekit-api/livekit/api/livekit_api.py b/livekit-api/livekit/api/livekit_api.py index 5973a918..4b4443b0 100644 --- a/livekit-api/livekit/api/livekit_api.py +++ b/livekit-api/livekit/api/livekit_api.py @@ -29,28 +29,35 @@ def __init__( api_key: Optional[str] = None, api_secret: Optional[str] = None, *, + token: Optional[str] = None, timeout: Optional[aiohttp.ClientTimeout] = None, session: Optional[aiohttp.ClientSession] = None, failover: bool = True, ): """Create a new LiveKitAPI instance. + Authenticate with an API key and secret (recommended for backend use). + For token auth (client-side use, where the API secret must not be + exposed), prefer the :meth:`with_token` constructor. + Args: url: LiveKit server URL (read from `LIVEKIT_URL` environment variable if not provided) api_key: API key (read from `LIVEKIT_API_KEY` environment variable if not provided) api_secret: API secret (read from `LIVEKIT_API_SECRET` environment variable if not provided) + token: Pre-signed access token (read from `LIVEKIT_TOKEN` environment variable if not provided) timeout: Request timeout (default: 10 seconds) session: aiohttp.ClientSession instance to use for requests, if not provided, a new one will be created """ url = url or os.getenv("LIVEKIT_URL") + token = token or os.getenv("LIVEKIT_TOKEN") api_key = api_key or os.getenv("LIVEKIT_API_KEY") api_secret = api_secret or os.getenv("LIVEKIT_API_SECRET") if not url: raise ValueError("url must be set") - if not api_key or not api_secret: - raise ValueError("api_key and api_secret must be set") + if not token and (not api_key or not api_secret): + raise ValueError("either token, or api_key and api_secret, must be set") self._custom_session = True self._session = session @@ -60,14 +67,51 @@ def __init__( timeout = aiohttp.ClientTimeout(total=10) self._session = aiohttp.ClientSession(timeout=timeout) - self._room = RoomService(self._session, url, api_key, api_secret, failover) - self._ingress = IngressService(self._session, url, api_key, api_secret, failover) - self._egress = EgressService(self._session, url, api_key, api_secret, failover) - self._sip = SipService(self._session, url, api_key, api_secret, failover) - self._agent_dispatch = AgentDispatchService( - self._session, url, api_key, api_secret, failover - ) - self._connector = ConnectorService(self._session, url, api_key, api_secret, failover) + # In token mode there is no key/secret; pass empty strings and rely on + # the token, injected into each service below. + key = api_key or "" + secret = api_secret or "" + self._room = RoomService(self._session, url, key, secret, failover) + self._ingress = IngressService(self._session, url, key, secret, failover) + self._egress = EgressService(self._session, url, key, secret, failover) + self._sip = SipService(self._session, url, key, secret, failover) + self._agent_dispatch = AgentDispatchService(self._session, url, key, secret, failover) + self._connector = ConnectorService(self._session, url, key, secret, failover) + + if token: + for svc in ( + self._room, + self._ingress, + self._egress, + self._sip, + self._agent_dispatch, + self._connector, + ): + svc._token = token + + @classmethod + def with_token( + cls, + token: str, + url: Optional[str] = None, + *, + timeout: Optional[aiohttp.ClientTimeout] = None, + session: Optional[aiohttp.ClientSession] = None, + failover: bool = True, + ) -> "LiveKitAPI": + """Create a LiveKitAPI authenticated with a pre-signed token. + + The token is sent verbatim and must already carry the grants for the calls + it's used with. Since it needs no secret, this is suitable for client-side + use. `url` falls back to the `LIVEKIT_URL` environment variable. + + Args: + token: Pre-signed access token + url: LiveKit server URL (read from `LIVEKIT_URL` if not provided) + timeout: Request timeout (default: 10 seconds) + session: aiohttp.ClientSession to use; a new one is created if omitted + """ + return cls(url, token=token, timeout=timeout, session=session, failover=failover) @property def agent_dispatch(self) -> AgentDispatchService: diff --git a/livekit-api/livekit/api/sip_service.py b/livekit-api/livekit/api/sip_service.py index 70818c92..0b46cf20 100644 --- a/livekit-api/livekit/api/sip_service.py +++ b/livekit-api/livekit/api/sip_service.py @@ -36,6 +36,7 @@ SIPMediaConfig, ) from ._service import Service +from .twirp_client import SipCallError, ServerError from ._dial_timeout import ( dial_timeout as _dial_timeout, pin_ringing_timeout as _pin_ringing_timeout, @@ -46,6 +47,14 @@ """@private""" +def _as_sip_error(err: ServerError) -> ServerError: + """Surface a SIP dialing failure as a SipCallError so callers can branch on + the SIP status; other failures (auth, validation) are returned unchanged.""" + if "sip_status_code" in err.metadata: + return SipCallError.from_server_error(err) + return err + + class SipService(Service): """Client for LiveKit SIP Service API @@ -806,14 +815,17 @@ async def create_sip_participant( if outbound_trunk_config: create.trunk = outbound_trunk_config - return await self._client.request( - SVC, - "CreateSIPParticipant", - create, - self._auth_header(VideoGrants(), sip=SIPGrants(call=True)), - SIPParticipantInfo, - timeout=client_timeout, - ) + try: + return await self._client.request( + SVC, + "CreateSIPParticipant", + create, + self._auth_header(VideoGrants(), sip=SIPGrants(call=True)), + SIPParticipantInfo, + timeout=client_timeout, + ) + except ServerError as e: + raise _as_sip_error(e) from None async def transfer_sip_participant( self, @@ -837,20 +849,23 @@ async def transfer_sip_participant( # timeout doesn't depend on the server's default. _pin_ringing_timeout(transfer) client_timeout = aiohttp.ClientTimeout(total=_dial_timeout(timeout, transfer)) - return await self._client.request( - SVC, - "TransferSIPParticipant", - transfer, - self._auth_header( - VideoGrants( - room_admin=True, - room=transfer.room_name, + try: + return await self._client.request( + SVC, + "TransferSIPParticipant", + transfer, + self._auth_header( + VideoGrants( + room_admin=True, + room=transfer.room_name, + ), + sip=SIPGrants(call=True), ), - sip=SIPGrants(call=True), - ), - SIPParticipantInfo, - timeout=client_timeout, - ) + SIPParticipantInfo, + timeout=client_timeout, + ) + except ServerError as e: + raise _as_sip_error(e) from None def _admin_headers(self) -> dict[str, str]: return self._auth_header(VideoGrants(), sip=SIPGrants(admin=True)) diff --git a/livekit-api/livekit/api/twirp_client.py b/livekit-api/livekit/api/twirp_client.py index 8eba94fd..2f5cdaaf 100644 --- a/livekit-api/livekit/api/twirp_client.py +++ b/livekit-api/livekit/api/twirp_client.py @@ -28,16 +28,20 @@ origin_of, pick_next, ) +from .version import __version__ DEFAULT_PREFIX = "twirp" logger = logging.getLogger("livekit") +# Identifies the SDK and version to the server on every request. +_USER_AGENT = f"livekit-server-sdk-python/{__version__}" + # Shared across all clients in the process so the region list is fetched once. _REGION_CACHE = RegionCache() -class TwirpError(Exception): +class ServerError(Exception): def __init__( self, code: str, @@ -66,17 +70,62 @@ def status(self) -> int: @property def metadata(self) -> Dict[str, str]: - """Twirp metadata""" + """Server-provided error metadata""" return self._metadata def __str__(self) -> str: - result = f"TwirpError(code={self.code}, message={self.message}, status={self.status}" + result = f"ServerError(code={self.code}, message={self.message}, status={self.status}" if self.metadata: result += f", metadata={self.metadata}" result += ")" return result +class SipCallError(ServerError): + """A :class:`ServerError` from a SIP dialing call (``create_sip_participant`` / + ``transfer_sip_participant``) that failed with a SIP response status. The SIP + code and reason are exposed as properties; any other error metadata remains + available via :attr:`metadata`.""" + + @property + def sip_status_code(self) -> Optional[int]: + """The SIP response code of the failed call, e.g. 486 (Busy Here).""" + raw = self.metadata.get("sip_status_code") + return int(raw) if raw is not None else None + + @property + def sip_status(self) -> Optional[str]: + """The SIP reason phrase of the failed call, e.g. "Busy Here".""" + return self.metadata.get("sip_status") + + @classmethod + def from_server_error(cls, err: ServerError) -> "SipCallError": + return cls(err.code, err.message, status=err.status, metadata=err.metadata) + + def __str__(self) -> str: + code = self.metadata.get("sip_status_code") + if code is None: + return super().__str__() + # A clear, SIP-specific representation, including any extra metadata. + reason = self.metadata.get("sip_status") + result = f"SIP call failed: {code}" + if reason: + result += f" {reason}" + result += f" ({self.code})" + extra = { + k: v + for k, v in self.metadata.items() + if k not in ("sip_status_code", "sip_status", "error_details") + } + if extra: + result += " [" + ", ".join(f"{k}={v}" for k, v in extra.items()) + "]" + return result + + +# Deprecated alias for :class:`ServerError`, kept for backwards compatibility. +TwirpError = ServerError + + class TwirpErrorCode: CANCELED = "canceled" UNKNOWN = "unknown" @@ -145,8 +194,9 @@ async def request( headers intact — against the next untried region, with exponential backoff. A 4xx is returned immediately.""" path = f"{self.prefix}/{self.pkg}.{service}/{method}" - forward_headers = dict(headers) # for the discovery fetch (no content-type yet) headers = dict(headers) + headers["User-Agent"] = _USER_AGENT + forward_headers = dict(headers) # for the discovery fetch (no content-type yet) headers["Content-Type"] = "application/protobuf" serialized_data = data.SerializeToString() @@ -183,7 +233,7 @@ async def request( error_data = {} if resp.status < 500: # 4xx is terminal. - raise self._twirp_error(error_data, resp.status) + raise self._server_error(error_data, resp.status) retryable_status = resp.status except (aiohttp.ClientError, asyncio.TimeoutError) as e: transport_exc = e @@ -200,7 +250,7 @@ async def request( if next_origin is None: if transport_exc is not None: raise transport_exc - raise self._twirp_error(error_data, retryable_status or 500) + raise self._server_error(error_data, retryable_status or 500) reason = transport_exc if transport_exc is not None else f"status {retryable_status}" logger.warning( @@ -216,8 +266,8 @@ async def request( raise RuntimeError("failover loop exited without returning") # unreachable @staticmethod - def _twirp_error(error_data: Dict, status: int) -> "TwirpError": - return TwirpError( + def _server_error(error_data: Dict, status: int) -> "ServerError": + return ServerError( error_data.get("code", "unknown"), error_data.get("msg", ""), status=status, diff --git a/tests/api/test_failover.py b/tests/api/test_failover.py index c941761a..f9521c44 100644 --- a/tests/api/test_failover.py +++ b/tests/api/test_failover.py @@ -18,19 +18,21 @@ reachable. The mock returns Cache-Control: max-age=0, so the region cache never stores entries and scenarios don't interfere. -See cmd/test-server/README.md for the X-Lk-Mock-* control protocol. These tests -drive TwirpClient.request() directly because the public service methods do not -expose per-call headers. +See cmd/test-server/README.md for the X-Lk-Mock JSON control protocol. These +tests drive TwirpClient.request() directly because failover relies on internal +test-only knobs (_failover_force/_failover_backoff) the public methods don't +expose. """ import asyncio +import json import os import urllib.request import aiohttp import pytest -from livekit.api import CreateRoomRequest, Room, TwirpError +from livekit.api import CreateRoomRequest, Room, ServerError from livekit.api.twirp_client import TwirpClient BASE = os.getenv("LK_TEST_SERVER_URL", "http://127.0.0.1:9999") @@ -51,7 +53,7 @@ def _server_up() -> bool: # _failover_force bypasses the cloud-host check (the mock is on 127.0.0.1) and a # tiny backoff keeps the tests fast — both are internal, test-only knobs. -async def _call(directives: dict, *, failover: bool = True, force: bool = True) -> Room: +async def _call(mock: dict, *, failover: bool = True, force: bool = True) -> Room: async with aiohttp.ClientSession() as session: client = TwirpClient( session, @@ -63,8 +65,8 @@ async def _call(directives: dict, *, failover: bool = True, force: bool = True) ) headers = { "authorization": "Bearer test-token", - "x-lk-mock-skip-auth": "true", - **directives, + # These tests exercise failover, not authz; skip the mock's permission check. + "X-Lk-Mock": json.dumps({"skipAuth": True, **mock}), } return await client.request("RoomService", "CreateRoom", CreateRoomRequest(), headers, Room) @@ -74,40 +76,40 @@ def test_healthy(): def test_primary_unavailable(): - asyncio.run(_call({"x-lk-mock-fail-regions": "0"})) + asyncio.run(_call({"failRegions": [0]})) def test_two_regions_unavailable(): - asyncio.run(_call({"x-lk-mock-fail-regions": "0,1"})) + asyncio.run(_call({"failRegions": [0, 1]})) def test_all_unavailable(): - with pytest.raises(TwirpError): - asyncio.run(_call({"x-lk-mock-fail-regions": "0,1,2,3"})) + with pytest.raises(ServerError): + asyncio.run(_call({"failRegions": [0, 1, 2, 3]})) def test_client_error_not_retried(): - with pytest.raises(TwirpError) as exc: - asyncio.run(_call({"x-lk-mock-fail-regions": "0", "x-lk-mock-fail-status": "400"})) + with pytest.raises(ServerError) as exc: + asyncio.run(_call({"failRegions": [0], "failStatus": 400})) assert exc.value.code == "invalid_argument" def test_transport_error_failover(): - asyncio.run(_call({"x-lk-mock-fail-regions": "0", "x-lk-mock-fail-mode": "drop"})) + asyncio.run(_call({"failRegions": [0], "failMode": "drop"})) def test_region_discovery_unreachable(): - with pytest.raises(TwirpError): - asyncio.run(_call({"x-lk-mock-fail-regions": "0", "x-lk-mock-regions-status": "500"})) + with pytest.raises(ServerError): + asyncio.run(_call({"failRegions": [0], "regionsStatus": 500})) def test_not_cloud_host(): # Enabled but not forced; 127.0.0.1 is not a cloud host, so no failover. - with pytest.raises(TwirpError): - asyncio.run(_call({"x-lk-mock-fail-regions": "0"}, force=False)) + with pytest.raises(ServerError): + asyncio.run(_call({"failRegions": [0]}, force=False)) def test_disabled(): # failover=False disables failover entirely. - with pytest.raises(TwirpError): - asyncio.run(_call({"x-lk-mock-fail-regions": "0"}, failover=False)) + with pytest.raises(ServerError): + asyncio.run(_call({"failRegions": [0]}, failover=False)) diff --git a/tests/api/test_livekitapi.py b/tests/api/test_livekitapi.py new file mode 100644 index 00000000..76727b13 --- /dev/null +++ b/tests/api/test_livekitapi.py @@ -0,0 +1,565 @@ +# 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. + +"""API tests that drive the unified LiveKitAPI against the mock LiveKit server +(livekit/livekit cmd/test-server). Point them at a running instance with +LK_TEST_SERVER_URL (default http://127.0.0.1:9999); they skip when no server is +reachable. + +Because the mock enforces the same per-method grants as the real server, a call +that succeeds also proves the SDK attached the right grants automatically. The +smoke tests fully populate each request so they double as a reference for a +complete call and exercise field serialization. Mock directives are passed via +the X-Lk-Mock header, set as a default header on the session (the public service +methods don't expose per-call headers). +""" + +import asyncio +import contextlib +import json +import os +import urllib.request +from typing import Optional + +import aiohttp +import pytest +from google.protobuf.duration_pb2 import Duration + +import livekit.api as api +from livekit.api import SipCallError, ServerError +from livekit.protocol.rtc import SessionDescription + +BASE = os.getenv("LK_TEST_SERVER_URL", "http://127.0.0.1:9999") +# devkey/secret match `livekit-server --dev`, which the mock verifies against. +KEY, SECRET = "devkey", "secret" + + +def _server_up() -> bool: + try: + with urllib.request.urlopen(f"{BASE}/settings/regions", timeout=1) as r: + return r.status == 200 + except Exception: + return False + + +pytestmark = pytest.mark.skipif( + not _server_up(), reason=f"mock test server not reachable at {BASE}" +) + + +@contextlib.asynccontextmanager +async def _api(mock: Optional[dict] = None): + """A LiveKitAPI whose session carries the given X-Lk-Mock directives (if any) + as a default header on every request.""" + headers = {"X-Lk-Mock": json.dumps(mock)} if mock else None + async with aiohttp.ClientSession(headers=headers) as session: + yield api.LiveKitAPI(BASE, KEY, SECRET, session=session) + + +def _mp4(path: str) -> api.EncodedFileOutput: + return api.EncodedFileOutput(file_type=api.EncodedFileType.MP4, filepath=path) + + +# -- smoke: fully-populated calls across every service ------------------------ + + +async def _room_smoke(): + async with _api() as lk: + await lk.room.create_room( + api.CreateRoomRequest( + name="test-room", + empty_timeout=300, + departure_timeout=60, + max_participants=50, + metadata='{"scene":"lobby"}', + min_playout_delay=100, + max_playout_delay=2000, + sync_streams=True, + agents=[api.RoomAgentDispatch(agent_name="greeter", metadata='{"lang":"en"}')], + ) + ) + await lk.room.list_rooms(api.ListRoomsRequest(names=["test-room", "lobby"])) + await lk.room.delete_room(api.DeleteRoomRequest(room="test-room")) + await lk.room.list_participants(api.ListParticipantsRequest(room="test-room")) + await lk.room.get_participant( + api.RoomParticipantIdentity(room="test-room", identity="participant-42") + ) + await lk.room.remove_participant( + api.RoomParticipantIdentity(room="test-room", identity="participant-42") + ) + await lk.room.forward_participant( + api.ForwardParticipantRequest( + room="test-room", identity="participant-42", destination_room="overflow-room" + ) + ) + await lk.room.move_participant( + api.MoveParticipantRequest( + room="test-room", identity="participant-42", destination_room="breakout-room" + ) + ) + await lk.room.mute_published_track( + api.MuteRoomTrackRequest( + room="test-room", identity="participant-42", track_sid="TR_video1", muted=True + ) + ) + await lk.room.update_participant( + api.UpdateParticipantRequest( + room="test-room", + identity="participant-42", + name="Alice", + metadata='{"role":"host"}', + attributes={"seat": "1A"}, + permission=api.ParticipantPermission( + can_subscribe=True, + can_publish=True, + can_publish_data=True, + can_publish_sources=[api.TrackSource.MICROPHONE, api.TrackSource.CAMERA], + can_update_metadata=True, + ), + ) + ) + await lk.room.update_subscriptions( + api.UpdateSubscriptionsRequest( + room="test-room", + identity="participant-42", + track_sids=["TR_video1"], + subscribe=True, + participant_tracks=[ + api.ParticipantTracks(participant_sid="PA_xyz789", track_sids=["TR_video1"]) + ], + ) + ) + await lk.room.update_room_metadata( + api.UpdateRoomMetadataRequest(room="test-room", metadata='{"scene":"intro"}') + ) + await lk.room.send_data( + api.SendDataRequest( + room="test-room", + data=b"hello world", + kind=api.DataPacket.RELIABLE, + destination_identities=["participant-42"], + topic="chat", + ) + ) + + +async def _egress_smoke(): + async with _api() as lk: + await lk.egress.start_room_composite_egress( + api.RoomCompositeEgressRequest( + room_name="test-room", layout="grid", file_outputs=[_mp4("room.mp4")] + ) + ) + await lk.egress.start_web_egress( + api.WebEgressRequest( + url="https://example.com/scene", + stream_outputs=[ + api.StreamOutput( + protocol=api.StreamProtocol.RTMP, urls=["rtmps://a.example.com/live/key"] + ) + ], + ) + ) + await lk.egress.start_participant_egress( + api.ParticipantEgressRequest( + room_name="test-room", + identity="participant-42", + screen_share=True, + file_outputs=[_mp4("participant.mp4")], + ) + ) + await lk.egress.start_track_composite_egress( + api.TrackCompositeEgressRequest( + room_name="test-room", + audio_track_id="TR_audio1", + video_track_id="TR_video1", + file_outputs=[_mp4("track-composite.mp4")], + ) + ) + await lk.egress.start_track_egress( + api.TrackEgressRequest( + room_name="test-room", + track_id="TR_video1", + file=api.DirectFileOutput(filepath="track.mp4"), + ) + ) + await lk.egress.update_layout( + api.UpdateLayoutRequest(egress_id="EG_abc123", layout="speaker") + ) + await lk.egress.update_stream( + api.UpdateStreamRequest( + egress_id="EG_abc123", + add_output_urls=["rtmps://b.example.com/live/key"], + remove_output_urls=["rtmps://a.example.com/live/key"], + ) + ) + await lk.egress.list_egress( + api.ListEgressRequest(room_name="test-room", egress_id="EG_abc123", active=True) + ) + await lk.egress.stop_egress(api.StopEgressRequest(egress_id="EG_abc123")) + + +async def _ingress_smoke(): + async with _api() as lk: + await lk.ingress.create_ingress( + api.CreateIngressRequest( + input_type=api.IngressInput.RTMP_INPUT, + name="stream-input", + room_name="test-room", + participant_identity="ingress-bot", + participant_name="Live Stream", + participant_metadata='{"source":"rtmp"}', + enable_transcoding=True, + audio=api.IngressAudioOptions( + name="audio", + source=api.TrackSource.MICROPHONE, + preset=api.IngressAudioEncodingPreset.OPUS_STEREO_96KBPS, + ), + video=api.IngressVideoOptions( + name="video", + source=api.TrackSource.CAMERA, + preset=api.IngressVideoEncodingPreset.H264_1080P_30FPS_3_LAYERS, + ), + ) + ) + await lk.ingress.update_ingress( + api.UpdateIngressRequest( + ingress_id="IN_abc123", + name="stream-input-v2", + room_name="test-room", + participant_identity="ingress-bot", + participant_name="Live Stream", + enable_transcoding=True, + ) + ) + await lk.ingress.list_ingress( + api.ListIngressRequest(room_name="test-room", ingress_id="IN_abc123") + ) + await lk.ingress.delete_ingress(api.DeleteIngressRequest(ingress_id="IN_abc123")) + + +async def _sip_smoke(): + async with _api() as lk: + await lk.sip.create_inbound_trunk( + api.CreateSIPInboundTrunkRequest( + trunk=api.SIPInboundTrunkInfo( + name="inbound", + metadata='{"provider":"telco"}', + numbers=["+15105550100"], + allowed_addresses=["203.0.113.0/24"], + allowed_numbers=["+15105550111"], + auth_username="sip-user", + auth_password="sip-pass", + krisp_enabled=True, + ) + ) + ) + await lk.sip.create_outbound_trunk( + api.CreateSIPOutboundTrunkRequest( + trunk=api.SIPOutboundTrunkInfo( + name="outbound", + address="sip.telco.example.com", + transport=api.SIPTransport.SIP_TRANSPORT_TLS, + destination_country="US", + numbers=["+15105550100"], + auth_username="sip-user", + auth_password="sip-pass", + ) + ) + ) + await lk.sip.update_inbound_trunk( + "ST_abc123", + api.SIPInboundTrunkInfo(name="inbound-v2", numbers=["+15105550100"]), + ) + await lk.sip.update_outbound_trunk( + "ST_abc123", + api.SIPOutboundTrunkInfo( + name="outbound-v2", + address="sip.telco.example.com", + transport=api.SIPTransport.SIP_TRANSPORT_TLS, + numbers=["+15105550100"], + ), + ) + await lk.sip.list_inbound_trunk( + api.ListSIPInboundTrunkRequest(trunk_ids=["ST_abc123"], numbers=["+15105550100"]) + ) + await lk.sip.list_outbound_trunk(api.ListSIPOutboundTrunkRequest(trunk_ids=["ST_abc123"])) + await lk.sip.delete_trunk(api.DeleteSIPTrunkRequest(sip_trunk_id="ST_abc123")) + await lk.sip.create_dispatch_rule( + api.CreateSIPDispatchRuleRequest( + dispatch_rule=api.SIPDispatchRuleInfo( + name="direct-to-support", + metadata='{"team":"support"}', + trunk_ids=["ST_abc123"], + rule=api.SIPDispatchRule( + dispatch_rule_direct=api.SIPDispatchRuleDirect( + room_name="support", pin="1234" + ) + ), + ) + ) + ) + await lk.sip.update_dispatch_rule( + "SDR_abc123", + api.SIPDispatchRuleInfo( + name="individual-v2", + rule=api.SIPDispatchRule( + dispatch_rule_individual=api.SIPDispatchRuleIndividual(room_prefix="call-") + ), + ), + ) + await lk.sip.list_dispatch_rule( + api.ListSIPDispatchRuleRequest( + dispatch_rule_ids=["SDR_abc123"], trunk_ids=["ST_abc123"] + ) + ) + await lk.sip.delete_dispatch_rule( + api.DeleteSIPDispatchRuleRequest(sip_dispatch_rule_id="SDR_abc123") + ) + + +async def _connector_smoke(): + offer = SessionDescription(type="offer", sdp="v=0\r\no=- 0 0 IN IP4 127.0.0.1\r\n") + answer = SessionDescription(type="answer", sdp="v=0\r\no=- 0 0 IN IP4 127.0.0.1\r\n") + async with _api() as lk: + await lk.connector.dial_whatsapp_call( + api.DialWhatsAppCallRequest( + whatsapp_phone_number_id="123456789012345", + whatsapp_to_phone_number="+15105550100", + whatsapp_api_key="wa-secret-key", + whatsapp_cloud_api_version="23.0", + room_name="test-room", + participant_identity="whatsapp-caller", + participant_name="WhatsApp Caller", + destination_country="US", + ringing_timeout=Duration(seconds=30), + ) + ) + await lk.connector.accept_whatsapp_call( + api.AcceptWhatsAppCallRequest( + whatsapp_phone_number_id="123456789012345", + whatsapp_api_key="wa-secret-key", + whatsapp_cloud_api_version="23.0", + whatsapp_call_id="wacid.HBgLABC", + sdp=answer, + room_name="test-room", + participant_identity="whatsapp-callee", + ) + ) + await lk.connector.connect_whatsapp_call( + api.ConnectWhatsAppCallRequest(whatsapp_call_id="wacid.HBgLABC", sdp=offer) + ) + await lk.connector.disconnect_whatsapp_call( + api.DisconnectWhatsAppCallRequest( + whatsapp_call_id="wacid.HBgLABC", + whatsapp_api_key="wa-secret-key", + disconnect_reason=api.DisconnectWhatsAppCallRequest.BUSINESS_INITIATED, + ) + ) + await lk.connector.connect_twilio_call( + api.ConnectTwilioCallRequest( + twilio_call_direction=api.ConnectTwilioCallRequest.TWILIO_CALL_DIRECTION_INBOUND, + room_name="test-room", + participant_identity="twilio-caller", + destination_country="US", + ) + ) + + +async def _agent_dispatch_smoke(): + async with _api() as lk: + await lk.agent_dispatch.create_dispatch( + api.CreateAgentDispatchRequest( + room="test-room", agent_name="inbound-agent", metadata='{"lang":"en"}' + ) + ) + await lk.agent_dispatch.get_dispatch("AD_abc123", "test-room") + await lk.agent_dispatch.list_dispatch("test-room") + await lk.agent_dispatch.delete_dispatch("AD_abc123", "test-room") + + +def test_room_smoke(): + asyncio.run(_room_smoke()) + + +def test_egress_smoke(): + asyncio.run(_egress_smoke()) + + +def test_ingress_smoke(): + asyncio.run(_ingress_smoke()) + + +def test_sip_smoke(): + asyncio.run(_sip_smoke()) + + +def test_connector_smoke(): + asyncio.run(_connector_smoke()) + + +def test_agent_dispatch_smoke(): + asyncio.run(_agent_dispatch_smoke()) + + +# -- deep: create_room round-trip + error propagation ------------------------- + + +async def _create_room_echo(): + async with _api() as lk: + req = api.CreateRoomRequest( + name="echo-room", metadata='{"scene":"lobby"}', empty_timeout=300, max_participants=50 + ) + room = await lk.room.create_room(req) + assert room.name == "echo-room" + assert room.metadata == '{"scene":"lobby"}' + assert room.empty_timeout == 300 + assert room.max_participants == 50 + assert room.sid != "" # placeholder assigned by the mock + + +def test_create_room_echoes_fields(): + asyncio.run(_create_room_echo()) + + +async def _create_room_error(): + mock = {"failRegions": [0], "failStatus": 400, "failTwirpCode": "invalid_argument"} + async with _api(mock) as lk: + with pytest.raises(ServerError) as exc: + await lk.room.create_room(api.CreateRoomRequest(name="test-room")) + assert exc.value.code == "invalid_argument" + + +def test_create_room_propagates_twirp_error(): + asyncio.run(_create_room_error()) + + +# -- deep: SIP participant (delayMs:0 skips the mock's answer wait) ------------ + + +async def _sip_participant(): + async with _api() as lk: + p = await lk.sip.create_sip_participant( + api.CreateSIPParticipantRequest( + sip_trunk_id="ST_abc123", + sip_call_to="+15105550100", + room_name="test-room", + participant_identity="sip-caller", + participant_name="SIP Caller", + participant_metadata='{"source":"pstn"}', + dtmf="1234#", + play_dialtone=True, + max_call_duration=Duration(seconds=3600), + ) + ) + assert p.room_name == "test-room" + assert p.participant_identity == "sip-caller" + + async with _api({"delayMs": 0}) as lk: + await lk.sip.create_sip_participant( + api.CreateSIPParticipantRequest( + sip_trunk_id="ST_abc123", + sip_call_to="+15105550100", + room_name="test-room", + wait_until_answered=True, + ringing_timeout=Duration(seconds=2), + ) + ) + await lk.sip.transfer_sip_participant( + api.TransferSIPParticipantRequest( + room_name="test-room", + participant_identity="sip-caller", + transfer_to="tel:+15105550122", + ringing_timeout=Duration(seconds=2), + ) + ) + + +def test_sip_participant(): + asyncio.run(_sip_participant()) + + +# -- cross-cutting: token auth ------------------------------------------------ + + +async def _token_auth(): + token = api.AccessToken(KEY, SECRET).with_grants(api.VideoGrants(room_create=True)).to_jwt() + async with api.LiveKitAPI.with_token(token, BASE) as lk: + room = await lk.room.create_room(api.CreateRoomRequest(name="token-room")) + assert room.name == "token-room" + + +def test_token_auth(): + asyncio.run(_token_auth()) + + +# -- cross-cutting: SIP call errors surface as SipCallError ------------------- + + +async def _sip_call_error(sip: dict, *, wait: bool = False): + create = api.CreateSIPParticipantRequest( + sip_trunk_id="ST_abc123", sip_call_to="+15105550100", room_name="test-room" + ) + if wait: + create.wait_until_answered = True + create.ringing_timeout.CopyFrom(Duration(seconds=2)) + async with _api({"delayMs": 0, "sipStatus": sip}) as lk: + with pytest.raises(SipCallError) as exc: + await lk.sip.create_sip_participant(create) + return exc.value + + +def test_sip_busy(): + err = asyncio.run(_sip_call_error({"code": 486, "status": "Busy Here"})) + assert isinstance(err, ServerError) + assert err.code == "resource_exhausted" + assert err.sip_status_code == 486 + assert err.sip_status == "Busy Here" + # printable representation makes the failure clear + assert "486" in str(err) and "Busy Here" in str(err) + + +def test_sip_declined(): + err = asyncio.run(_sip_call_error({"code": 603, "status": "Decline"})) + assert err.code == "permission_denied" + assert err.sip_status_code == 603 + + +def test_sip_no_answer(): + # Callee doesn't pick up within ringing_timeout -> SIP 408 Request Timeout. + err = asyncio.run(_sip_call_error({"code": 408, "status": "Request Timeout"}, wait=True)) + assert err.code == "deadline_exceeded" + assert err.sip_status_code == 408 + + +# -- cross-cutting: client-side dial timeout ---------------------------------- + + +async def _sip_dial_timeout(): + # ringing 1s -> ~3s dial budget; the mock delays the answer past it. + async with _api({"delayMs": 4000}) as lk: + await lk.sip.create_sip_participant( + api.CreateSIPParticipantRequest( + sip_trunk_id="ST_abc123", + sip_call_to="+15105550100", + room_name="test-room", + wait_until_answered=True, + ringing_timeout=Duration(seconds=1), + ) + ) + + +def test_sip_dial_timeout(): + with pytest.raises((asyncio.TimeoutError, TimeoutError)): + asyncio.run(_sip_dial_timeout()) From f1c91ef01a27a1bd5d4fdbf9fbff9dc2d71d4720 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Mon, 6 Jul 2026 12:44:54 +0200 Subject: [PATCH 2/2] add test cases --- tests/api/test_livekitapi.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/tests/api/test_livekitapi.py b/tests/api/test_livekitapi.py index 76727b13..3ce1ead2 100644 --- a/tests/api/test_livekitapi.py +++ b/tests/api/test_livekitapi.py @@ -291,6 +291,13 @@ async def _sip_smoke(): numbers=["+15105550100"], ), ) + # field-level (kwargs) update helpers + await lk.sip.update_inbound_trunk_fields( + "ST_abc123", name="inbound-v3", metadata="{}", numbers=["+15105550100"] + ) + await lk.sip.update_outbound_trunk_fields( + "ST_abc123", name="outbound-v3", address="sip.telco.example.com" + ) await lk.sip.list_inbound_trunk( api.ListSIPInboundTrunkRequest(trunk_ids=["ST_abc123"], numbers=["+15105550100"]) ) @@ -319,6 +326,7 @@ async def _sip_smoke(): ), ), ) + await lk.sip.update_dispatch_rule_fields("SDR_abc123", name="rule-v3", metadata="{}") await lk.sip.list_dispatch_rule( api.ListSIPDispatchRuleRequest( dispatch_rule_ids=["SDR_abc123"], trunk_ids=["ST_abc123"]