From 57059b71362a6eb25a989c16e4b0f72c373a4541 Mon Sep 17 00:00:00 2001 From: Adrian Czerwiec Date: Thu, 13 Aug 2026 15:11:41 +0200 Subject: [PATCH 1/4] add recording status notifications --- fishjam/events/__init__.py | 4 ++++ fishjam/events/_protos/fishjam/__init__.py | 17 +++++++++++++++++ fishjam/events/allowed_notifications.py | 3 +++ protos | 2 +- 4 files changed, 25 insertions(+), 1 deletion(-) diff --git a/fishjam/events/__init__.py b/fishjam/events/__init__.py index 57c113c..1759a77 100644 --- a/fishjam/events/__init__.py +++ b/fishjam/events/__init__.py @@ -11,6 +11,8 @@ ServerMessagePeerDisconnected, ServerMessagePeerMetadataUpdated, ServerMessagePeerType, + ServerMessageRecordingStatusChanged, + ServerMessageRecordingStatusChangedStatus, ServerMessageRoomCrashed, ServerMessageRoomCreated, ServerMessageRoomDeleted, @@ -36,6 +38,8 @@ "ServerMessagePeerDisconnected", "ServerMessagePeerMetadataUpdated", "ServerMessagePeerCrashed", + "ServerMessageRecordingStatusChanged", + "ServerMessageRecordingStatusChangedStatus", "ServerMessageStreamConnected", "ServerMessageStreamDisconnected", "ServerMessageStreamerConnected", diff --git a/fishjam/events/_protos/fishjam/__init__.py b/fishjam/events/_protos/fishjam/__init__.py index eeb8ab8..b627969 100644 --- a/fishjam/events/_protos/fishjam/__init__.py +++ b/fishjam/events/_protos/fishjam/__init__.py @@ -34,6 +34,13 @@ class ServerMessageVadNotificationStatus(betterproto.Enum): STATUS_SPEECH = 2 +class ServerMessageRecordingStatusChangedStatus(betterproto.Enum): + STATUS_ACTIVE = 0 + STATUS_FINISHED = 1 + STATUS_AVAILABLE = 2 + STATUS_FAILED = 3 + + @dataclass(eq=False, repr=False) class AgentRequest(betterproto.Message): """Defines any type of message passed from agent peer to Fishjam""" @@ -229,6 +236,9 @@ class ServerMessage(betterproto.Message): streamer_disconnected: "ServerMessageStreamerDisconnected" = ( betterproto.message_field(27, group="content") ) + recording_status_changed: "ServerMessageRecordingStatusChanged" = ( + betterproto.message_field(34, group="content") + ) notification_batch: "ServerMessageNotificationBatch" = betterproto.message_field( 33, group="content" ) @@ -549,6 +559,13 @@ class ServerMessageStreamerDisconnected(betterproto.Message): streamer_id: str = betterproto.string_field(2) +@dataclass(eq=False, repr=False) +class ServerMessageRecordingStatusChanged(betterproto.Message): + recording_id: str = betterproto.string_field(1) + status: "ServerMessageRecordingStatusChangedStatus" = betterproto.enum_field(2) + metadata: str = betterproto.string_field(3) + + @dataclass(eq=False, repr=False) class ServerMessageNotificationBatch(betterproto.Message): """ diff --git a/fishjam/events/allowed_notifications.py b/fishjam/events/allowed_notifications.py index 6b4b7d3..2ae3ee7 100644 --- a/fishjam/events/allowed_notifications.py +++ b/fishjam/events/allowed_notifications.py @@ -9,6 +9,7 @@ ServerMessagePeerDeleted, ServerMessagePeerDisconnected, ServerMessagePeerMetadataUpdated, + ServerMessageRecordingStatusChanged, ServerMessageRoomCrashed, ServerMessageRoomCreated, ServerMessageRoomDeleted, @@ -40,6 +41,7 @@ ServerMessageTrackAdded, ServerMessageTrackRemoved, ServerMessageTrackMetadataUpdated, + ServerMessageRecordingStatusChanged, ) AllowedNotification = Union[ @@ -61,4 +63,5 @@ ServerMessageTrackAdded, ServerMessageTrackRemoved, ServerMessageTrackMetadataUpdated, + ServerMessageRecordingStatusChanged, ] diff --git a/protos b/protos index 50aacf9..3538624 160000 --- a/protos +++ b/protos @@ -1 +1 @@ -Subproject commit 50aacf9839c7e1f67e983f775f04206050b99cee +Subproject commit 3538624ebaa93dc46642b3ec7c840109f758bad7 From 65ed516a2c7b4fd6845a825124fe8eb7f46ae8e5 Mon Sep 17 00:00:00 2001 From: Adrian Czerwiec Date: Thu, 13 Aug 2026 15:11:41 +0200 Subject: [PATCH 2/4] add recording client methods --- fishjam/__init__.py | 5 +- .../api/recordings/delete_recording.py | 21 +- .../api/recordings/stop_recording.py | 186 ++++++++++++++++++ .../models/recording_status.py | 1 + fishjam/api/_client.py | 9 +- fishjam/api/_fishjam_client.py | 147 ++++++++++++++ fishjam/errors.py | 17 +- fishjam/recording/__init__.py | 11 ++ tests/test_recording_api.py | 144 ++++++++++++++ 9 files changed, 531 insertions(+), 10 deletions(-) create mode 100644 fishjam/_openapi_client/api/recordings/stop_recording.py create mode 100644 fishjam/recording/__init__.py create mode 100644 tests/test_recording_api.py diff --git a/fishjam/__init__.py b/fishjam/__init__.py index 054c80c..7decd73 100644 --- a/fishjam/__init__.py +++ b/fishjam/__init__.py @@ -8,7 +8,7 @@ # pylint: disable=locally-disabled, no-name-in-module, import-error # Exceptions and Server Messages -from fishjam import agent, errors, events, integrations, peer, room, version +from fishjam import agent, errors, events, integrations, peer, recording, room, version from fishjam._openapi_client.models import PeerMetadata # API @@ -26,6 +26,7 @@ Peer, PeerOptions, PeerOptionsVapi, + Recording, Room, RoomOptions, ) @@ -47,6 +48,7 @@ "AgentOutputOptions", "Room", "Peer", + "Recording", "MoqAccess", "MissingFishjamIdError", "InvalidFishjamCredentialsError", @@ -54,6 +56,7 @@ "errors", "room", "peer", + "recording", "agent", "integrations", ] diff --git a/fishjam/_openapi_client/api/recordings/delete_recording.py b/fishjam/_openapi_client/api/recordings/delete_recording.py index cba80de..e3bce76 100644 --- a/fishjam/_openapi_client/api/recordings/delete_recording.py +++ b/fishjam/_openapi_client/api/recordings/delete_recording.py @@ -35,6 +35,11 @@ def _parse_response( return response_401 + if response.status_code == 409: + response_409 = Error.from_dict(response.json()) + + return response_409 + if response.status_code == 503: response_503 = Error.from_dict(response.json()) @@ -64,7 +69,9 @@ def sync_detailed( ) -> Response[Any | Error]: """Delete a recording - Delete a recording by id. + Delete a recording by id. The recording disappears from the API immediately; its stored media is + removed asynchronously by a background job. A recording that is still `active` cannot be deleted — + wait for it to finish or fail. Args: recording_id (str): @@ -95,7 +102,9 @@ def sync( ) -> Any | Error | None: """Delete a recording - Delete a recording by id. + Delete a recording by id. The recording disappears from the API immediately; its stored media is + removed asynchronously by a background job. A recording that is still `active` cannot be deleted — + wait for it to finish or fail. Args: recording_id (str): @@ -121,7 +130,9 @@ async def asyncio_detailed( ) -> Response[Any | Error]: """Delete a recording - Delete a recording by id. + Delete a recording by id. The recording disappears from the API immediately; its stored media is + removed asynchronously by a background job. A recording that is still `active` cannot be deleted — + wait for it to finish or fail. Args: recording_id (str): @@ -150,7 +161,9 @@ async def asyncio( ) -> Any | Error | None: """Delete a recording - Delete a recording by id. + Delete a recording by id. The recording disappears from the API immediately; its stored media is + removed asynchronously by a background job. A recording that is still `active` cannot be deleted — + wait for it to finish or fail. Args: recording_id (str): diff --git a/fishjam/_openapi_client/api/recordings/stop_recording.py b/fishjam/_openapi_client/api/recordings/stop_recording.py new file mode 100644 index 0000000..6b082cd --- /dev/null +++ b/fishjam/_openapi_client/api/recordings/stop_recording.py @@ -0,0 +1,186 @@ +from http import HTTPStatus +from typing import Any +from urllib.parse import quote + +import httpx + +from ... import errors +from ...client import AuthenticatedClient, Client +from ...models.error import Error +from ...models.recording_details_response import RecordingDetailsResponse +from ...types import Response + + +def _get_kwargs( + recording_id: str, +) -> dict[str, Any]: + _kwargs: dict[str, Any] = { + "method": "post", + "url": "/recordings/{recording_id}/stop".format( + recording_id=quote(str(recording_id), safe=""), + ), + } + + return _kwargs + + +def _parse_response( + *, client: AuthenticatedClient | Client, response: httpx.Response +) -> Error | RecordingDetailsResponse | None: + if response.status_code == 200: + response_200 = RecordingDetailsResponse.from_dict(response.json()) + + return response_200 + + if response.status_code == 401: + response_401 = Error.from_dict(response.json()) + + return response_401 + + if response.status_code == 404: + response_404 = Error.from_dict(response.json()) + + return response_404 + + if response.status_code == 503: + response_503 = Error.from_dict(response.json()) + + return response_503 + + if client.raise_on_unexpected_status: + raise errors.UnexpectedStatus(response.status_code, response.content) + else: + return None + + +def _build_response( + *, client: AuthenticatedClient | Client, response: httpx.Response +) -> Response[Error | RecordingDetailsResponse]: + return Response( + status_code=HTTPStatus(response.status_code), + content=response.content, + headers=response.headers, + parsed=_parse_response(client=client, response=response), + ) + + +def sync_detailed( + recording_id: str, + *, + client: AuthenticatedClient, +) -> Response[Error | RecordingDetailsResponse]: + """Stop a recording + + Request the recorder to stop capturing. Finalization is asynchronous: the recording stays `active` + until the capture is finalized, then becomes `finished`. Stopping a recording that is no longer + active is a no-op. + + Args: + recording_id (str): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Response[Error | RecordingDetailsResponse] + """ + + kwargs = _get_kwargs( + recording_id=recording_id, + ) + + response = client.get_httpx_client().request( + **kwargs, + ) + + return _build_response(client=client, response=response) + + +def sync( + recording_id: str, + *, + client: AuthenticatedClient, +) -> Error | RecordingDetailsResponse | None: + """Stop a recording + + Request the recorder to stop capturing. Finalization is asynchronous: the recording stays `active` + until the capture is finalized, then becomes `finished`. Stopping a recording that is no longer + active is a no-op. + + Args: + recording_id (str): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Error | RecordingDetailsResponse + """ + + return sync_detailed( + recording_id=recording_id, + client=client, + ).parsed + + +async def asyncio_detailed( + recording_id: str, + *, + client: AuthenticatedClient, +) -> Response[Error | RecordingDetailsResponse]: + """Stop a recording + + Request the recorder to stop capturing. Finalization is asynchronous: the recording stays `active` + until the capture is finalized, then becomes `finished`. Stopping a recording that is no longer + active is a no-op. + + Args: + recording_id (str): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Response[Error | RecordingDetailsResponse] + """ + + kwargs = _get_kwargs( + recording_id=recording_id, + ) + + response = await client.get_async_httpx_client().request(**kwargs) + + return _build_response(client=client, response=response) + + +async def asyncio( + recording_id: str, + *, + client: AuthenticatedClient, +) -> Error | RecordingDetailsResponse | None: + """Stop a recording + + Request the recorder to stop capturing. Finalization is asynchronous: the recording stays `active` + until the capture is finalized, then becomes `finished`. Stopping a recording that is no longer + active is a no-op. + + Args: + recording_id (str): + + Raises: + errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. + httpx.TimeoutException: If the request takes longer than Client.timeout. + + Returns: + Error | RecordingDetailsResponse + """ + + return ( + await asyncio_detailed( + recording_id=recording_id, + client=client, + ) + ).parsed diff --git a/fishjam/_openapi_client/models/recording_status.py b/fishjam/_openapi_client/models/recording_status.py index 60da974..9ae43c8 100644 --- a/fishjam/_openapi_client/models/recording_status.py +++ b/fishjam/_openapi_client/models/recording_status.py @@ -7,6 +7,7 @@ class RecordingStatus(str, Enum): ACTIVE = "active" AVAILABLE = "available" FAILED = "failed" + FINISHED = "finished" def __str__(self) -> str: return str(self.value) diff --git a/fishjam/api/_client.py b/fishjam/api/_client.py index 4a30e7a..10541ac 100644 --- a/fishjam/api/_client.py +++ b/fishjam/api/_client.py @@ -1,5 +1,6 @@ import json import warnings +from http import HTTPStatus from typing import cast from fishjam._openapi_client.client import AuthenticatedClient @@ -24,7 +25,13 @@ def _request(self, method, **kwargs): response = method.sync_detailed(client=self.client, **kwargs) self._handle_deprecation_header(response.headers) - if isinstance(response.parsed, Error): + # `parsed` is None for error statuses the endpoint spec doesn't + # document, so check the status code as well to never report + # success for a failed request + if ( + isinstance(response.parsed, Error) + or response.status_code >= HTTPStatus.BAD_REQUEST + ): response = cast(Response[Error], response) raise HTTPError.from_response(response) diff --git a/fishjam/api/_fishjam_client.py b/fishjam/api/_fishjam_client.py index eddaad6..b7c352e 100644 --- a/fishjam/api/_fishjam_client.py +++ b/fishjam/api/_fishjam_client.py @@ -10,6 +10,21 @@ from fishjam._openapi_client.api.mo_q import ( create_moq_access as moq_create_access, ) +from fishjam._openapi_client.api.recordings import ( + create_recording as recording_create_recording, +) +from fishjam._openapi_client.api.recordings import ( + delete_recording as recording_delete_recording, +) +from fishjam._openapi_client.api.recordings import ( + get_recording as recording_get_recording, +) +from fishjam._openapi_client.api.recordings import ( + list_recordings as recording_list_recordings, +) +from fishjam._openapi_client.api.recordings import ( + stop_recording as recording_stop_recording, +) from fishjam._openapi_client.api.rooms import add_peer as room_add_peer from fishjam._openapi_client.api.rooms import create_room as room_create_room from fishjam._openapi_client.api.rooms import delete_peer as room_delete_peer @@ -29,6 +44,8 @@ AgentOutput, AudioFormat, AudioSampleRate, + CompositionSource, + ListRecordingsMetadata, MoqAccess, MoqAccessConfig, Peer, @@ -43,6 +60,11 @@ PeerOptionsVapi, PeerOptionsWebRTC, PeerRefreshTokenResponse, + Recording, + RecordingConfig, + RecordingConfigMetadataType0, + RecordingDetailsResponse, + RecordingListResponse, RoomConfig, RoomCreateDetailsResponse, RoomDetailsResponse, @@ -457,6 +479,117 @@ def create_moq_access( return response + def create_recording( + self, + source: CompositionSource, + metadata: dict[str, Any] | None = None, + ) -> Recording: + """Creates a new recording. + + Capturing starts synchronously, so the returned recording is `active`. + + Args: + source: The source of the recording. + metadata: Free-form metadata used to organize and filter recordings. + + Returns: + Recording: The created recording. + """ + if metadata is None: + config_metadata = UNSET + else: + config_metadata = RecordingConfigMetadataType0() + for key, value in metadata.items(): + config_metadata.additional_properties[key] = value + + config = RecordingConfig(source=source, metadata=config_metadata) + + resp = cast( + RecordingDetailsResponse, + self._request(recording_create_recording, body=config), + ) + + return resp.data + + def get_recording(self, recording_id: str) -> Recording: + """Returns the recording with the given id. + + Args: + recording_id: The ID of the recording to retrieve. + + Returns: + Recording: The recording corresponding to the given ID. + """ + resp = cast( + RecordingDetailsResponse, + self._request(recording_get_recording, recording_id=recording_id), + ) + + return resp.data + + def get_all_recordings( + self, metadata: dict[str, Any] | None = None + ) -> list[Recording]: + """Returns a list of all recordings, optionally filtered by metadata. + + Args: + metadata: If given, only recordings whose metadata contains all + the given key-value pairs are returned. Nested dicts match + nested metadata keys. + + Returns: + list[Recording]: A list of all matching recordings. + """ + # the API expects the deepObject query format + # (`metadata[key]=value`, `metadata[key][nested]=value`), but the + # generated client serializes the filter keys at the top level, so + # prefix and flatten them here + if metadata is None: + metadata_query = UNSET + else: + metadata_query = ListRecordingsMetadata() + self.__flatten_metadata_filter( + "metadata", metadata, metadata_query.additional_properties + ) + + resp = cast( + RecordingListResponse, + self._request(recording_list_recordings, metadata=metadata_query), + ) + + return resp.data + + def stop_recording(self, recording_id: str) -> Recording: + """Stops an active recording. + + Finalization is asynchronous: the recording stays `active` until the + capture is finalized, then becomes `finished`. Stopping a recording + that is no longer active is a no-op. + + Args: + recording_id: The ID of the recording to stop. + + Returns: + Recording: The stopped recording. + """ + resp = cast( + RecordingDetailsResponse, + self._request(recording_stop_recording, recording_id=recording_id), + ) + + return resp.data + + def delete_recording(self, recording_id: str) -> None: + """Deletes a recording. Its stored media is removed asynchronously. + + A recording that is still `active` cannot be deleted — stop it first + or wait for it to finish. + + Args: + recording_id: The ID of the recording to delete. + """ + self._request(recording_delete_recording, recording_id=recording_id) + def subscribe_peer(self, room_id: str, peer_id: str, target_peer_id: str): """Subscribes a peer to all tracks of another peer. @@ -487,6 +620,20 @@ def subscribe_tracks(self, room_id: str, peer_id: str, track_ids: list[str]): body=SubscribeTracksBody(track_ids=track_ids), ) + def __flatten_metadata_filter( + self, prefix: str, metadata: dict[str, Any], params: dict[str, Any] + ) -> None: + for key, value in metadata.items(): + param_key = f"{prefix}[{key}]" + if isinstance(value, dict): + self.__flatten_metadata_filter(param_key, value, params) + elif value is None: + # the generated client drops `None` params; the API compares + # values as JSON strings, so send the JSON representation + params[param_key] = "null" + else: + params[param_key] = value + def __parse_peer_metadata(self, metadata: dict | None) -> WebRTCMetadata: peer_metadata = WebRTCMetadata() diff --git a/fishjam/errors.py b/fishjam/errors.py index 620b4ac..b36963b 100644 --- a/fishjam/errors.py +++ b/fishjam/errors.py @@ -15,10 +15,10 @@ class HTTPError(Exception): @staticmethod def from_response(response: Response[Error]): """@private""" - if not response.parsed: - raise RuntimeError("Got endpoint error reponse without parsed field") - - errors = response.parsed.errors + if response.parsed: + errors = response.parsed.errors + else: + errors = response.content.decode(errors="replace") match response.status_code: case HTTPStatus.BAD_REQUEST: @@ -27,6 +27,9 @@ def from_response(response: Response[Error]): case HTTPStatus.UNAUTHORIZED: return UnauthorizedError(errors) + case HTTPStatus.PAYMENT_REQUIRED: + return QuotaExceededError(errors) + case HTTPStatus.NOT_FOUND: return NotFoundError(errors) @@ -76,6 +79,12 @@ def __init__(self, errors): super().__init__(errors) +class QuotaExceededError(HTTPError): + def __init__(self, errors): + """@private""" + super().__init__(errors) + + class InvalidFishjamCredentialsError(HTTPError): def __init__(self, errors): """@private""" diff --git a/fishjam/recording/__init__.py b/fishjam/recording/__init__.py new file mode 100644 index 0000000..7dc5d3f --- /dev/null +++ b/fishjam/recording/__init__.py @@ -0,0 +1,11 @@ +from fishjam._openapi_client.models import ( + CompositionSource, + Recording, + RecordingStatus, +) + +__all__ = [ + "CompositionSource", + "Recording", + "RecordingStatus", +] diff --git a/tests/test_recording_api.py b/tests/test_recording_api.py new file mode 100644 index 0000000..21d56ff --- /dev/null +++ b/tests/test_recording_api.py @@ -0,0 +1,144 @@ +import json +from unittest.mock import patch + +import httpx +import pytest + +from fishjam import FishjamClient, Recording +from fishjam.errors import ( + InternalServerError, + NotFoundError, + QuotaExceededError, + ServiceUnavailableError, + UnauthorizedError, +) +from fishjam.recording import CompositionSource, RecordingStatus +from tests.support.env import FISHJAM_ID, FISHJAM_MANAGEMENT_TOKEN + +NONEXISTENT_RECORDING_ID = "515c8b52-168b-4b39-a227-4d6b4f102a56" + + +@pytest.fixture +def recording_api(): + return FishjamClient(FISHJAM_ID, FISHJAM_MANAGEMENT_TOKEN) + + +def mock_request(status_code: int, json_body): + captured_requests = [] + + def mock_send(request, **kwargs): + captured_requests.append(request) + return httpx.Response(status_code, json=json_body, request=request) + + return captured_requests, patch.object( + httpx.HTTPTransport, "handle_request", side_effect=mock_send + ) + + +class TestGetAllRecordings: + def test_returns_list(self, recording_api: FishjamClient): + recordings = recording_api.get_all_recordings() + + assert isinstance(recordings, list) + + def test_unauthorized(self): + recording_api = FishjamClient(FISHJAM_ID, "invalid") + + with pytest.raises(UnauthorizedError): + recording_api.get_all_recordings() + + def test_metadata_filter_uses_deep_object_format( + self, recording_api: FishjamClient + ): + captured_requests, request_patch = mock_request(200, {"data": []}) + + with request_patch: + recording_api.get_all_recordings(metadata={"env": "prod"}) + + assert len(captured_requests) == 1 + params = captured_requests[0].url.params + assert params.get("metadata[env]") == "prod" + + def test_metadata_filter_supports_nested_keys_and_none( + self, recording_api: FishjamClient + ): + captured_requests, request_patch = mock_request(200, {"data": []}) + + with request_patch: + recording_api.get_all_recordings(metadata={"a": {"b": "c"}, "env": None}) + + params = captured_requests[0].url.params + assert params.get("metadata[a][b]") == "c" + # values are compared as JSON strings by the API, so None must be + # sent as "null" instead of being dropped from the query + assert params.get("metadata[env]") == "null" + + def test_service_unavailable(self, recording_api: FishjamClient): + _, request_patch = mock_request(503, {"errors": "service unavailable"}) + + with request_patch, pytest.raises(ServiceUnavailableError): + recording_api.get_all_recordings() + + +class TestCreateRecording: + def test_returns_created_recording(self, recording_api: FishjamClient): + source = CompositionSource( + composition_url="https://example.com/composition", + output_id="output-1", + ) + recording_json = { + "id": NONEXISTENT_RECORDING_ID, + "source": source.to_dict(), + "status": "active", + "metadata": {"env": "test"}, + } + captured_requests, request_patch = mock_request(201, {"data": recording_json}) + + with request_patch: + recording = recording_api.create_recording(source, metadata={"env": "test"}) + + assert isinstance(recording, Recording) + assert recording.id == NONEXISTENT_RECORDING_ID + assert recording.status == RecordingStatus.ACTIVE + + assert len(captured_requests) == 1 + request = captured_requests[0] + assert request.method == "POST" + assert request.url.path.endswith("/recordings") + assert json.loads(request.content) == { + "source": source.to_dict(), + "metadata": {"env": "test"}, + } + + def test_quota_exceeded(self, recording_api: FishjamClient): + source = CompositionSource( + composition_url="https://example.com/composition", + output_id="output-1", + ) + _, request_patch = mock_request(402, {"errors": "quota exceeded"}) + + with request_patch, pytest.raises(QuotaExceededError): + recording_api.create_recording(source) + + +class TestGetRecording: + def test_id_not_found(self, recording_api: FishjamClient): + with pytest.raises(NotFoundError): + recording_api.get_recording(NONEXISTENT_RECORDING_ID) + + +class TestStopRecording: + def test_id_not_found(self, recording_api: FishjamClient): + with pytest.raises(NotFoundError): + recording_api.stop_recording(NONEXISTENT_RECORDING_ID) + + +class TestDeleteRecording: + def test_nonexistent_id_is_noop(self, recording_api: FishjamClient): + recording_api.delete_recording(NONEXISTENT_RECORDING_ID) + + def test_server_error_raises(self, recording_api: FishjamClient): + _, request_patch = mock_request(500, {"errors": "internal error"}) + + with request_patch, pytest.raises(InternalServerError): + recording_api.delete_recording(NONEXISTENT_RECORDING_ID) From adc9676b3dbeb1f14f2ff0daef57cffa9c6495ad Mon Sep 17 00:00:00 2001 From: Adrian Czerwiec Date: Thu, 13 Aug 2026 15:23:50 +0200 Subject: [PATCH 3/4] add recording success-path and notification decode tests Address PR review comments: cover get_recording/stop_recording 200 responses and the recording status notification webhook round-trip. Co-Authored-By: Claude Fable 5 --- tests/test_recording_api.py | 71 ++++++++++++++++++++++++++-------- tests/test_webhook_notifier.py | 24 ++++++++++++ 2 files changed, 79 insertions(+), 16 deletions(-) diff --git a/tests/test_recording_api.py b/tests/test_recording_api.py index 21d56ff..5456724 100644 --- a/tests/test_recording_api.py +++ b/tests/test_recording_api.py @@ -16,6 +16,7 @@ from tests.support.env import FISHJAM_ID, FISHJAM_MANAGEMENT_TOKEN NONEXISTENT_RECORDING_ID = "515c8b52-168b-4b39-a227-4d6b4f102a56" +RECORDING_ID = "8e9b40aa-27d5-4e05-b6c1-27eb85f603f7" @pytest.fixture @@ -23,6 +24,21 @@ def recording_api(): return FishjamClient(FISHJAM_ID, FISHJAM_MANAGEMENT_TOKEN) +def make_composition_source(): + return CompositionSource( + composition_url="https://example.com/composition", + output_id="output-1", + ) + + +def make_recording_json(source: CompositionSource, status: str): + return { + "id": RECORDING_ID, + "source": source.to_dict(), + "status": status, + } + + def mock_request(status_code: int, json_body): captured_requests = [] @@ -82,23 +98,16 @@ def test_service_unavailable(self, recording_api: FishjamClient): class TestCreateRecording: def test_returns_created_recording(self, recording_api: FishjamClient): - source = CompositionSource( - composition_url="https://example.com/composition", - output_id="output-1", - ) - recording_json = { - "id": NONEXISTENT_RECORDING_ID, - "source": source.to_dict(), - "status": "active", - "metadata": {"env": "test"}, - } + source = make_composition_source() + recording_json = make_recording_json(source, "active") + recording_json["metadata"] = {"env": "test"} captured_requests, request_patch = mock_request(201, {"data": recording_json}) with request_patch: recording = recording_api.create_recording(source, metadata={"env": "test"}) assert isinstance(recording, Recording) - assert recording.id == NONEXISTENT_RECORDING_ID + assert recording.id == RECORDING_ID assert recording.status == RecordingStatus.ACTIVE assert len(captured_requests) == 1 @@ -111,23 +120,53 @@ def test_returns_created_recording(self, recording_api: FishjamClient): } def test_quota_exceeded(self, recording_api: FishjamClient): - source = CompositionSource( - composition_url="https://example.com/composition", - output_id="output-1", - ) _, request_patch = mock_request(402, {"errors": "quota exceeded"}) with request_patch, pytest.raises(QuotaExceededError): - recording_api.create_recording(source) + recording_api.create_recording(make_composition_source()) class TestGetRecording: + def test_returns_recording(self, recording_api: FishjamClient): + source = make_composition_source() + recording_json = make_recording_json(source, "available") + captured_requests, request_patch = mock_request(200, {"data": recording_json}) + + with request_patch: + recording = recording_api.get_recording(RECORDING_ID) + + assert isinstance(recording, Recording) + assert recording.id == RECORDING_ID + assert recording.status == RecordingStatus.AVAILABLE + assert recording.source == source + + request = captured_requests[0] + assert request.method == "GET" + assert request.url.path.endswith(f"/recordings/{RECORDING_ID}") + def test_id_not_found(self, recording_api: FishjamClient): with pytest.raises(NotFoundError): recording_api.get_recording(NONEXISTENT_RECORDING_ID) class TestStopRecording: + def test_returns_stopped_recording(self, recording_api: FishjamClient): + source = make_composition_source() + # the recording stays `active` until finalization completes + recording_json = make_recording_json(source, "active") + captured_requests, request_patch = mock_request(200, {"data": recording_json}) + + with request_patch: + recording = recording_api.stop_recording(RECORDING_ID) + + assert isinstance(recording, Recording) + assert recording.id == RECORDING_ID + assert recording.status == RecordingStatus.ACTIVE + + request = captured_requests[0] + assert request.method == "POST" + assert request.url.path.endswith(f"/recordings/{RECORDING_ID}/stop") + def test_id_not_found(self, recording_api: FishjamClient): with pytest.raises(NotFoundError): recording_api.stop_recording(NONEXISTENT_RECORDING_ID) diff --git a/tests/test_webhook_notifier.py b/tests/test_webhook_notifier.py index 9cebd4a..e537e3a 100644 --- a/tests/test_webhook_notifier.py +++ b/tests/test_webhook_notifier.py @@ -9,6 +9,8 @@ ) from fishjam.events import ( ServerMessagePeerConnected, + ServerMessageRecordingStatusChanged, + ServerMessageRecordingStatusChangedStatus, ServerMessageRoomCreated, ServerMessageRoomDeleted, ) @@ -179,6 +181,28 @@ def test_decode_empty_batch_returns_empty_list(): assert decode_server_notifications(binary) == [] +def test_decode_recording_status_changed_round_trip(): + binary = bytes( + ServerMessage( + recording_status_changed=ServerMessageRecordingStatusChanged( + recording_id="rec1", + status=ServerMessageRecordingStatusChangedStatus.STATUS_FINISHED, + metadata='{"session": "s1"}', + ) + ) + ) + + result = decode_server_notifications(binary) + + assert [type(n) for n in result] == [ServerMessageRecordingStatusChanged] + notification = result[0] + assert notification.recording_id == "rec1" + assert ( + notification.status == ServerMessageRecordingStatusChangedStatus.STATUS_FINISHED + ) + assert notification.metadata == '{"session": "s1"}' + + BODY = bytes(ServerMessage(room_created=ServerMessageRoomCreated(room_id="r1"))) SECRET = "webhook-secret" SIGNATURE = hmac.new(SECRET.encode(), BODY, "sha256").hexdigest() From ab06bfa845683e05752eb885fda03a3261ba7a26 Mon Sep 17 00:00:00 2001 From: Adrian Czerwiec Date: Fri, 14 Aug 2026 15:53:50 +0200 Subject: [PATCH 4/4] validate unspecified status MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bump protos to 6d469f9 (recording status enum gains STATUS_UNSPECIFIED) and raise StaleSdkError when a recording status notification carries UNSPECIFIED or an unknown wire value — both mean this SDK is likely too old to parse the statuses the server sends. Mirrors the js-server-sdk change. Co-Authored-By: Claude Fable 5 --- fishjam/__init__.py | 7 +++- fishjam/_webhook_notifier.py | 8 ++++ fishjam/_ws_notifier.py | 2 + fishjam/errors.py | 11 ++++++ fishjam/events/_protos/fishjam/__init__.py | 9 +++-- fishjam/events/allowed_notifications.py | 14 +++++++ protos | 2 +- tests/test_webhook_notifier.py | 46 ++++++++++++++++++++++ 8 files changed, 93 insertions(+), 6 deletions(-) diff --git a/fishjam/__init__.py b/fishjam/__init__.py index 7decd73..d3e2017 100644 --- a/fishjam/__init__.py +++ b/fishjam/__init__.py @@ -30,7 +30,11 @@ Room, RoomOptions, ) -from fishjam.errors import InvalidFishjamCredentialsError, MissingFishjamIdError +from fishjam.errors import ( + InvalidFishjamCredentialsError, + MissingFishjamIdError, + StaleSdkError, +) __version__ = version.__version__ @@ -52,6 +56,7 @@ "MoqAccess", "MissingFishjamIdError", "InvalidFishjamCredentialsError", + "StaleSdkError", "events", "errors", "room", diff --git a/fishjam/_webhook_notifier.py b/fishjam/_webhook_notifier.py index 3b75edb..99c0c13 100644 --- a/fishjam/_webhook_notifier.py +++ b/fishjam/_webhook_notifier.py @@ -13,6 +13,7 @@ from fishjam.events.allowed_notifications import ( ALLOWED_NOTIFICATIONS, AllowedNotification, + validate_notification, ) @@ -20,6 +21,7 @@ def _content_of(message: ServerMessage) -> Union[AllowedNotification, None]: """Return the message's `content` oneof if it is a supported notification.""" _which, content = betterproto.which_one_of(message, "content") if isinstance(content, ALLOWED_NOTIFICATIONS): + validate_notification(content) return content return None @@ -57,6 +59,10 @@ def decode_server_notifications(binary: bytes) -> List[AllowedNotification]: Returns: list[AllowedNotification]: The decoded notifications, in order. Empty when the payload carries no supported notification. + + Raises: + fishjam.errors.StaleSdkError: When a notification carries a value this + SDK cannot parse, which likely means the SDK is outdated. """ message = ServerMessage().parse(binary) _which, content = betterproto.which_one_of(message, "content") @@ -65,6 +71,7 @@ def decode_server_notifications(binary: bytes) -> List[AllowedNotification]: return _unpack_batch(content) if isinstance(content, ALLOWED_NOTIFICATIONS): + validate_notification(content) return [content] return [] @@ -123,6 +130,7 @@ def receive_binary( return _unpack_batch(content) if isinstance(content, ALLOWED_NOTIFICATIONS): + validate_notification(content) return content return None diff --git a/fishjam/_ws_notifier.py b/fishjam/_ws_notifier.py index bfbd2da..4d12cfe 100644 --- a/fishjam/_ws_notifier.py +++ b/fishjam/_ws_notifier.py @@ -20,6 +20,7 @@ from fishjam.events.allowed_notifications import ( ALLOWED_NOTIFICATIONS, AllowedNotification, + validate_notification, ) from fishjam.utils import get_fishjam_url @@ -135,6 +136,7 @@ async def _receive_loop(self): _which, message = betterproto.which_one_of(message, "content") if isinstance(message, ALLOWED_NOTIFICATIONS): + validate_notification(message) res = self._notification_handler(message) if inspect.isawaitable(res): await res diff --git a/fishjam/errors.py b/fishjam/errors.py index b36963b..9202ea4 100644 --- a/fishjam/errors.py +++ b/fishjam/errors.py @@ -9,6 +9,17 @@ def __init__(self) -> None: super().__init__("Fishjam ID is required") +class StaleSdkError(Exception): + def __init__(self, status: int) -> None: + super().__init__( + f"Received a recording status this SDK cannot parse ({int(status)})." + " You are probably using an outdated version of fishjam-server-sdk" + " - please update it." + ) + self.status = int(status) + """Raw wire value received from the server.""" + + class HTTPError(Exception): """""" diff --git a/fishjam/events/_protos/fishjam/__init__.py b/fishjam/events/_protos/fishjam/__init__.py index b627969..c8325fd 100644 --- a/fishjam/events/_protos/fishjam/__init__.py +++ b/fishjam/events/_protos/fishjam/__init__.py @@ -35,10 +35,11 @@ class ServerMessageVadNotificationStatus(betterproto.Enum): class ServerMessageRecordingStatusChangedStatus(betterproto.Enum): - STATUS_ACTIVE = 0 - STATUS_FINISHED = 1 - STATUS_AVAILABLE = 2 - STATUS_FAILED = 3 + STATUS_UNSPECIFIED = 0 + STATUS_ACTIVE = 1 + STATUS_FINISHED = 2 + STATUS_AVAILABLE = 3 + STATUS_FAILED = 4 @dataclass(eq=False, repr=False) diff --git a/fishjam/events/allowed_notifications.py b/fishjam/events/allowed_notifications.py index 2ae3ee7..9cf7777 100644 --- a/fishjam/events/allowed_notifications.py +++ b/fishjam/events/allowed_notifications.py @@ -1,5 +1,6 @@ from typing import Union +from fishjam.errors import StaleSdkError from fishjam.events import ( ServerMessageChannelAdded, ServerMessageChannelRemoved, @@ -10,6 +11,7 @@ ServerMessagePeerDisconnected, ServerMessagePeerMetadataUpdated, ServerMessageRecordingStatusChanged, + ServerMessageRecordingStatusChangedStatus, ServerMessageRoomCrashed, ServerMessageRoomCreated, ServerMessageRoomDeleted, @@ -65,3 +67,15 @@ ServerMessageTrackMetadataUpdated, ServerMessageRecordingStatusChanged, ] + + +# Raises instead of falling back: STATUS_UNSPECIFIED or an unknown wire value +# both mean this SDK is likely too old to parse the statuses the server sends. +def validate_notification(notification: AllowedNotification) -> None: + if isinstance(notification, ServerMessageRecordingStatusChanged): + try: + status = ServerMessageRecordingStatusChangedStatus(notification.status) + except ValueError: + raise StaleSdkError(notification.status) from None + if status == ServerMessageRecordingStatusChangedStatus.STATUS_UNSPECIFIED: + raise StaleSdkError(status) diff --git a/protos b/protos index 3538624..6d469f9 160000 --- a/protos +++ b/protos @@ -1 +1 @@ -Subproject commit 3538624ebaa93dc46642b3ec7c840109f758bad7 +Subproject commit 6d469f99245b8f7fc0074c9cabfbcdaed8b92977 diff --git a/tests/test_webhook_notifier.py b/tests/test_webhook_notifier.py index e537e3a..e260870 100644 --- a/tests/test_webhook_notifier.py +++ b/tests/test_webhook_notifier.py @@ -7,6 +7,7 @@ receive_binary, verify_webhook_signature, ) +from fishjam.errors import StaleSdkError from fishjam.events import ( ServerMessagePeerConnected, ServerMessageRecordingStatusChanged, @@ -203,6 +204,51 @@ def test_decode_recording_status_changed_round_trip(): assert notification.metadata == '{"session": "s1"}' +@pytest.mark.parametrize( + "status", + [ + ServerMessageRecordingStatusChangedStatus.STATUS_UNSPECIFIED, + # A status added in a newer proto than this SDK was generated from + # arrives as its raw wire value. + 42, + ], +) +def test_decode_raises_on_unparsable_recording_status(status): + binary = bytes( + ServerMessage( + recording_status_changed=ServerMessageRecordingStatusChanged( + recording_id="rec1", status=status, metadata="" + ) + ) + ) + + with pytest.raises(StaleSdkError): + decode_server_notifications(binary) + + +def test_decode_batch_raises_on_unparsable_recording_status(): + binary = bytes( + ServerMessage( + notification_batch=ServerMessageNotificationBatch( + notifications=[ + ServerMessage( + recording_status_changed=ServerMessageRecordingStatusChanged( + recording_id="rec1", + status=( + ServerMessageRecordingStatusChangedStatus.STATUS_UNSPECIFIED + ), + metadata="", + ) + ), + ] + ) + ) + ) + + with pytest.raises(StaleSdkError): + decode_server_notifications(binary) + + BODY = bytes(ServerMessage(room_created=ServerMessageRoomCreated(room_id="r1"))) SECRET = "webhook-secret" SIGNATURE = hmac.new(SECRET.encode(), BODY, "sha256").hexdigest()