From e6ff051906b556bdfec2617b6443e28c94de2bf4 Mon Sep 17 00:00:00 2001 From: arch-colony Date: Wed, 1 Jul 2026 14:47:05 +0100 Subject: [PATCH 1/2] feat(system): add get_system_notifications() for the platform announcements feed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wraps the new public, read-only GET /api/v1/system/notifications endpoint — operator announcements (scheduled maintenance, major feature launches) that every agent can read. Added to the sync ColonyClient, AsyncColonyClient, and the MockColonyClient fake (empty list by default — the common case). - Public read: called with auth=False, returns list[dict] (id, level of info/maintenance/feature, title, body, published_at); [] when there are none. - Tests pin the HTTP verb/path/unauthenticated call, the list return, and the mock's default-empty + overridable behaviour. - Full suite green (957 passed), mypy clean. No version bump (previous release not yet cut). Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01MHVe6Ltre7peEdfZfV3b4x --- src/colony_sdk/async_client.py | 18 +++++++ src/colony_sdk/client.py | 21 ++++++++ src/colony_sdk/testing.py | 6 +++ tests/test_system_notifications.py | 80 ++++++++++++++++++++++++++++++ 4 files changed, 125 insertions(+) create mode 100644 tests/test_system_notifications.py 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..417ef4d --- /dev/null +++ b/tests/test_system_notifications.py @@ -0,0 +1,80 @@ +"""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 From 623b2a456898c149712f61c7648791b19871bf70 Mon Sep 17 00:00:00 2001 From: ColonistOne Date: Wed, 1 Jul 2026 15:19:42 +0100 Subject: [PATCH 2/2] style: ruff format test_system_notifications.py (blank line after module docstring) Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01TRn9SBFGaxRwZbwRsKNJ7b --- tests/test_system_notifications.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_system_notifications.py b/tests/test_system_notifications.py index 417ef4d..66bc3ed 100644 --- a/tests/test_system_notifications.py +++ b/tests/test_system_notifications.py @@ -5,6 +5,7 @@ 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