diff --git a/README.md b/README.md index 3506b08..15b50f9 100644 --- a/README.md +++ b/README.md @@ -154,6 +154,7 @@ Two entry points live in `divera247.websocket`: ```python from divera247.websocket import ( + ClusterPullEvent, UnknownEvent, UserStatusEvent, stream_websocket, @@ -161,8 +162,10 @@ from divera247.websocket import ( async for event in stream_websocket(client, ucr_id=527_459): match event: - case UserStatusEvent(payload=payload, ucr=ucr): - ... + case UserStatusEvent(ucr=ucr, status=status): + ... # status is a PullStatusData + case ClusterPullEvent(cluster=cluster, pull=pull): + ... # re-fetch pull.type (e.g. "alarm") + pull.id case UnknownEvent(): ... # forward-compatible fallback ``` diff --git a/src/divera247/websocket/__init__.py b/src/divera247/websocket/__init__.py index 9f705ce..e5c1cba 100644 --- a/src/divera247/websocket/__init__.py +++ b/src/divera247/websocket/__init__.py @@ -12,17 +12,19 @@ * :class:`WebSocketAuthenticationError` - raised when authentication keeps failing so the caller can react instead of silently looping. * Pydantic event envelopes (:class:`UserStatusEvent`, + :class:`ClusterPullEvent` with its :class:`ClusterPullRef`, :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. + Other server-side event types (e.g. ``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, @@ -30,13 +32,17 @@ async for event in subscribe_websocket(client, ucr_id=ucr_id): match event: - case UserStatusEvent(payload=payload, ucr=ucr): + case UserStatusEvent(ucr=ucr, status=status): ... + case ClusterPullEvent(cluster=cluster, pull=pull): + ... # re-fetch pull.type / pull.id for this cluster case UnknownEvent(type=msg_type): ... """ from divera247.websocket.models import ( + ClusterPullEvent, + ClusterPullRef, DiveraEvent, UnknownEvent, UserStatusEvent, @@ -49,6 +55,8 @@ ) __all__ = [ + 'ClusterPullEvent', + 'ClusterPullRef', 'DiveraEvent', 'UnknownEvent', 'UserStatusEvent', diff --git a/src/divera247/websocket/models.py b/src/divera247/websocket/models.py index 254c05a..a3b1c26 100644 --- a/src/divera247/websocket/models.py +++ b/src/divera247/websocket/models.py @@ -5,6 +5,7 @@ from typing import Annotated, Any, Literal from pydantic import ( + AliasPath, BaseModel, ConfigDict, Discriminator, @@ -19,18 +20,92 @@ logger = logging.getLogger(__name__) +class ClusterPullRef(BaseModel): + """Reference to the cluster resource that triggered a ``cluster-pull`` event. + + The ``type`` names the resource family (observed: ``alarm``; likely + also ``message``, ``event``, ``news``, ``vehicle``, etc., i.e. the + same families exposed by ``GET /api/v2/pull/all``) and ``id`` is the + primary key within that family. Callers should re-fetch the matching + pull endpoint to get the full object. + """ + + model_config = ConfigDict(extra='allow') + + type: str = Field(description='Ressource-Typ, der neu gepullt werden soll (z. B. ``alarm``)') + id: int = Field(description='ID der betroffenen Ressource innerhalb ihres Typs') + + +class ClusterPullEvent(BaseModel): + """``cluster-pull`` WebSocket event: a cluster-scoped resource changed. + + The server tells us *that* something in a given cluster changed and + *which* resource to refresh (type + id). The actual object is not + inlined; the client is expected to re-fetch the relevant pull + endpoint to materialise the update. + + Wire format, mirroring :class:`UserStatusEvent`: + + .. code-block:: json + + { + "type": "cluster-pull", + "payload": { + "type": "cluster-pull", + "pull": {"type": "alarm", "id": 123456}, + "cluster": 1234 + } + } + + Flattened via :class:`~pydantic.AliasPath` so callers access + ``event.pull`` / ``event.cluster`` directly. + """ + + type: Literal['cluster-pull'] = Field(description='Event-Typ') + pull: ClusterPullRef = Field( + validation_alias=AliasPath('payload', 'pull'), + description='Referenz auf die geänderte Ressource (aus payload.pull)', + ) + cluster: int = Field( + validation_alias=AliasPath('payload', 'cluster'), + description='ID des betroffenen Clusters (aus payload.cluster)', + ) + + class UserStatusEvent(BaseModel): """``user-status`` WebSocket event: own status changed for a given UCR. - The ``payload`` has the exact same shape as the ``status`` block of - ``GET /api/v2/pull/all`` (see :class:`PullStatusData`); the envelope - adds the event ``type`` discriminator and the ``ucr`` of the affected - UserClusterRelation. + The server wire-format wraps the actual status (same fields as the + ``status`` block of ``GET /api/v2/pull/all`` -- see + :class:`PullStatusData`) and the affected ``ucr`` inside a nested + ``payload`` object, with the event type repeated redundantly at both + levels: + + .. code-block:: json + + { + "type": "user-status", + "payload": { + "type": "user-status", + "status": { ...PullStatusData... }, + "ucr": 527459 + } + } + + We flatten this on validation via :class:`~pydantic.AliasPath`, so + callers only deal with ``event.status`` / ``event.ucr`` and never have + to reach through a redundant payload wrapper. """ type: Literal['user-status'] = Field(description='Event-Typ') - payload: PullStatusData = Field(description='Aktueller Status der UCR') - ucr: int = Field(description='ID der betroffenen UserClusterRelation') + status: PullStatusData = Field( + validation_alias=AliasPath('payload', 'status'), + description='Aktueller Status der UCR (aus payload.status)', + ) + ucr: int = Field( + validation_alias=AliasPath('payload', 'ucr'), + description='ID der betroffenen UserClusterRelation (aus payload.ucr)', + ) class UnknownEvent(BaseModel): @@ -39,9 +114,9 @@ class UnknownEvent(BaseModel): 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). 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`). + unknown event types (e.g. ``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') @@ -49,7 +124,7 @@ class UnknownEvent(BaseModel): type: str = Field(description='Raw event type as sent by the server') -_KNOWN_EVENT_TYPES: frozenset[str] = frozenset({'user-status'}) +_KNOWN_EVENT_TYPES: frozenset[str] = frozenset({'user-status', 'cluster-pull'}) def _event_discriminator(value: Any) -> str: @@ -66,7 +141,9 @@ def _event_discriminator(value: Any) -> str: DiveraEvent = Annotated[ - Annotated[UserStatusEvent, Tag('user-status')] | Annotated[UnknownEvent, Tag('unknown')], + Annotated[UserStatusEvent, Tag('user-status')] + | Annotated[ClusterPullEvent, Tag('cluster-pull')] + | Annotated[UnknownEvent, Tag('unknown')], Discriminator(_event_discriminator), ] """Discriminated union of every typed WebSocket event plus the catch-all. @@ -76,10 +153,10 @@ def _event_discriminator(value: Any) -> str: """ -_event_adapter: TypeAdapter[UserStatusEvent | UnknownEvent] = TypeAdapter(DiveraEvent) +_event_adapter: TypeAdapter[UserStatusEvent | ClusterPullEvent | UnknownEvent] = TypeAdapter(DiveraEvent) -def parse_event(event: Mapping[str, Any]) -> UserStatusEvent | UnknownEvent: +def parse_event(event: Mapping[str, Any]) -> UserStatusEvent | ClusterPullEvent | UnknownEvent: """Parse a raw WebSocket event into the matching typed model. Dispatches on ``type`` via :data:`DiveraEvent`; unknown or missing types diff --git a/src/divera247/websocket/session.py b/src/divera247/websocket/session.py index 169de87..ffbffcb 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.UserStatusEvent | models.UnknownEvent]: +) -> AsyncIterator[models.UserStatusEvent | models.ClusterPullEvent | 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.UserStatusEvent | models.UnknownEvent]: +) -> AsyncIterator[models.UserStatusEvent | models.ClusterPullEvent | 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 577fcd0..ae20ce5 100644 --- a/tests/websocket/test_models.py +++ b/tests/websocket/test_models.py @@ -5,6 +5,8 @@ import pytest from divera247.websocket.models import ( + ClusterPullEvent, + ClusterPullRef, UnknownEvent, UserStatusEvent, parse_event, @@ -13,47 +15,105 @@ SAMPLE_UCR = 527459 SAMPLE_STATUS_ID = 33035 SAMPLE_CLUSTER = 8381 -SAMPLE_PULL_ID = 2029889 +SAMPLE_PULL_ID = 33688274 +SAMPLE_PULL_TYPE = 'alarm' SAMPLE_REF_ID = 42 +_STATUS_BLOCK: dict = { + 'status_id': SAMPLE_STATUS_ID, + 'status_skip_statusplan': False, + 'status_skip_geofence': False, + 'status_set_date': 1776767153, + 'status_reset_date': '', + 'status_reset_id': 0, + 'status_log': [], + 'status_changes': [ + {'ts': 1776767114, 'status': 33035, 'note': '', 'vehicle': 0, 'event': 0, 'type': 0}, + {'ts': 1776767152, 'status': 33036, 'note': '', 'vehicle': 0, 'event': 0, 'type': 0}, + ], + 'note': '', + 'vehicle': 0, + 'ts': 1776767153, + 'cached': False, +} + _USER_STATUS_SAMPLE: dict = { 'type': 'user-status', 'payload': { - 'status_id': SAMPLE_STATUS_ID, - 'status_skip_statusplan': False, - 'status_skip_geofence': False, - 'status_set_date': 1776767153, - 'status_reset_date': '', - 'status_reset_id': 0, - 'status_log': [], - 'status_changes': [ - {'ts': 1776767114, 'status': 33035, 'note': '', 'vehicle': 0, 'event': 0, 'type': 0}, - {'ts': 1776767152, 'status': 33036, 'note': '', 'vehicle': 0, 'event': 0, 'type': 0}, - ], - 'note': '', - 'vehicle': 0, - 'ts': 1776767153, - 'cached': False, + 'type': 'user-status', + 'status': _STATUS_BLOCK, + 'ucr': SAMPLE_UCR, + }, +} + +_CLUSTER_PULL_SAMPLE: dict = { + 'type': 'cluster-pull', + 'payload': { + 'type': 'cluster-pull', + 'pull': {'type': SAMPLE_PULL_TYPE, 'id': SAMPLE_PULL_ID}, + 'cluster': SAMPLE_CLUSTER, }, - 'ucr': SAMPLE_UCR, } -def test_user_status_event_parses_sample() -> None: - """UserStatusEvent accepts the documented envelope + pull status payload.""" +def test_user_status_event_flattens_nested_payload() -> None: + """UserStatusEvent hoists ``status`` and ``ucr`` out of the nested payload via AliasPath.""" event = UserStatusEvent.model_validate(_USER_STATUS_SAMPLE) assert event.type == 'user-status' assert event.ucr == SAMPLE_UCR - assert event.payload.status_id == SAMPLE_STATUS_ID + assert event.status.status_id == SAMPLE_STATUS_ID -def test_user_status_event_rejects_wrong_type_literal() -> None: - """Non-matching ``type`` value must not validate as UserStatusEvent.""" +def test_user_status_event_rejects_wrong_outer_type_literal() -> None: + """Non-matching outer ``type`` value must not validate as UserStatusEvent.""" bad = dict(_USER_STATUS_SAMPLE, type='something-else') with pytest.raises(ValueError, match='user-status'): UserStatusEvent.model_validate(bad) +def test_user_status_event_requires_nested_payload_keys() -> None: + """Missing ``payload.status`` or ``payload.ucr`` breaks validation. + + Guards the :class:`~pydantic.AliasPath` flattening: if the server ever + drops the nested structure, we want a hard failure (and thus a + fallback to :class:`UnknownEvent` via :func:`parse_event`) instead of + silently producing an event with dummy fields. + """ + bad = dict(_USER_STATUS_SAMPLE, payload={'type': 'user-status'}) + with pytest.raises(ValueError, match=r'status|ucr'): + UserStatusEvent.model_validate(bad) + + +def test_cluster_pull_event_flattens_nested_payload() -> None: + """ClusterPullEvent hoists ``pull`` and ``cluster`` out of the nested payload via AliasPath.""" + event = ClusterPullEvent.model_validate(_CLUSTER_PULL_SAMPLE) + assert event.type == 'cluster-pull' + assert event.cluster == SAMPLE_CLUSTER + assert isinstance(event.pull, ClusterPullRef) + assert event.pull.type == SAMPLE_PULL_TYPE + assert event.pull.id == SAMPLE_PULL_ID + + +def test_cluster_pull_event_rejects_wrong_outer_type_literal() -> None: + """Non-matching outer ``type`` value must not validate as ClusterPullEvent.""" + bad = dict(_CLUSTER_PULL_SAMPLE, type='something-else') + with pytest.raises(ValueError, match='cluster-pull'): + ClusterPullEvent.model_validate(bad) + + +def test_cluster_pull_ref_preserves_unknown_fields() -> None: + """Extra fields on the pull reference round-trip via ``model_extra``. + + The server is free to enrich the ``pull`` reference (e.g. with a + ``title`` or ``updated_at``) without breaking this model. + """ + raw = {'type': SAMPLE_PULL_TYPE, 'id': SAMPLE_PULL_ID, 'title': 'Einsatz', 'prio': 1} + ref = ClusterPullRef.model_validate(raw) + assert ref.type == SAMPLE_PULL_TYPE + assert ref.id == SAMPLE_PULL_ID + assert ref.model_extra == {'title': 'Einsatz', 'prio': 1} + + 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'}) @@ -78,19 +138,24 @@ def test_parse_event_dispatches_user_status() -> None: parsed = parse_event(_USER_STATUS_SAMPLE) assert isinstance(parsed, UserStatusEvent) assert parsed.ucr == SAMPLE_UCR + assert parsed.status.status_id == SAMPLE_STATUS_ID + + +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 + assert parsed.pull.type == SAMPLE_PULL_TYPE + assert parsed.pull.id == SAMPLE_PULL_ID @pytest.mark.parametrize( 'event_type', - ['cluster-pull', 'cluster-message', 'cluster-vehicle', 'some-brand-new-event'], + ['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. - - ``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. - """ + """Unknown ``type`` values route to UnknownEvent with the original string + extras intact.""" raw = {'type': event_type, 'foo': 1, 'bar': [1, 2]} parsed = parse_event(raw) assert isinstance(parsed, UnknownEvent) @@ -119,12 +184,12 @@ def test_parse_event_falls_back_when_known_type_has_malformed_payload( 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'} + malformed = {'type': 'user-status', 'payload': 'not-a-payload-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 parsed.model_extra == {'payload': 'not-a-payload-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 9a48e3e..ec51df3 100644 --- a/tests/websocket/test_session.py +++ b/tests/websocket/test_session.py @@ -42,20 +42,23 @@ _USER_STATUS_FRAME = { '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, + 'type': 'user-status', + 'status': { + '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, }, - 'ucr': EXPECTED_UCR, } diff --git a/tests/websocket/test_stream.py b/tests/websocket/test_stream.py index 0b3e650..3b7a813 100644 --- a/tests/websocket/test_stream.py +++ b/tests/websocket/test_stream.py @@ -30,20 +30,23 @@ _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, + 'type': 'user-status', + 'status': { + '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, }, - 'ucr': EXPECTED_UCR, }