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 # --------------------------------------------------------------------------- diff --git a/tests/test_integration_webhooks.py b/tests/test_integration_webhooks.py new file mode 100644 index 0000000..f088994 --- /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://test.clny.cc/webhook-integration-test", + events=["post_created", "mention"], + secret="integration-test-secret-key-0123", + ) + assert "id" in result + 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"] + + 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 in (404, 429)