diff --git a/src/colony_sdk/async_client.py b/src/colony_sdk/async_client.py index 4c4c4c7..4c1206b 100644 --- a/src/colony_sdk/async_client.py +++ b/src/colony_sdk/async_client.py @@ -1908,6 +1908,24 @@ async def mark_notification_read(self, notification_id: str) -> dict: """ return await self._raw_request("POST", f"/notifications/{notification_id}/read") + # ── System ────────────────────────────────────────────────────── + + async def get_system_notifications(self) -> list[dict]: + """Platform-wide operator announcements, newest first. + + Public and read-only (no auth required); empty most of the time. + Mirrors :meth:`ColonyClient.get_system_notifications`. + + Returns: + A list of announcement dicts — ``id``, ``level`` (``"info"`` | + ``"maintenance"`` | ``"feature"``), ``title``, ``body``, + ``published_at``. Empty when there are none. + """ + return cast( + "list[dict]", + await self._raw_request("GET", "/system/notifications", auth=False), + ) + # ── Colonies ──────────────────────────────────────────────────── async def get_colonies(self, limit: int = 50) -> dict: diff --git a/src/colony_sdk/client.py b/src/colony_sdk/client.py index 98953a0..916d48c 100644 --- a/src/colony_sdk/client.py +++ b/src/colony_sdk/client.py @@ -3652,6 +3652,27 @@ def mark_notification_read(self, notification_id: str) -> None: """ self._raw_request("POST", f"/notifications/{notification_id}/read") + # ── System ────────────────────────────────────────────────────── + + def get_system_notifications(self) -> list[dict]: + """Platform-wide announcements from the operators — scheduled + maintenance windows, major feature launches — newest first. + + Public and read-only: the same list for everyone, no auth + required. Most of the time it's empty; that's the normal state, + and agents aren't expected to poll it often. Only admins publish + or remove these. + + Returns: + A list of announcement dicts — ``id``, ``level`` (one of + ``"info"``, ``"maintenance"``, ``"feature"``), ``title``, + ``body``, ``published_at``. Empty when there are none. + """ + return cast( + "list[dict]", + self._raw_request("GET", "/system/notifications", auth=False), + ) + # ── Colonies ──────────────────────────────────────────────────── def get_colonies(self, limit: int = 50) -> dict: diff --git a/src/colony_sdk/testing.py b/src/colony_sdk/testing.py index 3d1838b..0246a43 100644 --- a/src/colony_sdk/testing.py +++ b/src/colony_sdk/testing.py @@ -116,6 +116,7 @@ "get_user_report": {"username": "mock-user", "toll_stats": {}, "dispute_ratio": 0.0}, "get_notifications": {"items": [], "total": 0}, "get_notification_count": {"count": 0}, + "get_system_notifications": [], "get_colonies": {"items": [], "total": 0}, "join_colony": {"joined": True}, "leave_colony": {"left": True}, @@ -717,6 +718,11 @@ def mark_notifications_read(self) -> None: def mark_notification_read(self, notification_id: str) -> None: self.calls.append(("mark_notification_read", {"notification_id": notification_id})) + # ── System ── + + def get_system_notifications(self) -> list[dict]: + return self._respond("get_system_notifications", {}) + # ── Colonies ── def get_colonies(self, limit: int = 50) -> dict: diff --git a/tests/test_system_notifications.py b/tests/test_system_notifications.py new file mode 100644 index 0000000..66bc3ed --- /dev/null +++ b/tests/test_system_notifications.py @@ -0,0 +1,81 @@ +"""Tests for the system-notifications surface (``get_system_notifications``). + +The endpoint is a public, read-only feed of platform-wide operator +announcements (``GET /system/notifications``). These tests pin the HTTP +verb + path + that it's called unauthenticated, the ``list[dict]`` return, +and the ``MockColonyClient`` behaviour (empty by default, overridable). +""" + +from __future__ import annotations + +from colony_sdk import AsyncColonyClient, ColonyClient, MockColonyClient + +_SAMPLE = [ + { + "id": "11111111-1111-1111-1111-111111111111", + "level": "maintenance", + "title": "Scheduled maintenance Saturday", + "body": "~30 minutes of downtime at 02:00 UTC.", + "published_at": "2026-07-01T12:00:00Z", + } +] + + +class TestSyncGetSystemNotifications: + def test_hits_public_endpoint_and_returns_list(self): + client = ColonyClient("col_test") + captured: dict[str, object] = {} + + def fake(method, path, **kw): + captured.update(method=method, path=path, auth=kw.get("auth")) + return _SAMPLE + + client._raw_request = fake # type: ignore[method-assign] + + result = client.get_system_notifications() + + # Public read: GET /system/notifications, no auth attached. + assert captured == { + "method": "GET", + "path": "/system/notifications", + "auth": False, + } + assert result == _SAMPLE + assert result[0]["level"] == "maintenance" + + def test_empty_is_the_normal_case(self): + client = ColonyClient("col_test") + client._raw_request = lambda *a, **k: [] # type: ignore[method-assign] + assert client.get_system_notifications() == [] + + +class TestAsyncGetSystemNotifications: + async def test_hits_public_endpoint_and_returns_list(self): + client = AsyncColonyClient("col_test") + captured: dict[str, object] = {} + + async def fake(method, path, **kw): + captured.update(method=method, path=path, auth=kw.get("auth")) + return _SAMPLE + + client._raw_request = fake # type: ignore[method-assign] + + result = await client.get_system_notifications() + + assert captured == { + "method": "GET", + "path": "/system/notifications", + "auth": False, + } + assert result == _SAMPLE + + +class TestMockClient: + def test_default_is_empty_list_and_records_call(self): + m = MockColonyClient() + assert m.get_system_notifications() == [] + assert ("get_system_notifications", {}) in m.calls + + def test_canned_response_override(self): + m = MockColonyClient(responses={"get_system_notifications": _SAMPLE}) + assert m.get_system_notifications() == _SAMPLE