diff --git a/README.md b/README.md index 0351687..3506b08 100644 --- a/README.md +++ b/README.md @@ -154,7 +154,6 @@ Two entry points live in `divera247.websocket`: ```python from divera247.websocket import ( - ClusterPullEvent, UnknownEvent, UserStatusEvent, stream_websocket, @@ -164,9 +163,7 @@ async for event in stream_websocket(client, ucr_id=527_459): match event: case UserStatusEvent(payload=payload, ucr=ucr): ... - case ClusterPullEvent(pull=pull, cluster=cluster): - ... - case UnknownEvent(type=msg_type): + case UnknownEvent(): ... # forward-compatible fallback ``` diff --git a/src/divera247/websocket/__init__.py b/src/divera247/websocket/__init__.py index 9cdc26b..9f705ce 100644 --- a/src/divera247/websocket/__init__.py +++ b/src/divera247/websocket/__init__.py @@ -11,17 +11,18 @@ JWT re-auth. * :class:`WebSocketAuthenticationError` - raised when authentication keeps failing so the caller can react instead of silently looping. -* Pydantic event envelopes (:class:`ClusterPullEvent`, - :class:`UserStatusEvent`, :class:`UnknownEvent`) and the - :data:`DiveraEvent` discriminated union plus :func:`parse_event` for - dispatching raw frames onto typed models. +* Pydantic event envelopes (:class:`UserStatusEvent`, + :class:`UnknownEvent`) and the :data:`DiveraEvent` discriminated union + plus :func:`parse_event` for dispatching raw frames onto typed models. + Other server-side event types (e.g. ``cluster-pull``, + ``cluster-vehicle``) currently surface as :class:`UnknownEvent` until a + real sample is available to back a dedicated model. Typical usage: .. code-block:: python from divera247.websocket import ( - ClusterPullEvent, UnknownEvent, UserStatusEvent, subscribe_websocket, @@ -31,15 +32,11 @@ match event: case UserStatusEvent(payload=payload, ucr=ucr): ... - case ClusterPullEvent(pull=pull): - ... case UnknownEvent(type=msg_type): ... """ from divera247.websocket.models import ( - ClusterPullEvent, - ClusterPullRef, DiveraEvent, UnknownEvent, UserStatusEvent, @@ -52,8 +49,6 @@ ) __all__ = [ - 'ClusterPullEvent', - 'ClusterPullRef', 'DiveraEvent', 'UnknownEvent', 'UserStatusEvent', diff --git a/src/divera247/websocket/models.py b/src/divera247/websocket/models.py index b614968..254c05a 100644 --- a/src/divera247/websocket/models.py +++ b/src/divera247/websocket/models.py @@ -1,12 +1,23 @@ """Pydantic models for Divera 24/7 WebSocket push events.""" +import logging from collections.abc import Mapping from typing import Annotated, Any, Literal -from pydantic import BaseModel, ConfigDict, Discriminator, Field, Tag, TypeAdapter +from pydantic import ( + BaseModel, + ConfigDict, + Discriminator, + Field, + Tag, + TypeAdapter, + ValidationError, +) from divera247.models.pull import PullStatusData +logger = logging.getLogger(__name__) + class UserStatusEvent(BaseModel): """``user-status`` WebSocket event: own status changed for a given UCR. @@ -22,33 +33,15 @@ class UserStatusEvent(BaseModel): ucr: int = Field(description='ID der betroffenen UserClusterRelation') -class ClusterPullRef(BaseModel): - """Reference to the specific cluster sub-resource that changed.""" - - type: str = Field(description='Name des betroffenen Pull-Blocks') - id: int = Field(description='ID des geänderten Eintrags') - - -class ClusterPullEvent(BaseModel): - """``cluster-pull`` WebSocket event: a cluster resource was updated. - - Clients use this as a hint to re-fetch the affected block via - ``GET /api/v2/pull/all`` (or the matching scoped endpoint) to obtain the - new state. The event itself only carries the reference, not the payload. - """ - - type: Literal['cluster-pull'] = Field(description='Event-Typ') - pull: ClusterPullRef = Field(description='Referenz auf das geänderte Element') - cluster: int = Field(description='ID der betroffenen Einheit') - - class UnknownEvent(BaseModel): """Fallback for any WebSocket event type we don't have a dedicated model for. Keeps the raw ``type`` string so callers can still dispatch on it, and preserves every other top-level field as extras (accessible via - :attr:`model_extra` or direct attribute access). Use this to log - previously unseen event types so dedicated models can be added later. + :attr:`model_extra` or direct attribute access). Used both for genuinely + unknown event types (e.g. ``cluster-pull``, ``cluster-vehicle``) and as + a defensive fallback when a known event's inner payload fails its + dedicated validation (see :func:`parse_event`). """ model_config = ConfigDict(extra='allow') @@ -56,7 +49,7 @@ class UnknownEvent(BaseModel): type: str = Field(description='Raw event type as sent by the server') -_KNOWN_EVENT_TYPES: frozenset[str] = frozenset({'user-status', 'cluster-pull'}) +_KNOWN_EVENT_TYPES: frozenset[str] = frozenset({'user-status'}) def _event_discriminator(value: Any) -> str: @@ -73,9 +66,7 @@ def _event_discriminator(value: Any) -> str: DiveraEvent = Annotated[ - Annotated[UserStatusEvent, Tag('user-status')] - | Annotated[ClusterPullEvent, Tag('cluster-pull')] - | Annotated[UnknownEvent, Tag('unknown')], + Annotated[UserStatusEvent, Tag('user-status')] | Annotated[UnknownEvent, Tag('unknown')], Discriminator(_event_discriminator), ] """Discriminated union of every typed WebSocket event plus the catch-all. @@ -85,16 +76,37 @@ def _event_discriminator(value: Any) -> str: """ -_event_adapter: TypeAdapter[UserStatusEvent | ClusterPullEvent | UnknownEvent] = TypeAdapter( - DiveraEvent, -) +_event_adapter: TypeAdapter[UserStatusEvent | UnknownEvent] = TypeAdapter(DiveraEvent) -def parse_event(event: Mapping[str, Any]) -> UserStatusEvent | ClusterPullEvent | UnknownEvent: +def parse_event(event: Mapping[str, Any]) -> UserStatusEvent | UnknownEvent: """Parse a raw WebSocket event into the matching typed model. Dispatches on ``type`` via :data:`DiveraEvent`; unknown or missing types fall back to :class:`UnknownEvent` instead of raising, so the subscribe loop never dies on a newly introduced server-side event name. + + If a frame carries a known ``type`` but its nested payload fails the + dedicated validation (e.g. the server changed the wire format in a way + we don't yet model), this also falls back to :class:`UnknownEvent` and + logs the raw frame at WARNING level -- the subscribe loop keeps + yielding events instead of dying on a single unexpected shape, and + the log line gives you everything needed to update the typed model. + + Frames that are missing a ``type`` field entirely (or whose ``type`` is + not a string) are still rejected, since they are malformed and cannot + be routed to any model, not even :class:`UnknownEvent`. """ - return _event_adapter.validate_python(event) + try: + return _event_adapter.validate_python(event) + except ValidationError: + event_type = event.get('type') if isinstance(event, Mapping) else None + if not isinstance(event_type, str) or not event_type: + raise + logger.warning( + 'failed to validate %r WebSocket event against its typed model; ' + 'falling back to UnknownEvent. raw frame: %r', + event_type, + dict(event), + ) + return UnknownEvent.model_validate(event) diff --git a/src/divera247/websocket/session.py b/src/divera247/websocket/session.py index 1a86120..169de87 100644 --- a/src/divera247/websocket/session.py +++ b/src/divera247/websocket/session.py @@ -148,7 +148,7 @@ async def subscribe_websocket( ucr_id: int | None = None, ws_url: str = 'wss://ws.divera247.com/ws', max_auth_attempts: int = 3, -) -> AsyncIterator[models.ClusterPullEvent | models.UserStatusEvent | models.UnknownEvent]: +) -> AsyncIterator[models.UserStatusEvent | models.UnknownEvent]: """Yield typed Divera 24/7 WebSocket events from a single session. Exits when the underlying socket disconnects (by raising @@ -169,7 +169,7 @@ async def stream_websocket( # noqa: PLR0913 max_backoff: float = 60.0, backoff_factor: float = 2.0, backoff_jitter: float = 0.2, -) -> AsyncIterator[models.ClusterPullEvent | models.UserStatusEvent | models.UnknownEvent]: +) -> AsyncIterator[models.UserStatusEvent | models.UnknownEvent]: """Yield events forever, transparently reconnecting on any disconnect. Reconnect delay follows jittered exponential backoff bounded by diff --git a/tests/websocket/test_models.py b/tests/websocket/test_models.py index e1cea05..577fcd0 100644 --- a/tests/websocket/test_models.py +++ b/tests/websocket/test_models.py @@ -5,8 +5,6 @@ import pytest from divera247.websocket.models import ( - ClusterPullEvent, - ClusterPullRef, UnknownEvent, UserStatusEvent, parse_event, @@ -40,12 +38,6 @@ 'ucr': SAMPLE_UCR, } -_CLUSTER_PULL_SAMPLE: dict = { - 'type': 'cluster-pull', - 'pull': {'type': 'news', 'id': SAMPLE_PULL_ID}, - 'cluster': SAMPLE_CLUSTER, -} - def test_user_status_event_parses_sample() -> None: """UserStatusEvent accepts the documented envelope + pull status payload.""" @@ -62,29 +54,6 @@ def test_user_status_event_rejects_wrong_type_literal() -> None: UserStatusEvent.model_validate(bad) -def test_cluster_pull_ref_parses_minimal_fields() -> None: - """ClusterPullRef exposes ``type`` + ``id`` from the nested ``pull`` block.""" - ref = ClusterPullRef.model_validate({'type': 'news', 'id': SAMPLE_REF_ID}) - assert ref.type == 'news' - assert ref.id == SAMPLE_REF_ID - - -def test_cluster_pull_event_parses_sample() -> None: - """ClusterPullEvent exposes both the cluster id and the typed reference.""" - event = ClusterPullEvent.model_validate(_CLUSTER_PULL_SAMPLE) - assert event.type == 'cluster-pull' - assert event.cluster == SAMPLE_CLUSTER - assert event.pull.type == 'news' - assert event.pull.id == SAMPLE_PULL_ID - - -def test_cluster_pull_event_rejects_wrong_type_literal() -> None: - """Non-matching ``type`` value must not validate as ClusterPullEvent.""" - bad = dict(_CLUSTER_PULL_SAMPLE, type='user-status') - with pytest.raises(ValueError, match='cluster-pull'): - ClusterPullEvent.model_validate(bad) - - def test_unknown_event_keeps_arbitrary_type() -> None: """UnknownEvent stores the original ``type`` string instead of overwriting it.""" event = UnknownEvent.model_validate({'type': 'cluster-vehicle'}) @@ -111,16 +80,17 @@ def test_parse_event_dispatches_user_status() -> None: assert parsed.ucr == SAMPLE_UCR -def test_parse_event_dispatches_cluster_pull() -> None: - """``parse_event`` routes ``cluster-pull`` frames to ClusterPullEvent.""" - parsed = parse_event(_CLUSTER_PULL_SAMPLE) - assert isinstance(parsed, ClusterPullEvent) - assert parsed.cluster == SAMPLE_CLUSTER - - -@pytest.mark.parametrize('event_type', ['cluster-message', 'cluster-vehicle', 'some-brand-new-event']) +@pytest.mark.parametrize( + 'event_type', + ['cluster-pull', 'cluster-message', 'cluster-vehicle', 'some-brand-new-event'], +) def test_parse_event_falls_back_to_unknown_and_preserves_type(event_type: str) -> None: - """Unknown ``type`` values route to UnknownEvent with the original string + extras intact.""" + """Unknown ``type`` values route to UnknownEvent with the original string + extras intact. + + ``cluster-pull`` is included here explicitly: it is a known server event + but has no typed envelope yet (awaiting a real live-API sample), so it + must currently flow through the unknown-event fallback. + """ raw = {'type': event_type, 'foo': 1, 'bar': [1, 2]} parsed = parse_event(raw) assert isinstance(parsed, UnknownEvent) @@ -138,3 +108,23 @@ def test_parse_event_non_string_type_still_raises() -> None: """A numeric ``type`` is not a valid tag and must not be coerced into UnknownEvent.""" with pytest.raises(ValueError, match='type'): parse_event({'type': 123}) + + +def test_parse_event_falls_back_when_known_type_has_malformed_payload( + caplog: pytest.LogCaptureFixture, +) -> None: + """Known ``type`` with an unexpectedly shaped body degrades to UnknownEvent with a warning. + + Guards against the subscribe loop dying on a single malformed frame when + the server starts sending a slightly different shape; the raw frame is + logged so the typed model can be updated. + """ + malformed = {'type': 'user-status', 'ucr': SAMPLE_UCR, 'payload': 'not-a-status-object'} + + with caplog.at_level('WARNING', logger='divera247.websocket.models'): + parsed = parse_event(malformed) + + assert isinstance(parsed, UnknownEvent) + assert parsed.type == 'user-status' + assert parsed.model_extra == {'ucr': SAMPLE_UCR, 'payload': 'not-a-status-object'} + assert any('user-status' in record.message for record in caplog.records) diff --git a/tests/websocket/test_session.py b/tests/websocket/test_session.py index 16a5f63..9a48e3e 100644 --- a/tests/websocket/test_session.py +++ b/tests/websocket/test_session.py @@ -21,7 +21,6 @@ from divera247.client import Divera247Client from divera247.websocket import session as session_module from divera247.websocket.models import ( - ClusterPullEvent, UnknownEvent, UserStatusEvent, ) @@ -37,7 +36,6 @@ EXPECTED_UCR = 527459 -EXPECTED_CLUSTER = 8381 EXPECTED_AUTH_ATTEMPT_BUDGET = 3 EXPECTED_REAUTH_FRAMES = 2 @@ -59,11 +57,6 @@ }, 'ucr': EXPECTED_UCR, } -_CLUSTER_PULL_FRAME = { - 'type': 'cluster-pull', - 'pull': {'type': 'news', 'id': 2029889}, - 'cluster': EXPECTED_CLUSTER, -} class FakeWebSocketSession: @@ -221,17 +214,14 @@ async def test_subscribe_websocket_yields_known_event_types( [ {'type': 'init'}, _USER_STATUS_FRAME, - _CLUSTER_PULL_FRAME, ] ) _install_fake_ws(monkeypatch, fake) - events = await _collect(subscribe_websocket(ws_client), limit=2) + events = await _collect(subscribe_websocket(ws_client), limit=1) assert isinstance(events[0], UserStatusEvent) assert events[0].ucr == EXPECTED_UCR - assert isinstance(events[1], ClusterPullEvent) - assert events[1].cluster == EXPECTED_CLUSTER async def test_subscribe_websocket_unknown_event_falls_back_to_unknown_model( @@ -239,10 +229,11 @@ async def test_subscribe_websocket_unknown_event_falls_back_to_unknown_model( ws_client: Divera247Client, ) -> None: """Unknown ``type`` frames surface as UnknownEvent with the original data intact.""" + raw_extras = {'payload': {'id': 42}, 'cluster': 8381} fake = FakeWebSocketSession( [ {'type': 'init'}, - {'type': 'some-brand-new-event', 'payload': {'id': 42}, 'cluster': EXPECTED_CLUSTER}, + {'type': 'some-brand-new-event', **raw_extras}, ] ) _install_fake_ws(monkeypatch, fake) @@ -251,7 +242,7 @@ async def test_subscribe_websocket_unknown_event_falls_back_to_unknown_model( assert isinstance(events[0], UnknownEvent) assert events[0].type == 'some-brand-new-event' - assert events[0].model_extra == {'payload': {'id': 42}, 'cluster': EXPECTED_CLUSTER} + assert events[0].model_extra == raw_extras async def test_subscribe_websocket_init_frame_is_not_yielded( @@ -259,13 +250,13 @@ async def test_subscribe_websocket_init_frame_is_not_yielded( ws_client: Divera247Client, ) -> None: """``init`` is session-level and must stay internal to the session loop.""" - fake = FakeWebSocketSession([{'type': 'init'}, _CLUSTER_PULL_FRAME]) + fake = FakeWebSocketSession([{'type': 'init'}, _USER_STATUS_FRAME]) _install_fake_ws(monkeypatch, fake) events = await _collect(subscribe_websocket(ws_client), limit=1) assert len(events) == 1 - assert isinstance(events[0], ClusterPullEvent) + assert isinstance(events[0], UserStatusEvent) async def test_subscribe_websocket_ignores_non_json_frames( @@ -277,14 +268,14 @@ async def test_subscribe_websocket_ignores_non_json_frames( [ {'type': 'init'}, 'this is not json', - _CLUSTER_PULL_FRAME, + _USER_STATUS_FRAME, ] ) _install_fake_ws(monkeypatch, fake) events = await _collect(subscribe_websocket(ws_client), limit=1) - assert isinstance(events[0], ClusterPullEvent) + assert isinstance(events[0], UserStatusEvent) async def test_subscribe_websocket_reauthenticates_on_jwt_expired( @@ -297,14 +288,14 @@ async def test_subscribe_websocket_reauthenticates_on_jwt_expired( {'type': 'init'}, {'type': 'jwtExpired'}, {'type': 'init'}, - _CLUSTER_PULL_FRAME, + _USER_STATUS_FRAME, ] ) _install_fake_ws(monkeypatch, fake) events = await _collect(subscribe_websocket(ws_client), limit=1) - assert isinstance(events[0], ClusterPullEvent) + assert isinstance(events[0], UserStatusEvent) auth_frames = [frame for frame in fake.sent if frame.get('type') == 'authenticate'] assert len(auth_frames) == EXPECTED_REAUTH_FRAMES @@ -338,7 +329,7 @@ async def test_subscribe_websocket_init_resets_auth_attempt_budget( {'type': 'jwtExpired'}, {'type': 'jwtExpired'}, {'type': 'init'}, - _CLUSTER_PULL_FRAME, + _USER_STATUS_FRAME, ] fake = FakeWebSocketSession(frames) _install_fake_ws(monkeypatch, fake) @@ -348,7 +339,7 @@ async def test_subscribe_websocket_init_resets_auth_attempt_budget( limit=1, ) - assert isinstance(events[0], ClusterPullEvent) + assert isinstance(events[0], UserStatusEvent) async def test_subscribe_websocket_propagates_disconnect( diff --git a/tests/websocket/test_stream.py b/tests/websocket/test_stream.py index f9165db..0b3e650 100644 --- a/tests/websocket/test_stream.py +++ b/tests/websocket/test_stream.py @@ -17,7 +17,7 @@ from divera247.auth import AccessKeyAuth from divera247.client import Divera247Client from divera247.websocket import session as session_module -from divera247.websocket.models import ClusterPullEvent +from divera247.websocket.models import UserStatusEvent from divera247.websocket.session import ( WebSocketAuthenticationError, stream_websocket, @@ -26,11 +26,24 @@ if TYPE_CHECKING: from collections.abc import AsyncGenerator, Iterable, Sequence -EXPECTED_CLUSTER = 8381 -_CLUSTER_PULL_FRAME: dict[str, Any] = { - 'type': 'cluster-pull', - 'pull': {'type': 'news', 'id': 2029889}, - 'cluster': EXPECTED_CLUSTER, +EXPECTED_UCR = 527459 +_USER_STATUS_FRAME: dict[str, Any] = { + 'type': 'user-status', + 'payload': { + 'status_id': 33035, + 'status_skip_statusplan': False, + 'status_skip_geofence': False, + 'status_set_date': 1776767153, + 'status_reset_date': '', + 'status_reset_id': 0, + 'status_log': [], + 'status_changes': [], + 'note': '', + 'vehicle': 0, + 'ts': 1776767153, + 'cached': False, + }, + 'ucr': EXPECTED_UCR, } @@ -105,15 +118,15 @@ async def test_stream_websocket_reconnects_after_disconnect( opened = _install_stream_ws( monkeypatch, [ - _FakeSession([{'type': 'init'}, _CLUSTER_PULL_FRAME]), - _FakeSession([{'type': 'init'}, _CLUSTER_PULL_FRAME]), + _FakeSession([{'type': 'init'}, _USER_STATUS_FRAME]), + _FakeSession([{'type': 'init'}, _USER_STATUS_FRAME]), ], ) events = await _collect(stream_websocket(ws_client, initial_backoff=0.0), limit=2) assert len(events) == 2 - assert all(isinstance(event, ClusterPullEvent) for event in events) + assert all(isinstance(event, UserStatusEvent) for event in events) assert len(opened) == 2