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_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) 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