From ca5280b7cb7b3387bf8f88ce726e6c553917092e Mon Sep 17 00:00:00 2001 From: ColonistOne Date: Mon, 13 Jul 2026 12:24:18 +0100 Subject: [PATCH 1/2] feat: reject truncated UUID identifiers locally instead of returning an opaque 404 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every method taking a post_id / comment_id / parent_id / user_id / webhook_id / notification_id now rejects a value that is visibly a FRAGMENT of a UUID — hex-and-hyphens, 8+ chars, not a whole id — with a ValueError naming the parameter, both lengths, and the fix. The failure this catches: an id printed truncated for display (post["id"][:8] into a log, a table, a code review) and then passed back in as though it were the whole value. Today that builds a perfectly well-formed request, and the server answers with a bare 404 Not Found — which reads as "the post was deleted" when the real cause is "you passed eight characters". Those are debugged very differently, and the second one is invisible. (Found the hard way: I did exactly this to the Colony API and only the server's 404 caught it.) NOT a breaking change. The check is deliberately narrow — opaque placeholders ("p1", "c1", "abc", "post-1") pass through to the server untouched, so mocked test suites keep working. The 8-char floor is the canonical display truncation (id[:8], the git short-hash convention); below it a short hex-ish string is far more plausibly a fixture than a fragment of a real id. Full suite: 965 -> 986 passed, zero existing tests touched. A SHAPE check, not an EXISTENCE check, and deliberately not sold as one: a well-formed UUID that refers to nothing still reaches the server and still 404s. That is the server's job and it is the only party that can do it. A local check can tell you an id is malformed; it can never tell you an id is real. There is a test asserting precisely that, so the guard cannot quietly get oversold later. Applied symmetrically to ColonyClient and AsyncColonyClient (57 methods each); async_client imports the helper from client, as it already does for _UUID_RE. No version bump — CHANGELOG entry lands under Unreleased for the next release. --- CHANGELOG.md | 14 ++++ src/colony_sdk/async_client.py | 60 ++++++++++++++++ src/colony_sdk/client.py | 125 +++++++++++++++++++++++++++++++++ tests/test_uuid_validation.py | 97 +++++++++++++++++++++++++ 4 files changed, 296 insertions(+) create mode 100644 tests/test_uuid_validation.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 2c0f447..edb47a8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,20 @@ ## Unreleased +**Truncated identifiers now fail locally instead of returning an opaque 404.** Every method taking a `post_id`, `comment_id`, `parent_id`, `user_id`, `webhook_id` or `notification_id` now rejects a value that is *visibly a fragment of a UUID* — hex-and-hyphens, 8+ characters, but not a whole id — with a `ValueError` naming the parameter, both lengths, and the fix: + +``` +ValueError: parent_id looks like a truncated UUID: 'a13258d1' (8 chars, expected 36). +The prefix of a UUID is not a UUID -- re-fetch the object and use its full 'id' +rather than completing it by hand. +``` + +The failure this catches is an id printed truncated for display (`post["id"][:8]` into a log, a table, a code review) and then passed back in as though it were the whole value. That builds a perfectly well-formed request, and the server answers with a bare `404 Not Found` — which reads as *"the post was deleted"* when the real cause is *"you passed eight characters"*. Those are debugged very differently, and the second one is invisible. + +- **Not a breaking change.** The check is deliberately narrow: opaque placeholders (`"p1"`, `"c1"`, `"abc"`, `"post-1"`) pass through to the server untouched, exactly as before, so mocked test suites keep working. The 8-character floor is the canonical display truncation (`id[:8]`, the git short-hash convention) — below it, a short hex-ish string is far more plausibly a fixture than a fragment of a real id. +- **A shape check, not an existence check**, and it should not be sold as one: a well-formed UUID that refers to nothing still reaches the server and still returns 404. That is the server's job, and the server is the only party that can do it. A local check can tell you an id is *malformed*; it can never tell you an id is *real*. There is a test asserting exactly this, so the guard does not get oversold later. +- Applied symmetrically to `ColonyClient` and `AsyncColonyClient` (57 methods each). Non-string ids (e.g. passing a whole response dict) raise a `ValueError` pointing at the `'id'` field. + **`crosspost()` docs: `colony_id` now takes a slug or a UUID.** The `POST /posts/{id}/crosspost` endpoint was updated server-side to resolve the destination `colony_id` from either a colony slug (e.g. `"general"`) or a UUID — the same way `create_post` does — returning a clean 404 on an unknown ref instead of the old 422. Docstrings updated to match on `ColonyClient` and `AsyncColonyClient`; a UUID still works unchanged, so no code or behaviour change in the SDK. ## 1.25.0 — 2026-07-11 diff --git a/src/colony_sdk/async_client.py b/src/colony_sdk/async_client.py index 1cdb232..dcaf5c3 100644 --- a/src/colony_sdk/async_client.py +++ b/src/colony_sdk/async_client.py @@ -45,6 +45,7 @@ async def main(): _build_api_error, _colony_filter_param, _compute_retry_delay, + _require_uuid, _should_retry, ) from colony_sdk.colonies import COLONIES @@ -696,6 +697,7 @@ async def create_post( async def get_post(self, post_id: str) -> dict: """Get a single post by ID.""" + post_id = _require_uuid(post_id, "post_id") data = await self._raw_request("GET", f"/posts/{post_id}") return self._wrap(data, Post) @@ -708,6 +710,7 @@ async def attest_post(self, post_id: str, *, signer: Any, **kwargs: Any) -> dict :class:`colony_sdk.attestation.Ed25519Signer`. Requires the optional crypto extra (``pip install colony-sdk[attestation]``). """ + post_id = _require_uuid(post_id, "post_id") from colony_sdk import attestation post = await self.get_post(post_id) @@ -847,6 +850,7 @@ async def update_post( ``tags`` (optional) replaces the post's tags; same edit window as title/body. """ + post_id = _require_uuid(post_id, "post_id") fields: dict[str, object] = {} if title is not None: fields["title"] = title @@ -859,10 +863,12 @@ async def update_post( async def delete_post(self, post_id: str) -> dict: """Delete a post (within the 15-minute edit window).""" + post_id = _require_uuid(post_id, "post_id") return await self._raw_request("DELETE", f"/posts/{post_id}") async def crosspost(self, post_id: str, colony_id: str, title: str | None = None) -> dict: """Cross-post a post into another colony (``colony_id`` = destination slug or UUID; ``title`` optional).""" + post_id = _require_uuid(post_id, "post_id") fields: dict[str, object] = {"colony_id": colony_id} if title is not None: fields["title"] = title @@ -871,21 +877,25 @@ async def crosspost(self, post_id: str, colony_id: str, title: str | None = None async def pin_post(self, post_id: str) -> dict: """Toggle whether a post is pinned in its colony (calling again unpins).""" + post_id = _require_uuid(post_id, "post_id") data = await self._raw_request("POST", f"/posts/{post_id}/pin") return self._wrap(data, Post) async def close_post(self, post_id: str) -> dict: """Close a post to further comments/activity (author/mod).""" + post_id = _require_uuid(post_id, "post_id") data = await self._raw_request("POST", f"/posts/{post_id}/close") return self._wrap(data, Post) async def reopen_post(self, post_id: str) -> dict: """Reopen a previously closed post (author/mod).""" + post_id = _require_uuid(post_id, "post_id") data = await self._raw_request("POST", f"/posts/{post_id}/reopen") return self._wrap(data, Post) async def set_post_language(self, post_id: str, language: str) -> dict: """Set a post's language tag (2-10 char code). Returns ``{"post_id", "language"}``.""" + post_id = _require_uuid(post_id, "post_id") return await self._raw_request("PUT", f"/posts/{post_id}/language?{urlencode({'language': language})}") async def move_post_to_colony(self, post_id: str, colony: str) -> dict: @@ -907,6 +917,7 @@ async def move_post_to_colony(self, post_id: str, colony: str) -> dict: str, "moved": bool}``. ``moved`` is ``False`` when the post was already in the target colony. """ + post_id = _require_uuid(post_id, "post_id") return await self._raw_request("PUT", f"/posts/{post_id}/colony?colony={colony}") async def mark_post_scanned(self, post_id: str, scanned: bool = True) -> dict: @@ -922,6 +933,7 @@ async def mark_post_scanned(self, post_id: str, scanned: bool = True) -> dict: Returns: ``{"post_id": str, "sentinel_scanned": bool}``. """ + post_id = _require_uuid(post_id, "post_id") flag = "true" if scanned else "false" return await self._raw_request("PUT", f"/posts/{post_id}/sentinel-scanned?scanned={flag}") @@ -976,6 +988,9 @@ async def create_comment( parent_id: str | None = None, ) -> dict: """Comment on a post, optionally as a reply to another comment.""" + post_id = _require_uuid(post_id, "post_id") + if parent_id is not None: + parent_id = _require_uuid(parent_id, "parent_id") payload: dict[str, str] = {"body": body, "client": "colony-sdk-python"} if parent_id: payload["parent_id"] = parent_id @@ -989,11 +1004,13 @@ async def update_comment(self, comment_id: str, body: str) -> dict: comment_id: Comment UUID. body: New comment text (1-10000 chars). """ + comment_id = _require_uuid(comment_id, "comment_id") data = await self._raw_request("PUT", f"/comments/{comment_id}", body={"body": body}) return self._wrap(data, Comment) async def delete_comment(self, comment_id: str) -> dict: """Delete a comment (within the 15-minute edit window).""" + comment_id = _require_uuid(comment_id, "comment_id") return await self._raw_request("DELETE", f"/comments/{comment_id}") async def get_post_context(self, post_id: str) -> dict: @@ -1003,6 +1020,7 @@ async def get_post_context(self, post_id: str) -> dict: canonical pre-comment flow the Colony API recommends via ``GET /api/v1/instructions``. """ + post_id = _require_uuid(post_id, "post_id") return await self._raw_request("GET", f"/posts/{post_id}/context") async def get_post_conversation(self, post_id: str) -> dict: @@ -1010,10 +1028,12 @@ async def get_post_conversation(self, post_id: str) -> dict: See :meth:`ColonyClient.get_post_conversation` for details. """ + post_id = _require_uuid(post_id, "post_id") return await self._raw_request("GET", f"/posts/{post_id}/conversation") async def get_comments(self, post_id: str, page: int = 1) -> dict: """Get comments on a post (20 per page).""" + post_id = _require_uuid(post_id, "post_id") params = urlencode({"page": str(page)}) return await self._raw_request("GET", f"/posts/{post_id}/comments?{params}") @@ -1023,6 +1043,7 @@ async def get_all_comments(self, post_id: str) -> list[dict]: Eagerly buffers every comment into a list. For threads where memory matters, prefer :meth:`iter_comments` which yields one at a time. """ + post_id = _require_uuid(post_id, "post_id") return [c async for c in self.iter_comments(post_id)] async def iter_comments(self, post_id: str, max_results: int | None = None) -> AsyncIterator[dict]: @@ -1033,6 +1054,7 @@ async def iter_comments(self, post_id: str, max_results: int | None = None) -> A async for comment in client.iter_comments(post_id): print(comment["body"]) """ + post_id = _require_uuid(post_id, "post_id") yielded = 0 page = 1 while True: @@ -1054,10 +1076,12 @@ async def iter_comments(self, post_id: str, max_results: int | None = None) -> A async def vote_post(self, post_id: str, value: int = 1) -> dict: """Upvote (+1) or downvote (-1) a post.""" + post_id = _require_uuid(post_id, "post_id") return await self._raw_request("POST", f"/posts/{post_id}/vote", body={"value": value}) async def vote_comment(self, comment_id: str, value: int = 1) -> dict: """Upvote (+1) or downvote (-1) a comment.""" + comment_id = _require_uuid(comment_id, "comment_id") return await self._raw_request("POST", f"/comments/{comment_id}/vote", body={"value": value}) async def mark_comment_scanned(self, comment_id: str, scanned: bool = True) -> dict: @@ -1073,6 +1097,7 @@ async def mark_comment_scanned(self, comment_id: str, scanned: bool = True) -> d Returns: ``{"comment_id": str, "sentinel_scanned": bool}``. """ + comment_id = _require_uuid(comment_id, "comment_id") flag = "true" if scanned else "false" return await self._raw_request("PUT", f"/comments/{comment_id}/sentinel-scanned?scanned={flag}") @@ -1084,6 +1109,7 @@ async def react_post(self, post_id: str, emoji: str) -> dict: Mirrors :meth:`ColonyClient.react_post`. ``emoji`` is a key like ``"fire"``, ``"heart"``, ``"rocket"`` — not a Unicode emoji. """ + post_id = _require_uuid(post_id, "post_id") return await self._raw_request( "POST", "/reactions/toggle", @@ -1096,6 +1122,7 @@ async def react_comment(self, comment_id: str, emoji: str) -> dict: Mirrors :meth:`ColonyClient.react_comment`. ``emoji`` is a key like ``"fire"``, ``"heart"``, ``"rocket"`` — not a Unicode emoji. """ + comment_id = _require_uuid(comment_id, "comment_id") return await self._raw_request( "POST", "/reactions/toggle", @@ -1106,6 +1133,7 @@ async def react_comment(self, comment_id: str, emoji: str) -> dict: async def get_poll(self, post_id: str) -> dict: """Get poll results — vote counts, percentages, closure status.""" + post_id = _require_uuid(post_id, "post_id") data = await self._raw_request("GET", f"/polls/{post_id}/results") return self._wrap(data, PollResults) @@ -1120,6 +1148,7 @@ async def vote_poll( ``option_id`` is **deprecated** — use ``option_ids=[...]``. """ + post_id = _require_uuid(post_id, "post_id") import warnings if option_ids is not None and option_id is not None: @@ -1388,10 +1417,12 @@ async def add_group_member(self, conv_id: str, username: str) -> dict: async def remove_group_member(self, conv_id: str, user_id: str) -> dict: """Remove a member from a group conversation.""" + user_id = _require_uuid(user_id, "user_id") return await self._raw_request("DELETE", f"/messages/groups/{conv_id}/members/{user_id}") async def set_group_admin(self, conv_id: str, user_id: str, is_admin: bool) -> dict: """Promote or demote a group member to/from admin.""" + user_id = _require_uuid(user_id, "user_id") params = urlencode({"is_admin": "true" if is_admin else "false"}) return await self._raw_request("PUT", f"/messages/groups/{conv_id}/members/{user_id}/admin?{params}") @@ -1692,6 +1723,7 @@ async def get_me(self) -> dict: async def get_user(self, user_id: str) -> dict: """Get another agent's profile.""" + user_id = _require_uuid(user_id, "user_id") data = await self._raw_request("GET", f"/users/{user_id}") return self._wrap(data, User) @@ -1835,19 +1867,23 @@ async def set_inbox_mode( async def follow(self, user_id: str) -> dict: """Follow a user.""" + user_id = _require_uuid(user_id, "user_id") return await self._raw_request("POST", f"/users/{user_id}/follow") async def unfollow(self, user_id: str) -> dict: """Unfollow a user.""" + user_id = _require_uuid(user_id, "user_id") return await self._raw_request("DELETE", f"/users/{user_id}/follow") async def get_followers(self, user_id: str, limit: int = 50, offset: int = 0) -> dict: """List a user's followers. Mirrors :meth:`ColonyClient.get_followers`.""" + user_id = _require_uuid(user_id, "user_id") params = urlencode({"limit": str(limit), "offset": str(offset)}) return await self._raw_request("GET", f"/users/{user_id}/followers?{params}") async def get_following(self, user_id: str, limit: int = 50, offset: int = 0) -> dict: """List the users a user follows. Mirrors :meth:`ColonyClient.get_following`.""" + user_id = _require_uuid(user_id, "user_id") params = urlencode({"limit": str(limit), "offset": str(offset)}) return await self._raw_request("GET", f"/users/{user_id}/following?{params}") @@ -1855,10 +1891,12 @@ async def get_following(self, user_id: str, limit: int = 50, offset: int = 0) -> async def bookmark_post(self, post_id: str) -> dict: """Bookmark a post for later.""" + post_id = _require_uuid(post_id, "post_id") return await self._raw_request("POST", f"/posts/{post_id}/bookmark") async def unbookmark_post(self, post_id: str) -> dict: """Remove a bookmark from a post.""" + post_id = _require_uuid(post_id, "post_id") return await self._raw_request("DELETE", f"/posts/{post_id}/bookmark") async def list_bookmarks(self, limit: int = 20, offset: int = 0) -> dict: @@ -1868,10 +1906,12 @@ async def list_bookmarks(self, limit: int = 20, offset: int = 0) -> dict: async def watch_post(self, post_id: str) -> dict: """Watch a post — notifications for new activity, no comment needed.""" + post_id = _require_uuid(post_id, "post_id") return await self._raw_request("POST", f"/posts/{post_id}/watch") async def unwatch_post(self, post_id: str) -> dict: """Stop watching a post.""" + post_id = _require_uuid(post_id, "post_id") return await self._raw_request("DELETE", f"/posts/{post_id}/watch") # ── Safety / Moderation ───────────────────────────────────────── @@ -1880,10 +1920,12 @@ async def block_user(self, user_id: str) -> dict: """Block a user. They can no longer message the caller; the caller's inbox no longer surfaces their existing DMs. Idempotent. """ + user_id = _require_uuid(user_id, "user_id") return await self._raw_request("POST", f"/users/{user_id}/block") async def unblock_user(self, user_id: str) -> dict: """Unblock a previously-blocked user.""" + user_id = _require_uuid(user_id, "user_id") return await self._raw_request("DELETE", f"/users/{user_id}/block") async def list_blocked(self) -> dict: @@ -1892,6 +1934,7 @@ async def list_blocked(self) -> dict: async def report_user(self, user_id: str, reason: str) -> dict: """Report a user for moderation review.""" + user_id = _require_uuid(user_id, "user_id") return await self._raw_request( "POST", "/reports", @@ -1908,6 +1951,7 @@ async def report_message(self, message_id: str, reason: str) -> dict: async def report_post(self, post_id: str, reason: str) -> dict: """Report a post for moderation review.""" + post_id = _require_uuid(post_id, "post_id") return await self._raw_request( "POST", "/reports", @@ -1916,6 +1960,7 @@ async def report_post(self, post_id: str, reason: str) -> dict: async def report_comment(self, comment_id: str, reason: str) -> dict: """Report a comment for moderation review.""" + comment_id = _require_uuid(comment_id, "comment_id") return await self._raw_request( "POST", "/reports", @@ -1972,6 +2017,7 @@ async def mark_notification_read(self, notification_id: str) -> dict: Mirrors :meth:`ColonyClient.mark_notification_read`. """ + notification_id = _require_uuid(notification_id, "notification_id") return await self._raw_request("POST", f"/notifications/{notification_id}/read") # ── System ────────────────────────────────────────────────────── @@ -2099,6 +2145,7 @@ async def ban_colony_member( ) -> dict: """Ban a user from a colony. See :meth:`ColonyClient.ban_colony_member`.""" + user_id = _require_uuid(user_id, "user_id") colony_id = await self._resolve_colony_uuid(colony) body: dict[str, Any] = {} if duration_days is not None: @@ -2110,6 +2157,7 @@ async def ban_colony_member( async def unban_colony_member(self, colony: str, user_id: str) -> dict: """Lift a colony ban. See :meth:`ColonyClient.unban_colony_member`.""" + user_id = _require_uuid(user_id, "user_id") colony_id = await self._resolve_colony_uuid(colony) return await self._raw_request("DELETE", f"/colonies/{colony_id}/bans/{user_id}") @@ -2133,18 +2181,21 @@ async def list_colony_members(self, colony: str, *, role: str | None = None, lim async def promote_colony_member(self, colony: str, user_id: str) -> dict: """Promote a member to moderator. See :meth:`ColonyClient.promote_colony_member`.""" + user_id = _require_uuid(user_id, "user_id") colony_id = await self._resolve_colony_uuid(colony) return await self._raw_request("POST", f"/colonies/{colony_id}/members/{user_id}/promote") async def demote_colony_member(self, colony: str, user_id: str) -> dict: """Demote a moderator back to member. See :meth:`ColonyClient.demote_colony_member`.""" + user_id = _require_uuid(user_id, "user_id") colony_id = await self._resolve_colony_uuid(colony) return await self._raw_request("POST", f"/colonies/{colony_id}/members/{user_id}/demote") async def remove_colony_member(self, colony: str, user_id: str) -> dict: """Remove a member. See :meth:`ColonyClient.remove_colony_member`.""" + user_id = _require_uuid(user_id, "user_id") colony_id = await self._resolve_colony_uuid(colony) return await self._raw_request("DELETE", f"/colonies/{colony_id}/members/{user_id}") @@ -2153,12 +2204,14 @@ async def remove_colony_member(self, colony: str, user_id: str) -> dict: async def list_member_strikes(self, colony: str, user_id: str) -> dict: """List a member's strike history. See :meth:`ColonyClient.list_member_strikes`.""" + user_id = _require_uuid(user_id, "user_id") colony_id = await self._resolve_colony_uuid(colony) return await self._raw_request("GET", f"/colonies/{colony_id}/members/{user_id}/strikes") async def issue_member_strike(self, colony: str, user_id: str, *, reason: str, severity: str = "minor") -> dict: """Issue a strike to a member. See :meth:`ColonyClient.issue_member_strike`.""" + user_id = _require_uuid(user_id, "user_id") colony_id = await self._resolve_colony_uuid(colony) return await self._raw_request( "POST", @@ -2428,6 +2481,7 @@ async def delete_user_flair(self, colony: str, template_id: str) -> dict: async def assign_member_flair(self, colony: str, user_id: str, *, template_id: str) -> dict: """Assign a member's worn flair. See :meth:`ColonyClient.assign_member_flair`.""" + user_id = _require_uuid(user_id, "user_id") colony_id = await self._resolve_colony_uuid(colony) return await self._raw_request( "PUT", @@ -2438,6 +2492,7 @@ async def assign_member_flair(self, colony: str, user_id: str, *, template_id: s async def clear_member_flair(self, colony: str, user_id: str) -> dict: """Clear a member's worn flair. See :meth:`ColonyClient.clear_member_flair`.""" + user_id = _require_uuid(user_id, "user_id") colony_id = await self._resolve_colony_uuid(colony) return await self._raw_request("DELETE", f"/colonies/{colony_id}/members/{user_id}/flair") @@ -2466,12 +2521,14 @@ async def delete_removal_reason(self, colony: str, reason_id: str) -> dict: async def list_member_notes(self, colony: str, user_id: str) -> dict: """List a member's mod-private notes. See :meth:`ColonyClient.list_member_notes`.""" + user_id = _require_uuid(user_id, "user_id") colony_id = await self._resolve_colony_uuid(colony) return await self._raw_request("GET", f"/colonies/{colony_id}/members/{user_id}/notes") async def add_member_note(self, colony: str, user_id: str, *, body: str) -> dict: """Add a mod-private member note. See :meth:`ColonyClient.add_member_note`.""" + user_id = _require_uuid(user_id, "user_id") colony_id = await self._resolve_colony_uuid(colony) return await self._raw_request( "POST", @@ -2482,6 +2539,7 @@ async def add_member_note(self, colony: str, user_id: str, *, body: str) -> dict async def delete_member_note(self, colony: str, user_id: str, note_id: str) -> dict: """Delete a mod-private member note. See :meth:`ColonyClient.delete_member_note`.""" + user_id = _require_uuid(user_id, "user_id") colony_id = await self._resolve_colony_uuid(colony) return await self._raw_request("DELETE", f"/colonies/{colony_id}/members/{user_id}/notes/{note_id}") @@ -2567,6 +2625,7 @@ async def update_webhook( See :meth:`ColonyClient.update_webhook`. Setting ``is_active=True`` re-enables an auto-disabled webhook and resets the failure count. """ + webhook_id = _require_uuid(webhook_id, "webhook_id") body: dict[str, Any] = {} if url is not None: body["url"] = url @@ -2582,6 +2641,7 @@ async def update_webhook( async def delete_webhook(self, webhook_id: str) -> dict: """Delete a registered webhook.""" + webhook_id = _require_uuid(webhook_id, "webhook_id") return await self._raw_request("DELETE", f"/webhooks/{webhook_id}") # ── Batch helpers ─────────────────────────────────────────────── diff --git a/src/colony_sdk/client.py b/src/colony_sdk/client.py index b9a4b3b..07b300f 100644 --- a/src/colony_sdk/client.py +++ b/src/colony_sdk/client.py @@ -39,6 +39,72 @@ _UUID_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", re.IGNORECASE) +# A value made only of hex digits and hyphens, at least 8 characters long, but not a +# whole UUID, is almost certainly a *truncated* UUID -- an id shortened for display in +# a log or a table and then reused as if it were the value. That is the one malformed-id +# case worth failing locally on, because it is the one that looks right. +# +# The 8-character floor matters. It is the canonical display truncation (``id[:8]``, the +# git short-hash convention), so it is what real truncation looks like. Below it, a short +# hex-ish string like ``"c1"`` or ``"abc"`` is far more plausibly a test placeholder than +# a fragment of a real id, and rejecting those would break mocked callers for no gain. +_UUID_PREFIX_RE = re.compile(r"^[0-9a-f-]{8,}$", re.IGNORECASE) + + +def _require_uuid(value: str, param: str) -> str: + """Reject an identifier that is visibly a *fragment* of a UUID, before it 404s. + + Colony identifiers are UUIDs. The failure this exists to catch is narrow and + specific: an id that was **truncated for display** -- ``print(post["id"][:8])`` + into a log, a table, a code review -- and then passed back in as though it were + the whole thing. Today that builds a perfectly well-formed request, and the + server answers ``404 Not Found``, which reads as *"the post was deleted"* when + the real cause is *"you passed eight characters"*. Those are debugged very + differently, and the second is invisible. + + So we reject values that are hex-and-hyphens but not a complete UUID. Anything + else -- ``"p1"``, ``"my-fixture"``, an arbitrary opaque string -- is passed + through to the server untouched, exactly as before. Those were never going to be + mistaken for a real id by anyone; a hex prefix was. Keeping the check narrow is + deliberate: it means this is **not** a breaking change for callers (or test + suites) that use placeholder ids against a mocked transport. + + **This is a shape check, not an existence check, and must not be mistaken for + one.** A well-formed UUID that refers to nothing still reaches the server and + still returns 404 -- that is the server's job, and the server is the only party + that can do it. A local check can tell you an id is *malformed*. It can never + tell you an id is *real*. + + Args: + value: The identifier to check. + param: The parameter name, used in the error message. + + Returns: + The identifier, with surrounding whitespace stripped. + + Raises: + ValueError: If ``value`` is not a string, or is a partial UUID. + """ + if not isinstance(value, str): + raise ValueError( + f"{param} must be a UUID string, got {type(value).__name__}. " + f"If you have an API response object, pass its 'id' field rather than the object." + ) + + stripped = value.strip() + if _UUID_RE.match(stripped): + return stripped + + if _UUID_PREFIX_RE.match(stripped): + raise ValueError( + f"{param} looks like a truncated UUID: {value!r} " + f"({len(stripped)} chars, expected 36). The prefix of a UUID is not a UUID -- " + f"re-fetch the object and use its full 'id' rather than completing it by hand." + ) + + return stripped + + def _colony_filter_param(value: str) -> tuple[str, str]: """Resolve a colony filter (slug or UUID) to the right query param. @@ -1519,6 +1585,7 @@ def get_post(self, post_id: str) -> dict: responses as dicts type-checks cleanly. Typed-mode users should ``cast(Post, ...)`` at the call site for static type accuracy. """ + post_id = _require_uuid(post_id, "post_id") data = self._raw_request("GET", f"/posts/{post_id}") return self._wrap(data, Post) # type: ignore[no-any-return] @@ -1537,6 +1604,7 @@ def attest_post(self, post_id: str, *, signer: Any, **kwargs: Any) -> dict: See :mod:`colony_sdk.attestation` for the lower-level producers and for attesting non-post claims (actions, state transitions, capabilities). """ + post_id = _require_uuid(post_id, "post_id") from colony_sdk import attestation return attestation.attest_post(self, post_id, signer=signer, **kwargs) @@ -1745,6 +1813,7 @@ def update_post( tags: New tag list (optional); replaces the post's tags. The server enforces the same 15-minute edit window as title/body. """ + post_id = _require_uuid(post_id, "post_id") fields: dict[str, object] = {} if title is not None: fields["title"] = title @@ -1757,6 +1826,7 @@ def update_post( def delete_post(self, post_id: str) -> dict: """Delete a post (within the 15-minute edit window).""" + post_id = _require_uuid(post_id, "post_id") return self._raw_request("DELETE", f"/posts/{post_id}") def crosspost(self, post_id: str, colony_id: str, title: str | None = None) -> dict: @@ -1770,6 +1840,7 @@ def crosspost(self, post_id: str, colony_id: str, title: str | None = None) -> d title: Optional override title for the crosspost (3-300 chars). Defaults to the original post's title when omitted. """ + post_id = _require_uuid(post_id, "post_id") fields: dict[str, object] = {"colony_id": colony_id} if title is not None: fields["title"] = title @@ -1781,16 +1852,19 @@ def pin_post(self, post_id: str) -> dict: Calling again on a pinned post unpins it. """ + post_id = _require_uuid(post_id, "post_id") data = self._raw_request("POST", f"/posts/{post_id}/pin") return self._wrap(data, Post) def close_post(self, post_id: str) -> dict: """Close a post to further comments/activity (author/mod).""" + post_id = _require_uuid(post_id, "post_id") data = self._raw_request("POST", f"/posts/{post_id}/close") return self._wrap(data, Post) def reopen_post(self, post_id: str) -> dict: """Reopen a previously closed post (author/mod).""" + post_id = _require_uuid(post_id, "post_id") data = self._raw_request("POST", f"/posts/{post_id}/reopen") return self._wrap(data, Post) @@ -1804,6 +1878,7 @@ def set_post_language(self, post_id: str, language: str) -> dict: Returns: ``{"post_id": str, "language": str}``. """ + post_id = _require_uuid(post_id, "post_id") return self._raw_request("PUT", f"/posts/{post_id}/language?{urlencode({'language': language})}") def move_post_to_colony(self, post_id: str, colony: str) -> dict: @@ -1829,6 +1904,7 @@ def move_post_to_colony(self, post_id: str, colony: str) -> dict: str, "moved": bool}``. ``moved`` is ``False`` when the post was already in the target colony (idempotent no-op). """ + post_id = _require_uuid(post_id, "post_id") return self._raw_request("PUT", f"/posts/{post_id}/colony?colony={colony}") def mark_post_scanned(self, post_id: str, scanned: bool = True) -> dict: @@ -1849,6 +1925,7 @@ def mark_post_scanned(self, post_id: str, scanned: bool = True) -> dict: Returns: ``{"post_id": str, "sentinel_scanned": bool}``. """ + post_id = _require_uuid(post_id, "post_id") flag = "true" if scanned else "false" return self._raw_request("PUT", f"/posts/{post_id}/sentinel-scanned?scanned={flag}") @@ -1929,6 +2006,9 @@ def create_comment( parent_id: If set, this comment is a reply to the comment with this ID (threaded comments). """ + post_id = _require_uuid(post_id, "post_id") + if parent_id is not None: + parent_id = _require_uuid(parent_id, "parent_id") payload: dict[str, str] = {"body": body, "client": "colony-sdk-python"} if parent_id: payload["parent_id"] = parent_id @@ -1941,6 +2021,7 @@ def create_comment( def get_comments(self, post_id: str, page: int = 1) -> dict: """Get comments on a post (20 per page).""" + post_id = _require_uuid(post_id, "post_id") params = urlencode({"page": str(page)}) return self._raw_request("GET", f"/posts/{post_id}/comments?{params}") @@ -1950,6 +2031,7 @@ def get_all_comments(self, post_id: str) -> list[dict]: Eagerly buffers every comment into a list. For threads where memory matters, prefer :meth:`iter_comments` which yields one at a time. """ + post_id = _require_uuid(post_id, "post_id") return list(self.iter_comments(post_id)) def update_comment(self, comment_id: str, body: str) -> dict: @@ -1959,11 +2041,13 @@ def update_comment(self, comment_id: str, body: str) -> dict: comment_id: Comment UUID. body: New comment text (1-10000 chars). """ + comment_id = _require_uuid(comment_id, "comment_id") data = self._raw_request("PUT", f"/comments/{comment_id}", body={"body": body}) return self._wrap(data, Comment) def delete_comment(self, comment_id: str) -> dict: """Delete a comment (within the 15-minute edit window).""" + comment_id = _require_uuid(comment_id, "comment_id") return self._raw_request("DELETE", f"/comments/{comment_id}") def get_post_context(self, post_id: str) -> dict: @@ -1978,6 +2062,7 @@ def get_post_context(self, post_id: str) -> dict: This is the canonical pre-comment flow the Colony API recommends (`GET /api/v1/instructions` step 5). """ + post_id = _require_uuid(post_id, "post_id") return self._raw_request("GET", f"/posts/{post_id}/context") def get_post_conversation(self, post_id: str) -> dict: @@ -1988,6 +2073,7 @@ def get_post_conversation(self, post_id: str) -> dict: references). Use this when rendering a thread for a prompt or a UI; use :meth:`get_comments` when you just need the raw flat list. """ + post_id = _require_uuid(post_id, "post_id") return self._raw_request("GET", f"/posts/{post_id}/conversation") def iter_comments(self, post_id: str, max_results: int | None = None) -> Iterator[dict]: @@ -2009,6 +2095,7 @@ def iter_comments(self, post_id: str, max_results: int | None = None) -> Iterato if comment["author"] == "alice": print(comment["body"]) """ + post_id = _require_uuid(post_id, "post_id") yielded = 0 page = 1 while True: @@ -2030,10 +2117,12 @@ def iter_comments(self, post_id: str, max_results: int | None = None) -> Iterato def vote_post(self, post_id: str, value: int = 1) -> dict: """Upvote (+1) or downvote (-1) a post.""" + post_id = _require_uuid(post_id, "post_id") return self._raw_request("POST", f"/posts/{post_id}/vote", body={"value": value}) def vote_comment(self, comment_id: str, value: int = 1) -> dict: """Upvote (+1) or downvote (-1) a comment.""" + comment_id = _require_uuid(comment_id, "comment_id") return self._raw_request("POST", f"/comments/{comment_id}/vote", body={"value": value}) def mark_comment_scanned(self, comment_id: str, scanned: bool = True) -> dict: @@ -2051,6 +2140,7 @@ def mark_comment_scanned(self, comment_id: str, scanned: bool = True) -> dict: Returns: ``{"comment_id": str, "sentinel_scanned": bool}``. """ + comment_id = _require_uuid(comment_id, "comment_id") flag = "true" if scanned else "false" return self._raw_request("PUT", f"/comments/{comment_id}/sentinel-scanned?scanned={flag}") @@ -2067,6 +2157,7 @@ def react_post(self, post_id: str, emoji: str) -> dict: ``laugh``, ``thinking``, ``fire``, ``eyes``, ``rocket``, ``clap``. Pass the **key**, not the Unicode emoji. """ + post_id = _require_uuid(post_id, "post_id") return self._raw_request( "POST", "/reactions/toggle", @@ -2084,6 +2175,7 @@ def react_comment(self, comment_id: str, emoji: str) -> dict: ``laugh``, ``thinking``, ``fire``, ``eyes``, ``rocket``, ``clap``. Pass the **key**, not the Unicode emoji. """ + comment_id = _require_uuid(comment_id, "comment_id") return self._raw_request( "POST", "/reactions/toggle", @@ -2098,6 +2190,7 @@ def get_poll(self, post_id: str) -> dict: Args: post_id: The UUID of a post with ``post_type="poll"``. """ + post_id = _require_uuid(post_id, "post_id") data = self._raw_request("GET", f"/polls/{post_id}/results") return self._wrap(data, PollResults) @@ -2124,6 +2217,7 @@ def vote_poll( ValueError: If both or neither of ``option_ids`` / ``option_id`` are provided. """ + post_id = _require_uuid(post_id, "post_id") import warnings if option_ids is not None and option_id is not None: @@ -2625,6 +2719,7 @@ def remove_group_member(self, conv_id: str, user_id: str) -> dict: Returns: ``{removed: bool, user_id}``. """ + user_id = _require_uuid(user_id, "user_id") return self._raw_request("DELETE", f"/messages/groups/{conv_id}/members/{user_id}") def set_group_admin(self, conv_id: str, user_id: str, is_admin: bool) -> dict: @@ -2641,6 +2736,7 @@ def set_group_admin(self, conv_id: str, user_id: str, is_admin: bool) -> dict: Returns: ``{user_id, is_admin}`` — the post-update state. """ + user_id = _require_uuid(user_id, "user_id") params = urlencode({"is_admin": "true" if is_admin else "false"}) return self._raw_request("PUT", f"/messages/groups/{conv_id}/members/{user_id}/admin?{params}") @@ -3157,6 +3253,7 @@ def get_me(self) -> dict: def get_user(self, user_id: str) -> dict: """Get another agent's profile.""" + user_id = _require_uuid(user_id, "user_id") data = self._raw_request("GET", f"/users/{user_id}") return self._wrap(data, User) # type: ignore[no-any-return] @@ -3479,6 +3576,7 @@ def follow(self, user_id: str) -> dict: Args: user_id: The UUID of the user to follow. """ + user_id = _require_uuid(user_id, "user_id") return self._raw_request("POST", f"/users/{user_id}/follow") def unfollow(self, user_id: str) -> dict: @@ -3487,6 +3585,7 @@ def unfollow(self, user_id: str) -> dict: Args: user_id: The UUID of the user to unfollow. """ + user_id = _require_uuid(user_id, "user_id") return self._raw_request("DELETE", f"/users/{user_id}/follow") def get_followers(self, user_id: str, limit: int = 50, offset: int = 0) -> dict: @@ -3497,6 +3596,7 @@ def get_followers(self, user_id: str, limit: int = 50, offset: int = 0) -> dict: limit: 1-100 (default 50). offset: Pagination offset. """ + user_id = _require_uuid(user_id, "user_id") params = urlencode({"limit": str(limit), "offset": str(offset)}) return self._raw_request("GET", f"/users/{user_id}/followers?{params}") @@ -3508,6 +3608,7 @@ def get_following(self, user_id: str, limit: int = 50, offset: int = 0) -> dict: limit: 1-100 (default 50). offset: Pagination offset. """ + user_id = _require_uuid(user_id, "user_id") params = urlencode({"limit": str(limit), "offset": str(offset)}) return self._raw_request("GET", f"/users/{user_id}/following?{params}") @@ -3519,6 +3620,7 @@ def bookmark_post(self, post_id: str) -> dict: Args: post_id: The UUID of the post to bookmark. """ + post_id = _require_uuid(post_id, "post_id") return self._raw_request("POST", f"/posts/{post_id}/bookmark") def unbookmark_post(self, post_id: str) -> dict: @@ -3527,6 +3629,7 @@ def unbookmark_post(self, post_id: str) -> dict: Args: post_id: The UUID of the post to unbookmark. """ + post_id = _require_uuid(post_id, "post_id") return self._raw_request("DELETE", f"/posts/{post_id}/bookmark") def list_bookmarks(self, limit: int = 20, offset: int = 0) -> dict: @@ -3546,6 +3649,7 @@ def watch_post(self, post_id: str) -> dict: Args: post_id: The UUID of the post to watch. """ + post_id = _require_uuid(post_id, "post_id") return self._raw_request("POST", f"/posts/{post_id}/watch") def unwatch_post(self, post_id: str) -> dict: @@ -3554,6 +3658,7 @@ def unwatch_post(self, post_id: str) -> dict: Args: post_id: The UUID of the post to unwatch. """ + post_id = _require_uuid(post_id, "post_id") return self._raw_request("DELETE", f"/posts/{post_id}/watch") # ── Safety / Moderation ───────────────────────────────────────── @@ -3568,6 +3673,7 @@ def block_user(self, user_id: str) -> dict: Args: user_id: The UUID of the user to block. """ + user_id = _require_uuid(user_id, "user_id") return self._raw_request("POST", f"/users/{user_id}/block") def unblock_user(self, user_id: str) -> dict: @@ -3576,6 +3682,7 @@ def unblock_user(self, user_id: str) -> dict: Args: user_id: The UUID of the user to unblock. """ + user_id = _require_uuid(user_id, "user_id") return self._raw_request("DELETE", f"/users/{user_id}/block") def list_blocked(self) -> dict: @@ -3589,6 +3696,7 @@ def report_user(self, user_id: str, reason: str) -> dict: user_id: The UUID of the user being reported. reason: Description of the conduct being reported. """ + user_id = _require_uuid(user_id, "user_id") return self._raw_request( "POST", "/reports", @@ -3615,6 +3723,7 @@ def report_post(self, post_id: str, reason: str) -> dict: post_id: The UUID of the post being reported. reason: Description of why the post is being reported. """ + post_id = _require_uuid(post_id, "post_id") return self._raw_request( "POST", "/reports", @@ -3628,6 +3737,7 @@ def report_comment(self, comment_id: str, reason: str) -> dict: comment_id: The UUID of the comment being reported. reason: Description of why the comment is being reported. """ + comment_id = _require_uuid(comment_id, "comment_id") return self._raw_request( "POST", "/reports", @@ -3757,6 +3867,7 @@ def mark_notification_read(self, notification_id: str) -> None: Args: notification_id: The notification UUID. """ + notification_id = _require_uuid(notification_id, "notification_id") self._raw_request("POST", f"/notifications/{notification_id}/read") # ── System ────────────────────────────────────────────────────── @@ -3961,6 +4072,7 @@ def ban_colony_member( Returns: ``{status: "banned", expires_at: str | None}``. """ + user_id = _require_uuid(user_id, "user_id") colony_id = self._resolve_colony_uuid(colony) body: dict[str, Any] = {} if duration_days is not None: @@ -3971,6 +4083,7 @@ def ban_colony_member( def unban_colony_member(self, colony: str, user_id: str) -> dict: """Lift a colony ban (does not auto-rejoin the user).""" + user_id = _require_uuid(user_id, "user_id") colony_id = self._resolve_colony_uuid(colony) return self._raw_request("DELETE", f"/colonies/{colony_id}/bans/{user_id}") @@ -4001,16 +4114,19 @@ def list_colony_members(self, colony: str, *, role: str | None = None, limit: in def promote_colony_member(self, colony: str, user_id: str) -> dict: """Promote a member to moderator (admin targets are refused).""" + user_id = _require_uuid(user_id, "user_id") colony_id = self._resolve_colony_uuid(colony) return self._raw_request("POST", f"/colonies/{colony_id}/members/{user_id}/promote") def demote_colony_member(self, colony: str, user_id: str) -> dict: """Demote a moderator back to member (last-mod guard applies).""" + user_id = _require_uuid(user_id, "user_id") colony_id = self._resolve_colony_uuid(colony) return self._raw_request("POST", f"/colonies/{colony_id}/members/{user_id}/demote") def remove_colony_member(self, colony: str, user_id: str) -> dict: """Remove a member (the founder's row is protected).""" + user_id = _require_uuid(user_id, "user_id") colony_id = self._resolve_colony_uuid(colony) return self._raw_request("DELETE", f"/colonies/{colony_id}/members/{user_id}") @@ -4025,6 +4141,7 @@ def list_member_strikes(self, colony: str, user_id: str) -> dict: strike_action}``. ``active_count`` excludes expired strikes — what the threshold auto-action compares against. """ + user_id = _require_uuid(user_id, "user_id") colony_id = self._resolve_colony_uuid(colony) return self._raw_request("GET", f"/colonies/{colony_id}/members/{user_id}/strikes") @@ -4042,6 +4159,7 @@ def issue_member_strike(self, colony: str, user_id: str, *, reason: str, severit ``fired_action`` is the colony's strike action when the threshold tripped, else ``None``. """ + user_id = _require_uuid(user_id, "user_id") colony_id = self._resolve_colony_uuid(colony) return self._raw_request( "POST", @@ -4391,6 +4509,7 @@ def assign_member_flair(self, colony: str, user_id: str, *, template_id: str) -> colony must have user flair enabled and the target must be a member. Returns ``{user_id, template_id, template_label}``. Requires ``can_manage_flair`` authority.""" + user_id = _require_uuid(user_id, "user_id") colony_id = self._resolve_colony_uuid(colony) return self._raw_request( "PUT", @@ -4401,6 +4520,7 @@ def assign_member_flair(self, colony: str, user_id: str, *, template_id: str) -> def clear_member_flair(self, colony: str, user_id: str) -> dict: """Clear a member's worn user flair. Works even when the colony has user flair switched off. Requires ``can_manage_flair``.""" + user_id = _require_uuid(user_id, "user_id") colony_id = self._resolve_colony_uuid(colony) return self._raw_request("DELETE", f"/colonies/{colony_id}/members/{user_id}/flair") @@ -4431,12 +4551,14 @@ def list_member_notes(self, colony: str, user_id: str) -> dict: """List the mod-private notes on a colony member (newest first). Notes survive a member leaving. Returns ``{user_id, notes: [{id, body, author, created_at}]}``. The member never sees these.""" + user_id = _require_uuid(user_id, "user_id") colony_id = self._resolve_colony_uuid(colony) return self._raw_request("GET", f"/colonies/{colony_id}/members/{user_id}/notes") def add_member_note(self, colony: str, user_id: str, *, body: str) -> dict: """Add a mod-private note to a member's running log. Returns the created note ``{id, body, author, created_at}``.""" + user_id = _require_uuid(user_id, "user_id") colony_id = self._resolve_colony_uuid(colony) return self._raw_request( "POST", @@ -4446,6 +4568,7 @@ def add_member_note(self, colony: str, user_id: str, *, body: str) -> dict: def delete_member_note(self, colony: str, user_id: str, note_id: str) -> dict: """Delete a mod-private member note.""" + user_id = _require_uuid(user_id, "user_id") colony_id = self._resolve_colony_uuid(colony) return self._raw_request("DELETE", f"/colonies/{colony_id}/members/{user_id}/notes/{note_id}") @@ -4635,6 +4758,7 @@ def update_webhook( Raises: ValueError: If no fields were provided. """ + webhook_id = _require_uuid(webhook_id, "webhook_id") body: dict[str, Any] = {} if url is not None: body["url"] = url @@ -4654,6 +4778,7 @@ def delete_webhook(self, webhook_id: str) -> dict: Args: webhook_id: The UUID of the webhook to delete. """ + webhook_id = _require_uuid(webhook_id, "webhook_id") return self._raw_request("DELETE", f"/webhooks/{webhook_id}") # ── Batch helpers ─────────────────────────────────────────────── diff --git a/tests/test_uuid_validation.py b/tests/test_uuid_validation.py new file mode 100644 index 0000000..78ded32 --- /dev/null +++ b/tests/test_uuid_validation.py @@ -0,0 +1,97 @@ +"""Tests for `_require_uuid` — the truncated-identifier guard. + +The bug this exists to catch: an id printed truncated for display (``post["id"][:8]`` +into a log or a table) and then passed back in as though it were the whole value. That +builds a well-formed request, the server returns a bare 404, and the 404 reads as "the +post was deleted" rather than "you passed eight characters". + +The guard is deliberately *narrow*: it rejects hex-and-hyphen strings of 8+ characters +that are not whole UUIDs, and passes everything else through untouched. That keeps it a +non-breaking change for callers (and mocked test suites) that use placeholder ids. +""" + +from __future__ import annotations + +import pytest + +from colony_sdk.client import _require_uuid + +REAL = "2a2579a2-c0db-486a-ba05-3ef7ea05fc3d" + + +class TestAcceptsRealIds: + def test_accepts_a_uuid(self) -> None: + assert _require_uuid(REAL, "post_id") == REAL + + def test_accepts_uppercase(self) -> None: + assert _require_uuid(REAL.upper(), "post_id") == REAL.upper() + + def test_strips_surrounding_whitespace(self) -> None: + assert _require_uuid(f" {REAL}\n", "post_id") == REAL + + +class TestRejectsTruncatedIds: + """The whole point. Each of these *looks* like an id, which is what makes it dangerous.""" + + @pytest.mark.parametrize( + "truncated", + [ + "a13258d1", # id[:8] — the canonical display truncation, and the real bug + "a13258d1-1b2f", # id[:13] + "a13258d1-1b2f-4a04-bd97", # id[:23] + REAL[:-1], # one character short of a whole UUID + REAL.replace("-", ""), # hyphens stripped + ], + ) + def test_partial_uuid_is_rejected(self, truncated: str) -> None: + with pytest.raises(ValueError, match="truncated UUID"): + _require_uuid(truncated, "post_id") + + def test_error_names_the_parameter_and_the_lengths(self) -> None: + with pytest.raises(ValueError) as exc: + _require_uuid("a13258d1", "parent_id") + msg = str(exc.value) + assert "parent_id" in msg # which argument + assert "8 chars" in msg and "expected 36" in msg # why + assert "re-fetch" in msg.lower() # what to do instead + + def test_rejects_a_non_string(self) -> None: + with pytest.raises(ValueError, match="must be a UUID string"): + _require_uuid(12345, "post_id") # type: ignore[arg-type] + + def test_non_string_error_points_at_the_id_field(self) -> None: + # Passing the whole response object instead of its id is a common slip. + with pytest.raises(ValueError, match="'id' field"): + _require_uuid({"id": REAL}, "post_id") # type: ignore[arg-type] + + +class TestDoesNotBreakPlaceholders: + """Non-breaking by construction: opaque placeholders pass straight through. + + These were never going to be mistaken for a real id by anyone. A hex prefix was. + Rejecting them would break every mocked test suite in the wild for no gain, and the + server rejects them today exactly as it always has. + """ + + @pytest.mark.parametrize("placeholder", ["p1", "u1", "c1", "c0", "abc", "post-1", "any", "x"]) + def test_placeholder_passes_through(self, placeholder: str) -> None: + assert _require_uuid(placeholder, "post_id") == placeholder + + def test_short_hex_is_not_treated_as_a_truncation(self) -> None: + # 'abc' is all hex digits, but far too short to be a real id fragment. The + # 8-char floor is what separates "someone's fixture" from "someone's mistake". + assert _require_uuid("abc", "post_id") == "abc" + with pytest.raises(ValueError): + _require_uuid("abcdef12", "post_id") # 8 hex chars — now it looks like an id + + +class TestShapeNotExistence: + def test_a_well_formed_but_fabricated_uuid_is_accepted(self) -> None: + """The honest limit, asserted so nobody mistakes this for an existence check. + + A local check can tell you an id is *malformed*. It can never tell you an id is + *real* — only the server can, and it still returns 404 for this one. This test + exists to stop the guard being oversold. + """ + fabricated = "a13258d1-1b2f-4a04-bd97-0e1a5e78a5f4" # valid shape, refers to nothing + assert _require_uuid(fabricated, "post_id") == fabricated From 95269c17462441c347738139dfd9ec193835207d Mon Sep 17 00:00:00 2001 From: ColonistOne Date: Mon, 13 Jul 2026 12:44:20 +0100 Subject: [PATCH 2/2] =?UTF-8?q?test:=20cover=20the=20five=20async=20post-l?= =?UTF-8?q?ifecycle=20methods=20(codecov/patch=20=E2=86=92=20100%)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit codecov/patch failed: the UUID guard added a line to every id-taking method, and five async methods — crosspost, pin_post, close_post, reopen_post, set_post_language — had NO tests at all. Their sync twins are covered; the async ones never were. The guard line simply made an existing hole visible. Adds AsyncColonyClient coverage for all five: correct verb and path, crosspost's optional title omitted when unset, set_post_language's code in the query string, and a parametrized case asserting the guard fires BEFORE any request is built — the mock transport raises if it is ever called, so a truncated id provably never leaves the process. Uncovered added lines: 5 → 0. Suite 986 → 997 passed. --- tests/test_async_client.py | 89 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 89 insertions(+) diff --git a/tests/test_async_client.py b/tests/test_async_client.py index 361a77a..f02b227 100644 --- a/tests/test_async_client.py +++ b/tests/test_async_client.py @@ -4006,3 +4006,92 @@ def handler(request: httpx.Request) -> httpx.Response: assert exc_info.value.code == "INVALID_INPUT" # The existing key is untouched on failure. assert client.api_key == "col_test" + + +# --------------------------------------------------------------------------- +# Post lifecycle (crosspost / pin / close / reopen / language) +# +# These five async methods had no coverage at all — the sync twins are tested, +# the async ones never were. Added here because the UUID guard put a new line in +# each of them and Codecov (correctly) refused to let an untested line land. +# --------------------------------------------------------------------------- + +_POST_ID = "2a2579a2-c0db-486a-ba05-3ef7ea05fc3d" + + +class TestPostLifecycle: + @pytest.mark.asyncio + async def test_crosspost_posts_to_the_crosspost_path(self) -> None: + seen: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["method"] = request.method + seen["path"] = request.url.path + seen["body"] = json.loads(request.content) + return _json_response({"id": _POST_ID}) + + client = _make_client(handler) + await client.crosspost(_POST_ID, "general", title="Retitled") + assert seen["method"] == "POST" + assert seen["path"] == f"/api/v1/posts/{_POST_ID}/crosspost" + assert seen["body"]["colony_id"] == "general" + assert seen["body"]["title"] == "Retitled" + + @pytest.mark.asyncio + async def test_crosspost_omits_title_when_not_given(self) -> None: + seen: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["body"] = json.loads(request.content) + return _json_response({"id": _POST_ID}) + + client = _make_client(handler) + await client.crosspost(_POST_ID, "general") + assert "title" not in seen["body"] + + @pytest.mark.asyncio + @pytest.mark.parametrize( + ("method", "path"), + [("pin_post", "pin"), ("close_post", "close"), ("reopen_post", "reopen")], + ) + async def test_post_toggles(self, method: str, path: str) -> None: + seen: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["method"] = request.method + seen["path"] = request.url.path + return _json_response({"id": _POST_ID}) + + client = _make_client(handler) + await getattr(client, method)(_POST_ID) + assert seen["method"] == "POST" + assert seen["path"] == f"/api/v1/posts/{_POST_ID}/{path}" + + @pytest.mark.asyncio + async def test_set_post_language_puts_the_code_in_the_query(self) -> None: + seen: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["method"] = request.method + seen["url"] = str(request.url) + return _json_response({"post_id": _POST_ID, "language": "en"}) + + client = _make_client(handler) + result = await client.set_post_language(_POST_ID, "en") + assert seen["method"] == "PUT" + assert f"/posts/{_POST_ID}/language" in seen["url"] + assert "language=en" in seen["url"] + assert result["language"] == "en" + + @pytest.mark.asyncio + @pytest.mark.parametrize("method", ["crosspost", "pin_post", "close_post", "reopen_post", "set_post_language"]) + async def test_rejects_a_truncated_post_id(self, method: str) -> None: + """The guard fires before any request is built — a truncated id never leaves the process.""" + + def handler(request: httpx.Request) -> httpx.Response: # pragma: no cover - never called + raise AssertionError("request should not have been made") + + client = _make_client(handler) + args = {"crosspost": ("general",), "set_post_language": ("en",)}.get(method, ()) + with pytest.raises(ValueError, match="truncated UUID"): + await getattr(client, method)("2a2579a2", *args)