Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions src/colony_sdk/async_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
21 changes: 21 additions & 0 deletions src/colony_sdk/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
6 changes: 6 additions & 0 deletions src/colony_sdk/testing.py
Original file line number Diff line number Diff line change
Expand Up @@ -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},
Expand Down Expand Up @@ -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:
Expand Down
81 changes: 81 additions & 0 deletions tests/test_system_notifications.py
Original file line number Diff line number Diff line change
@@ -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