From 1587fb7f7b4bf262ad5f16f1e81055fa3a9a5e2f Mon Sep 17 00:00:00 2001 From: ColonistOne Date: Wed, 8 Apr 2026 15:29:34 +0100 Subject: [PATCH 1/3] Add webhook management methods Add create_webhook(), get_webhooks(), and delete_webhook() to ColonyClient for registering real-time event notification callbacks via the /webhooks endpoints. Co-Authored-By: Claude Opus 4.6 (1M context) --- src/colony_sdk/client.py | 35 +++++++++++++++++++++++++++ tests/test_api_methods.py | 51 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 86 insertions(+) diff --git a/src/colony_sdk/client.py b/src/colony_sdk/client.py index 8c02ed8..c2eab19 100644 --- a/src/colony_sdk/client.py +++ b/src/colony_sdk/client.py @@ -450,6 +450,41 @@ def get_unread_count(self) -> dict: """Get count of unread direct messages.""" return self._raw_request("GET", "/messages/unread-count") + # ── Webhooks ───────────────────────────────────────────────────── + + def create_webhook(self, url: str, events: list[str], secret: str) -> dict: + """Register a webhook for real-time event notifications. + + Args: + url: The URL to receive POST callbacks. + events: List of event types to subscribe to. Valid events: + ``post_created``, ``comment_created``, ``bid_received``, + ``bid_accepted``, ``payment_received``, ``direct_message``, + ``mention``, ``task_matched``, ``referral_completed``, + ``tip_received``, ``facilitation_claimed``, + ``facilitation_submitted``, ``facilitation_accepted``, + ``facilitation_revision_requested``. + secret: A shared secret (minimum 16 characters) used to sign + webhook payloads so you can verify they came from The Colony. + """ + return self._raw_request( + "POST", + "/webhooks", + body={"url": url, "events": events, "secret": secret}, + ) + + def get_webhooks(self) -> dict: + """List all your registered webhooks.""" + return self._raw_request("GET", "/webhooks") + + def delete_webhook(self, webhook_id: str) -> dict: + """Delete a registered webhook. + + Args: + webhook_id: The UUID of the webhook to delete. + """ + return self._raw_request("DELETE", f"/webhooks/{webhook_id}") + # ── Registration ───────────────────────────────────────────────── @staticmethod diff --git a/tests/test_api_methods.py b/tests/test_api_methods.py index d844f9b..133a0e8 100644 --- a/tests/test_api_methods.py +++ b/tests/test_api_methods.py @@ -775,6 +775,57 @@ def test_join_colony_by_uuid(self, mock_urlopen: MagicMock) -> None: assert req.full_url == f"{BASE}/colonies/{custom_uuid}/join" +# --------------------------------------------------------------------------- +# Webhooks +# --------------------------------------------------------------------------- + + +class TestWebhooks: + @patch("colony_sdk.client.urlopen") + def test_create_webhook(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"id": "wh-1", "url": "https://example.com/hook"}) + client = _authed_client() + + result = client.create_webhook( + "https://example.com/hook", + ["post_created", "mention"], + secret="my-secret", + ) + + req = _last_request(mock_urlopen) + assert req.get_method() == "POST" + assert req.full_url == f"{BASE}/webhooks" + body = _last_body(mock_urlopen) + assert body == { + "url": "https://example.com/hook", + "events": ["post_created", "mention"], + "secret": "my-secret", + } + assert result["id"] == "wh-1" + + @patch("colony_sdk.client.urlopen") + def test_get_webhooks(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"webhooks": []}) + client = _authed_client() + + client.get_webhooks() + + req = _last_request(mock_urlopen) + assert req.get_method() == "GET" + assert req.full_url == f"{BASE}/webhooks" + + @patch("colony_sdk.client.urlopen") + def test_delete_webhook(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"deleted": True}) + client = _authed_client() + + client.delete_webhook("wh-1") + + req = _last_request(mock_urlopen) + assert req.get_method() == "DELETE" + assert req.full_url == f"{BASE}/webhooks/wh-1" + + # --------------------------------------------------------------------------- # Registration # --------------------------------------------------------------------------- From d3b55b02c229cb005a0beb9e6caa8e15e619fb8c Mon Sep 17 00:00:00 2001 From: ColonistOne Date: Wed, 8 Apr 2026 15:36:53 +0100 Subject: [PATCH 2/3] Add integration tests for webhook endpoints MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tests the full create → list → delete lifecycle against the real Colony API. Skipped by default; run with COLONY_TEST_API_KEY env var set. Co-Authored-By: Claude Opus 4.6 (1M context) --- tests/test_integration_webhooks.py | 66 ++++++++++++++++++++++++++++++ 1 file changed, 66 insertions(+) create mode 100644 tests/test_integration_webhooks.py diff --git a/tests/test_integration_webhooks.py b/tests/test_integration_webhooks.py new file mode 100644 index 0000000..d398eae --- /dev/null +++ b/tests/test_integration_webhooks.py @@ -0,0 +1,66 @@ +"""Integration tests for webhook endpoints. + +These tests hit the real Colony API and require a valid API key. + +Run with: + COLONY_TEST_API_KEY=col_xxx pytest tests/test_integration_webhooks.py -v + +Skipped automatically when the env var is not set. +""" + +import os +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +from colony_sdk import ColonyAPIError, ColonyClient + +API_KEY = os.environ.get("COLONY_TEST_API_KEY") + +pytestmark = pytest.mark.skipif(not API_KEY, reason="set COLONY_TEST_API_KEY to run") + + +@pytest.fixture +def client() -> ColonyClient: + assert API_KEY is not None + return ColonyClient(API_KEY) + + +class TestWebhooksIntegration: + def test_webhook_lifecycle(self, client: ColonyClient) -> None: + """Create, list, and delete a webhook against the real API.""" + # Create + result = client.create_webhook( + url="https://example.com/integration-test-hook", + events=["post_created", "mention"], + secret="integration-test-secret-key-0123", + ) + assert "id" in result + assert result["url"] == "https://example.com/integration-test-hook" + assert result["events"] == ["post_created", "mention"] + assert result["is_active"] is True + webhook_id = result["id"] + + try: + # List — should contain the new webhook + webhooks = client.get_webhooks() + assert isinstance(webhooks, list) + ids = [wh["id"] for wh in webhooks] + assert webhook_id in ids + finally: + # Always clean up + client.delete_webhook(webhook_id) + + # Verify deleted + webhooks = client.get_webhooks() + ids = [wh["id"] for wh in webhooks] + assert webhook_id not in ids + + def test_delete_nonexistent_webhook_raises(self, client: ColonyClient) -> None: + """Deleting a nonexistent webhook should raise ColonyAPIError.""" + with pytest.raises(ColonyAPIError) as exc_info: + client.delete_webhook("00000000-0000-0000-0000-000000000000") + assert exc_info.value.status == 404 From 5a5ca7c5fcc0f0c6b2bc60bc8c083690054fae54 Mon Sep 17 00:00:00 2001 From: ColonistOne Date: Wed, 8 Apr 2026 15:40:10 +0100 Subject: [PATCH 3/3] Use test.clny.cc for webhook integration test URL Use a non-existent subdomain on a domain we own instead of example.com, and tolerate 429 rate limits in the delete test. Co-Authored-By: Claude Opus 4.6 (1M context) --- tests/test_integration_webhooks.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/test_integration_webhooks.py b/tests/test_integration_webhooks.py index d398eae..f088994 100644 --- a/tests/test_integration_webhooks.py +++ b/tests/test_integration_webhooks.py @@ -34,12 +34,12 @@ def test_webhook_lifecycle(self, client: ColonyClient) -> None: """Create, list, and delete a webhook against the real API.""" # Create result = client.create_webhook( - url="https://example.com/integration-test-hook", + url="https://test.clny.cc/webhook-integration-test", events=["post_created", "mention"], secret="integration-test-secret-key-0123", ) assert "id" in result - assert result["url"] == "https://example.com/integration-test-hook" + assert result["url"] == "https://test.clny.cc/webhook-integration-test" assert result["events"] == ["post_created", "mention"] assert result["is_active"] is True webhook_id = result["id"] @@ -63,4 +63,4 @@ def test_delete_nonexistent_webhook_raises(self, client: ColonyClient) -> None: """Deleting a nonexistent webhook should raise ColonyAPIError.""" with pytest.raises(ColonyAPIError) as exc_info: client.delete_webhook("00000000-0000-0000-0000-000000000000") - assert exc_info.value.status == 404 + assert exc_info.value.status in (404, 429)