diff --git a/tests/test_api_methods.py b/tests/test_api_methods.py new file mode 100644 index 0000000..f1034f6 --- /dev/null +++ b/tests/test_api_methods.py @@ -0,0 +1,697 @@ +"""Unit tests for ColonyClient API methods. + +Mocks urllib to verify each method sends the correct HTTP method, URL, +headers, and JSON payload without making real network requests. +""" + +import io +import json +import sys +import time +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +from colony_sdk import ColonyAPIError, ColonyClient +from colony_sdk.colonies import COLONIES + +BASE = "https://thecolony.cc/api/v1" + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _mock_response(data: dict | str = "", status: int = 200) -> MagicMock: + """Build a mock urllib response that behaves like a context manager.""" + body = json.dumps(data).encode() if isinstance(data, dict) else data.encode() + resp = MagicMock() + resp.read.return_value = body + resp.status = status + resp.__enter__ = lambda s: s + resp.__exit__ = MagicMock(return_value=False) + return resp + + +def _make_http_error(code: int, data: dict | None = None, headers: dict | None = None) -> Exception: + """Build a urllib HTTPError with a JSON body.""" + from urllib.error import HTTPError + + body = json.dumps(data or {}).encode() + err = HTTPError( + url="http://test", + code=code, + msg="error", + hdrs=MagicMock(), + fp=io.BytesIO(body), + ) + if headers: + err.headers.get = lambda key, default=None, _h=headers: _h.get(key, default) + return err + + +def _authed_client() -> ColonyClient: + """Return a client with a pre-set token so _ensure_token is a no-op.""" + client = ColonyClient("col_test") + client._token = "fake-jwt" + client._token_expiry = time.time() + 9999 + return client + + +def _last_request(mock_urlopen: MagicMock) -> MagicMock: + """Extract the Request object from the most recent urlopen call.""" + return mock_urlopen.call_args[0][0] + + +def _last_body(mock_urlopen: MagicMock) -> dict: + """Parse the JSON body from the most recent urlopen call.""" + req = _last_request(mock_urlopen) + return json.loads(req.data.decode()) + + +# --------------------------------------------------------------------------- +# Auth / token +# --------------------------------------------------------------------------- + + +class TestAuth: + @patch("colony_sdk.client.urlopen") + def test_ensure_token_fetches_on_first_request(self, mock_urlopen: MagicMock) -> None: + token_resp = _mock_response({"access_token": "jwt-123"}) + data_resp = _mock_response({"id": "user-1"}) + mock_urlopen.side_effect = [token_resp, data_resp] + + client = ColonyClient("col_mykey") + client.get_me() + + # First call is POST /auth/token + auth_req = mock_urlopen.call_args_list[0][0][0] + assert auth_req.get_method() == "POST" + assert auth_req.full_url == f"{BASE}/auth/token" + auth_body = json.loads(auth_req.data.decode()) + assert auth_body == {"api_key": "col_mykey"} + + assert client._token == "jwt-123" + + @patch("colony_sdk.client.urlopen") + def test_cached_token_skips_auth(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"ok": True}) + client = _authed_client() + + client.get_me() + + # Only one call (the actual request), no auth call + assert mock_urlopen.call_count == 1 + req = _last_request(mock_urlopen) + assert "/users/me" in req.full_url + + @patch("colony_sdk.client.urlopen") + def test_bearer_token_in_header(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"ok": True}) + client = _authed_client() + + client.get_me() + + req = _last_request(mock_urlopen) + assert req.get_header("Authorization") == "Bearer fake-jwt" + + @patch("colony_sdk.client.urlopen") + def test_no_auth_header_when_auth_false(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"access_token": "t"}) + client = ColonyClient("col_test") + + client._raw_request("POST", "/auth/token", body={"api_key": "k"}, auth=False) + + req = _last_request(mock_urlopen) + assert req.get_header("Authorization") is None + + +# --------------------------------------------------------------------------- +# Retry logic +# --------------------------------------------------------------------------- + + +class TestRetry: + @patch("colony_sdk.client.urlopen") + def test_401_retries_with_fresh_token(self, mock_urlopen: MagicMock) -> None: + """On 401, client should clear token, re-auth, and retry once.""" + err_401 = _make_http_error(401, {"detail": "expired"}) + token_resp = _mock_response({"access_token": "new-jwt"}) + data_resp = _mock_response({"id": "user-1"}) + mock_urlopen.side_effect = [err_401, token_resp, data_resp] + + client = _authed_client() + result = client.get_me() + + assert result == {"id": "user-1"} + assert client._token == "new-jwt" + + @patch("colony_sdk.client.urlopen") + def test_401_no_retry_when_auth_false(self, mock_urlopen: MagicMock) -> None: + """401 on an auth=False request should not retry.""" + mock_urlopen.side_effect = _make_http_error(401, {"detail": "bad key"}) + + client = ColonyClient("col_test") + with pytest.raises(ColonyAPIError) as exc_info: + client._raw_request("POST", "/auth/token", body={}, auth=False) + assert exc_info.value.status == 401 + + @patch("colony_sdk.client.time.sleep") + @patch("colony_sdk.client.urlopen") + def test_429_retries_with_backoff(self, mock_urlopen: MagicMock, mock_sleep: MagicMock) -> None: + err_429 = _make_http_error(429, {"detail": "rate limited"}) + success = _mock_response({"ok": True}) + mock_urlopen.side_effect = [err_429, success] + + client = _authed_client() + result = client._raw_request("GET", "/test", auth=False) + + assert result == {"ok": True} + mock_sleep.assert_called_once_with(1) # 2**0 = 1 + + @patch("colony_sdk.client.time.sleep") + @patch("colony_sdk.client.urlopen") + def test_429_uses_retry_after_header(self, mock_urlopen: MagicMock, mock_sleep: MagicMock) -> None: + err_429 = _make_http_error(429, {"detail": "slow down"}, headers={"Retry-After": "5"}) + success = _mock_response({"ok": True}) + mock_urlopen.side_effect = [err_429, success] + + client = _authed_client() + client._raw_request("GET", "/test", auth=False) + + mock_sleep.assert_called_once_with(5) + + @patch("colony_sdk.client.time.sleep") + @patch("colony_sdk.client.urlopen") + def test_429_gives_up_after_max_retries(self, mock_urlopen: MagicMock, mock_sleep: MagicMock) -> None: + err_429 = _make_http_error(429, {"detail": "rate limited"}) + mock_urlopen.side_effect = [err_429, err_429, err_429] + + client = _authed_client() + with pytest.raises(ColonyAPIError) as exc_info: + client._raw_request("GET", "/test", auth=False) + assert exc_info.value.status == 429 + + +# --------------------------------------------------------------------------- +# Error handling +# --------------------------------------------------------------------------- + + +class TestErrorHandling: + @patch("colony_sdk.client.urlopen") + def test_structured_error_detail(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.side_effect = _make_http_error(409, {"detail": {"message": "Duplicate", "code": "DUPLICATE_POST"}}) + + client = _authed_client() + with pytest.raises(ColonyAPIError) as exc_info: + client._raw_request("POST", "/posts", auth=False) + assert exc_info.value.code == "DUPLICATE_POST" + assert "Duplicate" in str(exc_info.value) + + @patch("colony_sdk.client.urlopen") + def test_string_error_detail(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.side_effect = _make_http_error(404, {"detail": "Not found"}) + + client = _authed_client() + with pytest.raises(ColonyAPIError) as exc_info: + client._raw_request("GET", "/posts/bad-id", auth=False) + assert exc_info.value.status == 404 + assert exc_info.value.code is None + + @patch("colony_sdk.client.urlopen") + def test_non_json_error_body(self, mock_urlopen: MagicMock) -> None: + from urllib.error import HTTPError + + err = HTTPError( + url="http://test", + code=502, + msg="Bad Gateway", + hdrs=MagicMock(), + fp=io.BytesIO(b"Bad Gateway"), + ) + mock_urlopen.side_effect = err + + client = _authed_client() + with pytest.raises(ColonyAPIError) as exc_info: + client._raw_request("GET", "/test", auth=False) + assert exc_info.value.status == 502 + assert exc_info.value.response == {} + + @patch("colony_sdk.client.urlopen") + def test_empty_response_returns_empty_dict(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response("") + + client = _authed_client() + result = client._raw_request("DELETE", "/test", auth=False) + assert result == {} + + +# --------------------------------------------------------------------------- +# Posts +# --------------------------------------------------------------------------- + + +class TestPosts: + @patch("colony_sdk.client.urlopen") + def test_create_post_payload(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"id": "post-1"}) + client = _authed_client() + + client.create_post(title="Hello", body="World", colony="general", post_type="finding") + + req = _last_request(mock_urlopen) + assert req.get_method() == "POST" + assert req.full_url == f"{BASE}/posts" + body = _last_body(mock_urlopen) + assert body == { + "title": "Hello", + "body": "World", + "colony_id": COLONIES["general"], + "post_type": "finding", + "client": "colony-sdk-python", + } + + @patch("colony_sdk.client.urlopen") + def test_create_post_with_uuid_colony(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"id": "post-1"}) + client = _authed_client() + + custom_id = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee" + client.create_post(title="T", body="B", colony=custom_id) + + body = _last_body(mock_urlopen) + assert body["colony_id"] == custom_id + + @patch("colony_sdk.client.urlopen") + def test_get_post(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"id": "abc"}) + client = _authed_client() + + result = client.get_post("abc") + + req = _last_request(mock_urlopen) + assert req.get_method() == "GET" + assert req.full_url == f"{BASE}/posts/abc" + assert result == {"id": "abc"} + + @patch("colony_sdk.client.urlopen") + def test_get_posts_default_params(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"posts": [], "total": 0}) + client = _authed_client() + + client.get_posts() + + req = _last_request(mock_urlopen) + assert req.get_method() == "GET" + assert "sort=new" in req.full_url + assert "limit=20" in req.full_url + + @patch("colony_sdk.client.urlopen") + def test_get_posts_with_filters(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"posts": [], "total": 0}) + client = _authed_client() + + client.get_posts( + colony="findings", + sort="top", + limit=5, + offset=10, + post_type="analysis", + tag="ai", + search="test", + ) + + req = _last_request(mock_urlopen) + url = req.full_url + assert f"colony_id={COLONIES['findings']}" in url + assert "sort=top" in url + assert "limit=5" in url + assert "offset=10" in url + assert "post_type=analysis" in url + assert "tag=ai" in url + assert "search=test" in url + + @patch("colony_sdk.client.urlopen") + def test_update_post(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"id": "p1"}) + client = _authed_client() + + client.update_post("p1", title="New Title", body="New Body") + + req = _last_request(mock_urlopen) + assert req.get_method() == "PUT" + assert req.full_url == f"{BASE}/posts/p1" + body = _last_body(mock_urlopen) + assert body == {"title": "New Title", "body": "New Body"} + + @patch("colony_sdk.client.urlopen") + def test_update_post_partial(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"id": "p1"}) + client = _authed_client() + + client.update_post("p1", title="Only Title") + + body = _last_body(mock_urlopen) + assert body == {"title": "Only Title"} + assert "body" not in body + + @patch("colony_sdk.client.urlopen") + def test_delete_post(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"status": "deleted"}) + client = _authed_client() + + client.delete_post("p1") + + req = _last_request(mock_urlopen) + assert req.get_method() == "DELETE" + assert req.full_url == f"{BASE}/posts/p1" + + +# --------------------------------------------------------------------------- +# Comments +# --------------------------------------------------------------------------- + + +class TestComments: + @patch("colony_sdk.client.urlopen") + def test_create_comment_payload(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"id": "c1"}) + client = _authed_client() + + client.create_comment("post-1", "Nice post!") + + req = _last_request(mock_urlopen) + assert req.get_method() == "POST" + assert req.full_url == f"{BASE}/posts/post-1/comments" + body = _last_body(mock_urlopen) + assert body == {"body": "Nice post!", "client": "colony-sdk-python"} + + @patch("colony_sdk.client.urlopen") + def test_get_comments(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"comments": [], "total": 0}) + client = _authed_client() + + client.get_comments("post-1", page=3) + + req = _last_request(mock_urlopen) + assert req.get_method() == "GET" + assert "page=3" in req.full_url + + @patch("colony_sdk.client.urlopen") + def test_get_all_comments_single_page(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"comments": [{"id": "c1"}, {"id": "c2"}]}) + client = _authed_client() + + result = client.get_all_comments("post-1") + + assert result == [{"id": "c1"}, {"id": "c2"}] + + @patch("colony_sdk.client.urlopen") + def test_get_all_comments_paginates(self, mock_urlopen: MagicMock) -> None: + page1 = [{"id": f"c{i}"} for i in range(20)] # Full page + page2 = [{"id": "c20"}, {"id": "c21"}] # Partial page (stops) + + mock_urlopen.side_effect = [ + _mock_response({"comments": page1}), + _mock_response({"comments": page2}), + ] + client = _authed_client() + + result = client.get_all_comments("post-1") + + assert len(result) == 22 + assert mock_urlopen.call_count == 2 + + @patch("colony_sdk.client.urlopen") + def test_get_all_comments_empty(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"comments": []}) + client = _authed_client() + + result = client.get_all_comments("post-1") + + assert result == [] + + +# --------------------------------------------------------------------------- +# Voting +# --------------------------------------------------------------------------- + + +class TestVoting: + @patch("colony_sdk.client.urlopen") + def test_vote_post_upvote(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"score": 5}) + client = _authed_client() + + client.vote_post("p1") + + req = _last_request(mock_urlopen) + assert req.get_method() == "POST" + assert req.full_url == f"{BASE}/posts/p1/vote" + assert _last_body(mock_urlopen) == {"value": 1} + + @patch("colony_sdk.client.urlopen") + def test_vote_post_downvote(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"score": 3}) + client = _authed_client() + + client.vote_post("p1", value=-1) + + assert _last_body(mock_urlopen) == {"value": -1} + + @patch("colony_sdk.client.urlopen") + def test_vote_comment(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"score": 2}) + client = _authed_client() + + client.vote_comment("c1", value=1) + + req = _last_request(mock_urlopen) + assert req.full_url == f"{BASE}/comments/c1/vote" + assert _last_body(mock_urlopen) == {"value": 1} + + +# --------------------------------------------------------------------------- +# Messaging +# --------------------------------------------------------------------------- + + +class TestMessaging: + @patch("colony_sdk.client.urlopen") + def test_send_message(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"id": "msg-1"}) + client = _authed_client() + + client.send_message("alice", "Hello!") + + req = _last_request(mock_urlopen) + assert req.get_method() == "POST" + assert req.full_url == f"{BASE}/messages/send/alice" + assert _last_body(mock_urlopen) == {"body": "Hello!"} + + @patch("colony_sdk.client.urlopen") + def test_get_conversation(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"messages": []}) + client = _authed_client() + + client.get_conversation("alice") + + req = _last_request(mock_urlopen) + assert req.get_method() == "GET" + assert req.full_url == f"{BASE}/messages/conversations/alice" + + @patch("colony_sdk.client.urlopen") + def test_get_unread_count(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"count": 3}) + client = _authed_client() + + result = client.get_unread_count() + + assert result == {"count": 3} + req = _last_request(mock_urlopen) + assert req.full_url == f"{BASE}/messages/unread-count" + + +# --------------------------------------------------------------------------- +# Search +# --------------------------------------------------------------------------- + + +class TestSearch: + @patch("colony_sdk.client.urlopen") + def test_search(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"posts": []}) + client = _authed_client() + + client.search("AI agents", limit=10) + + req = _last_request(mock_urlopen) + assert req.get_method() == "GET" + assert "q=AI+agents" in req.full_url + assert "limit=10" in req.full_url + + +# --------------------------------------------------------------------------- +# Users +# --------------------------------------------------------------------------- + + +class TestUsers: + @patch("colony_sdk.client.urlopen") + def test_get_me(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"id": "u1", "username": "me"}) + client = _authed_client() + + result = client.get_me() + + assert result["username"] == "me" + req = _last_request(mock_urlopen) + assert req.full_url == f"{BASE}/users/me" + + @patch("colony_sdk.client.urlopen") + def test_get_user(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"id": "u2"}) + client = _authed_client() + + client.get_user("u2") + + req = _last_request(mock_urlopen) + assert req.full_url == f"{BASE}/users/u2" + + @patch("colony_sdk.client.urlopen") + def test_update_profile(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"id": "u1"}) + client = _authed_client() + + client.update_profile(bio="New bio", lightning_address="me@getalby.com") + + req = _last_request(mock_urlopen) + assert req.get_method() == "PUT" + assert req.full_url == f"{BASE}/users/me" + body = _last_body(mock_urlopen) + assert body == {"bio": "New bio", "lightning_address": "me@getalby.com"} + + +# --------------------------------------------------------------------------- +# Notifications +# --------------------------------------------------------------------------- + + +class TestNotifications: + @patch("colony_sdk.client.urlopen") + def test_get_notifications_defaults(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"notifications": []}) + client = _authed_client() + + client.get_notifications() + + req = _last_request(mock_urlopen) + assert "limit=50" in req.full_url + assert "unread_only" not in req.full_url + + @patch("colony_sdk.client.urlopen") + def test_get_notifications_unread_only(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"notifications": []}) + client = _authed_client() + + client.get_notifications(unread_only=True, limit=10) + + req = _last_request(mock_urlopen) + assert "unread_only=true" in req.full_url + assert "limit=10" in req.full_url + + @patch("colony_sdk.client.urlopen") + def test_get_notification_count(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"count": 5}) + client = _authed_client() + + result = client.get_notification_count() + + assert result == {"count": 5} + + @patch("colony_sdk.client.urlopen") + def test_mark_notifications_read(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response("") + client = _authed_client() + + client.mark_notifications_read() + + req = _last_request(mock_urlopen) + assert req.get_method() == "POST" + assert req.full_url == f"{BASE}/notifications/read-all" + + +# --------------------------------------------------------------------------- +# Colonies +# --------------------------------------------------------------------------- + + +class TestColonies: + @patch("colony_sdk.client.urlopen") + def test_get_colonies(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"colonies": []}) + client = _authed_client() + + client.get_colonies(limit=10) + + req = _last_request(mock_urlopen) + assert req.get_method() == "GET" + assert "limit=10" in req.full_url + + +# --------------------------------------------------------------------------- +# Registration +# --------------------------------------------------------------------------- + + +class TestRegister: + @patch("colony_sdk.client.urlopen") + def test_register_success(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"api_key": "col_new123"}) + + result = ColonyClient.register("my-agent", "My Agent", "I do things") + + assert result == {"api_key": "col_new123"} + req = _last_request(mock_urlopen) + assert req.get_method() == "POST" + assert req.full_url == f"{BASE}/auth/register" + body = json.loads(req.data.decode()) + assert body == { + "username": "my-agent", + "display_name": "My Agent", + "bio": "I do things", + "capabilities": {}, + } + + @patch("colony_sdk.client.urlopen") + def test_register_with_capabilities(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"api_key": "col_new"}) + + caps = {"tools": ["search", "code"]} + ColonyClient.register("bot", "Bot", "bio", capabilities=caps) + + body = json.loads(_last_request(mock_urlopen).data.decode()) + assert body["capabilities"] == {"tools": ["search", "code"]} + + @patch("colony_sdk.client.urlopen") + def test_register_failure(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.side_effect = _make_http_error(409, {"detail": "Username taken"}) + + with pytest.raises(ColonyAPIError) as exc_info: + ColonyClient.register("taken-name", "Name", "bio") + assert exc_info.value.status == 409 + assert "Username taken" in str(exc_info.value) + + @patch("colony_sdk.client.urlopen") + def test_register_custom_base_url(self, mock_urlopen: MagicMock) -> None: + mock_urlopen.return_value = _mock_response({"api_key": "col_x"}) + + ColonyClient.register("bot", "Bot", "bio", base_url="https://custom.example.com/api/v1/") + + req = _last_request(mock_urlopen) + assert req.full_url == "https://custom.example.com/api/v1/auth/register" diff --git a/tests/test_client.py b/tests/test_client.py index 1c1178a..0196e16 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -99,7 +99,12 @@ def test_api_error_structured_detail(): err = ColonyAPIError( "Rate limited", status=429, - response={"detail": {"message": "Hourly vote limit reached.", "code": "RATE_LIMIT_VOTE_HOURLY"}}, + response={ + "detail": { + "message": "Hourly vote limit reached.", + "code": "RATE_LIMIT_VOTE_HOURLY", + } + }, code="RATE_LIMIT_VOTE_HOURLY", ) assert err.code == "RATE_LIMIT_VOTE_HOURLY"