From 117fd8325c946a835b28792fae4dfb766033439c Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 1 Nov 2026 14:41:45 +0000 Subject: [PATCH 1/2] Add the hibira Python SDK Client and AsyncClient for the Hibira HTTP API (trigger, batch, channel queries, terminate connections), channel and user auth, webhook verification, Hibira Push (interests, users, user tokens), approvals, and stream_to_channel / astream_to_channel for streaming LLM output to useStream() in @hibira-io/react and @hibira-io/vue. text_of reads OpenAI, Anthropic, OpenAI Agents SDK, LangChain/LangGraph and Gemini chunks. hibira.fastapi adds an auth router and a webhook dependency. Tested on Python 3.9 and 3.13, end to end against pusher-server (including channel history replay into the JS stream store). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01SHtsCCHwKcFDPRqugEWRfU --- .github/workflows/publish.yml | 43 ++++ .github/workflows/test.yml | 47 ++++ .gitignore | 10 + CHANGELOG.md | 12 + README.md | 266 +++++++++++++++++++ pyproject.toml | 71 +++++ src/hibira/__init__.py | 41 +++ src/hibira/_client.py | 472 ++++++++++++++++++++++++++++++++++ src/hibira/_errors.py | 67 +++++ src/hibira/_signing.py | 106 ++++++++ src/hibira/_version.py | 1 + src/hibira/fastapi.py | 128 +++++++++ src/hibira/py.typed | 0 src/hibira/stream.py | 419 ++++++++++++++++++++++++++++++ tests/test_client.py | 184 +++++++++++++ tests/test_e2e.py | 166 ++++++++++++ tests/test_fastapi.py | 71 +++++ tests/test_signing.py | 90 +++++++ tests/test_stream.py | 218 ++++++++++++++++ 19 files changed, 2412 insertions(+) create mode 100644 .github/workflows/publish.yml create mode 100644 .github/workflows/test.yml create mode 100644 .gitignore create mode 100644 CHANGELOG.md create mode 100644 README.md create mode 100644 pyproject.toml create mode 100644 src/hibira/__init__.py create mode 100644 src/hibira/_client.py create mode 100644 src/hibira/_errors.py create mode 100644 src/hibira/_signing.py create mode 100644 src/hibira/_version.py create mode 100644 src/hibira/fastapi.py create mode 100644 src/hibira/py.typed create mode 100644 src/hibira/stream.py create mode 100644 tests/test_client.py create mode 100644 tests/test_e2e.py create mode 100644 tests/test_fastapi.py create mode 100644 tests/test_signing.py create mode 100644 tests/test_stream.py diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml new file mode 100644 index 0000000..f53b178 --- /dev/null +++ b/.github/workflows/publish.yml @@ -0,0 +1,43 @@ +name: publish + +# Push a tag vX.Y.Z (matching src/hibira/_version.py) to publish to PyPI with +# trusted publishing (PyPI project "hibira", workflow publish.yml, +# environment pypi). +on: + push: + tags: ["v*"] + +jobs: + build: + runs-on: ubuntu-latest + permissions: + contents: read + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + - name: Check the tag matches the version + run: | + version=$(python -c "exec(open('src/hibira/_version.py').read()); print(__version__)") + test "v$version" = "$GITHUB_REF_NAME" || { echo "tag $GITHUB_REF_NAME does not match version $version"; exit 1; } + - run: pip install -e . pytest fastapi build + - run: pytest -q + - run: python -m build + - uses: actions/upload-artifact@v4 + with: + name: dist + path: dist/ + + publish: + needs: build + runs-on: ubuntu-latest + environment: pypi + permissions: + id-token: write + steps: + - uses: actions/download-artifact@v4 + with: + name: dist + path: dist/ + - uses: pypa/gh-action-pypi-publish@release/v1 diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml new file mode 100644 index 0000000..4bc6782 --- /dev/null +++ b/.github/workflows/test.yml @@ -0,0 +1,47 @@ +name: test + +on: + push: + branches: [main] + pull_request: + +permissions: + contents: read + +jobs: + lint: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + - run: pip install -e . fastapi ruff mypy + - run: ruff check . + - run: ruff format --check src + - run: mypy + + test: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python: ["3.9", "3.10", "3.11", "3.12", "3.13"] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python }} + - run: pip install -e . pytest fastapi + - run: pytest -q + + build: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + - run: pip install build twine + - run: python -m build + - run: twine check dist/* diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..86daf29 --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +.venv/ +__pycache__/ +*.pyc +dist/ +build/ +*.egg-info/ +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ +.env diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..7de2758 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,12 @@ +# Changelog + +## 0.1.0 + +First release. + +- `Client` and `AsyncClient`: `trigger`, `trigger_batch`, `channels`, `channel_info`, `users`, `terminate_user_connections`, settings from `HIBIRA_*` environment variables. +- Channel and user auth (`authenticate_channel`, `authenticate_user`) and webhook verification (`verify_webhook`). +- Hibira Push: `push_to_interests`, `push_to_users`, `push_token`, `delete_push_user`. +- Approvals: `request_approval`, `get_approval`, `wait_for_approval`. +- `stream_to_channel` / `astream_to_channel`: stream LLM output to `useStream()` in `@hibira-io/react` and `@hibira-io/vue`, with `text_of` for OpenAI, Anthropic, OpenAI Agents SDK, LangChain / LangGraph and Gemini chunks. +- `hibira.fastapi`: channel / user auth router and webhook dependency. diff --git a/README.md b/README.md new file mode 100644 index 0000000..71db114 --- /dev/null +++ b/README.md @@ -0,0 +1,266 @@ +# hibira + +Python server library for [Hibira](https://hibira.io), hosted realtime that +speaks the Pusher Channels protocol. Publish events, authorize private and +presence channels, verify webhooks, send push notifications, ask a person for +an approval, and stream LLM output to the browser as it is written. + +```sh +pip install hibira +``` + +Python 3.9+, one dependency ([httpx](https://www.python-httpx.org/)). Both a +regular and an asyncio client. pusher-js, Laravel Echo and the +`@hibira-io/*` front-end packages connect to Hibira unchanged. + +## Set up + +Create an app at [hibira.io/portal](https://hibira.io/portal) (the free plan +needs no card) and put its keys in the environment: + +```sh +HIBIRA_APP_ID=... +HIBIRA_APP_KEY=... +HIBIRA_APP_SECRET=... +``` + +```python +import hibira + +client = hibira.Client.from_env() # or hibira.Client(app_id, key, secret) +client.trigger("orders", "created", {"id": 1042}) +``` + +```python +async with hibira.AsyncClient.from_env() as client: + await client.trigger(["orders", "dashboard"], "created", {"id": 1042}) +``` + +| Setting | Environment | Default | +| --- | --- | --- | +| `app_id`, `key`, `secret` | `HIBIRA_APP_ID`, `HIBIRA_APP_KEY`, `HIBIRA_APP_SECRET` | | +| `host` | `HIBIRA_HOST` | `ws.hibira.io` (or the host shown on your app page) | +| `port`, `use_tls` | `HIBIRA_PORT`, `HIBIRA_USE_TLS` | `443`, `True` | +| `push_instance_id`, `push_secret_key` | `HIBIRA_PUSH_INSTANCE_ID`, `HIBIRA_PUSH_SECRET_KEY` | | +| `push_endpoint` | `HIBIRA_PUSH_ENDPOINT` | `https://` | +| `timeout` | | `10.0` seconds | + +Keyword arguments to `from_env()` win over the environment. Both clients +accept `http_client=` to reuse your own `httpx.Client` / `httpx.AsyncClient`. + +## Stream LLM output to the browser + +`stream_to_channel` sends a stream of text to a channel as it is produced, in +the format `useStream()` in [@hibira-io/react](https://github.com/hibira-io/react) +and [@hibira-io/vue](https://github.com/hibira-io/vue) puts back together. +Tokens are batched (about 10 events a second rather than one per token), each +event stays under the plan's message size, and the last event carries the +whole text when it fits, so people who open the page late catch up. With +channel history on (paid plans), `useStream(channel, { rewind: true })` +replays a stream after a reload. + +```python +from openai import OpenAI +import hibira + +client = hibira.Client.from_env() +openai = OpenAI() + +stream = openai.chat.completions.create(model="gpt-5-mini", messages=messages, stream=True) +result = hibira.stream_to_channel(client, f"private-conversation-{conversation_id}", stream, meta={"role": "assistant"}) +save_message(conversation_id, result.text) +``` + +```jsx +// React +const { text, done } = useStream(`private-conversation-${conversationId}`); +``` + +`text_of` reads the text out of chunks from the common SDKs, so you can pass +their streams as they are: + +| Source | Pass | +| --- | --- | +| OpenAI chat completions | `client.chat.completions.create(..., stream=True)` | +| OpenAI Responses | `client.responses.create(..., stream=True)` | +| Anthropic | `client.messages.create(..., stream=True)` (or the async client) | +| OpenAI Agents SDK | `Runner.run_streamed(agent, input).stream_events()` | +| LangGraph / LangChain | `graph.astream(input, stream_mode="messages")`, `llm.astream(...)` | +| Gemini (google-genai) | `client.models.generate_content_stream(...)` | +| Anything else | an iterable of `str` or `bytes`, or pass `pick=lambda chunk: ...` | + +With asyncio, use `astream_to_channel` (it takes async or plain iterables, +and an `AsyncClient` or a `Client`): + +```python +async def answer(conversation_id: str, question: str) -> str: + stream = graph.astream({"messages": [("user", question)]}, stream_mode="messages") + result = await hibira.astream_to_channel(client, f"private-conversation-{conversation_id}", stream) + return result.text +``` + +To stream only one LangGraph node, filter with `pick`: +`pick=lambda c: hibira.text_of(c) if c[1].get("langgraph_node") == "agent" else ""`. + +| Option | Default | | +| --- | --- | --- | +| `id` | random UUID | stream id, also in the result | +| `event` | `"ai-stream"` | event name | +| `interval` | `0.1` | seconds to gather tokens before sending them | +| `max_bytes` | `3500` | largest event payload (Free plans allow 4 KB per message) | +| `meta` | | JSON sent with the first event (model, message id, author) | +| `pick` | `text_of` | text of one chunk | +| `socket_id` | | leave out this connection (the sender's own tab) | +| `stop` | | a `threading.Event`; set it to end the stream with error `"aborted"` (sync only; cancel the task with asyncio) | + +If the source raises, the final event carries the error (useStream shows it) +and the exception is raised again. Both return a `StreamResult(id, text, events)`. + +## Authorize private and presence channels + +```python +client.authenticate_channel("private-user-42", socket_id) +# {"auth": "key:signature"} + +client.authenticate_channel("presence-room-7", socket_id, {"user_id": "42", "user_info": {"name": "Ann"}}) +# {"auth": "...", "channel_data": "{\"user_id\":\"42\",...}"} + +client.authenticate_user(socket_id, {"id": "42", "name": "Ann"}) # user sign-in +``` + +Return the dict as JSON from the endpoint pusher-js calls +(`channelAuthorization.endpoint`). It receives `socket_id` and +`channel_name` as form fields. These methods need no network and are plain +functions on `AsyncClient` too. + +### FastAPI + +```sh +pip install "hibira[fastapi]" +``` + +```python +from fastapi import Depends, FastAPI +from hibira import AsyncClient +from hibira.fastapi import auth_router, webhook + +client = AsyncClient.from_env() +app = FastAPI() + + +async def authorize(request, channel: str): + user = await current_user(request) + if channel == f"private-user-{user.id}": + return True + if channel.startswith("presence-room-"): + return {"user_id": str(user.id), "user_info": {"name": user.name}} + return False + + +app.include_router(auth_router(client, authorize)) # POST /hibira/auth + + +@app.post("/hibira/webhook") +async def on_webhook(payload: dict = Depends(webhook(client))): + for event in payload["events"]: + ... +``` + +`auth_router(client, authorize, user=...)` also answers user sign-in at +`/hibira/user-auth`. Django support is in +[django-hibira](https://github.com/hibira-io/django). + +## Webhooks + +```python +payload = client.verify_webhook( + request.headers.get("X-Pusher-Key"), + request.headers.get("X-Pusher-Signature"), + request.body, # the raw body, bytes or str +) +``` + +Raises `hibira.InvalidWebhook` when the key or signature does not match. + +## Channels API + +```python +client.trigger_batch([{"channel": "a", "name": "e", "data": {"x": 1}}]) +client.channels(prefix="presence-", info="user_count") +client.channel_info("presence-room-7", info=["user_count", "subscription_count"]) +client.users("presence-room-7") +client.terminate_user_connections("42") +``` + +`trigger` takes one channel or a list of up to 100. Data that is not a `str` +is sent as JSON. `socket_id=` leaves out the sender's own connection. + +## Push notifications + +Turn on Push for the app in the portal and set `HIBIRA_PUSH_INSTANCE_ID` and +`HIBIRA_PUSH_SECRET_KEY`. Bodies use the Beams format. + +```python +client.push_to_interests( + ["news"], + { + "web": {"notification": {"title": "New post", "deep_link": "https://example.com/p/1"}}, + "fcm": {"notification": {"title": "New post"}}, + "apns": {"aps": {"alert": {"title": "New post"}}}, + }, +) +client.push_to_users(["42"], {...}) +client.push_token("42") # {"token": ...} for setUserId in the web, Android and iOS SDKs +client.delete_push_user("42") +``` + +## Approvals + +Ask a person for a decision from a script, worker or agent. Hibira hosts the +answer page. + +```python +approval = client.request_approval( + "Refund order #1042 for $89?", + details="Customer says the parcel never arrived.", + options=["Refund", "Decline"], # the first one approves + push_interests=["ops"], # or push_users=[...], or share approval["url"] + channel="private-hibira-approvals.ops", # shows in the @hibira-io/approvals inbox +) +result = client.wait_for_approval(approval["approval_id"], timeout=3600) +if result["status"] == "approved": + refund(1042) +``` + +`status` is `pending`, `approved`, `rejected` or `expired`. `get_approval(id, wait=25)` +reads it once, waiting up to 25 seconds for an answer. + +## Errors + +Every error is a `hibira.HibiraError`. HTTP errors are `hibira.HTTPError` +(`status`, `message`, `body`), with subclasses `AuthenticationError` (401, +403), `NotFound` (404), `PayloadTooLarge` (413, the event is over the plan's +message size) and `RateLimited` (429). Missing settings raise `ConfigError`. + +## Development + +```sh +python -m venv .venv && .venv/bin/pip install -e . pytest fastapi ruff mypy +.venv/bin/pytest +.venv/bin/ruff check . && .venv/bin/mypy +``` + +`tests/test_e2e.py` runs against a real server when `HIBIRA_E2E=1` and the +`HIBIRA_*` settings point at it (it needs `pip install websockets`). + +## 中文摘要 + +`hibira` 是 Hibira 的 Python 伺服器端套件（Python 3.9 以上，同步與 asyncio 兩種 +client）：發送事件、私人 / presence 頻道授權、驗證 webhook、推播通知、人工核准， +以及用 `stream_to_channel` 把 LLM 的輸出即時串流到瀏覽器（React / Vue 的 +`useStream()` 直接接收）。支援 OpenAI、Anthropic、OpenAI Agents SDK、LangGraph、 +Gemini 的串流格式，並附 FastAPI 授權路由與 webhook 驗證。 + +## License + +MIT diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..34c666b --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,71 @@ +[build-system] +requires = ["hatchling>=1.24"] +build-backend = "hatchling.build" + +[project] +name = "hibira" +dynamic = ["version"] +description = "Hibira for Python: realtime events, channel auth, webhooks, push notifications, approvals and LLM streaming (Pusher-compatible)" +readme = "README.md" +license = "MIT" +license-files = ["LICENSE"] +requires-python = ">=3.9" +authors = [{ name = "Hibira", email = "dev@hibira.io" }] +keywords = ["hibira", "pusher", "websocket", "realtime", "channels", "push notifications", "llm", "streaming", "langgraph", "fastapi"] +classifiers = [ + "Development Status :: 4 - Beta", + "Framework :: AsyncIO", + "Framework :: FastAPI", + "Intended Audience :: Developers", + "Operating System :: OS Independent", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3 :: Only", + "Topic :: Internet :: WWW/HTTP", + "Topic :: Software Development :: Libraries", + "Typing :: Typed", +] +dependencies = ["httpx>=0.24,<1"] + +[project.optional-dependencies] +fastapi = ["fastapi>=0.100"] + +[project.urls] +Homepage = "https://hibira.io" +Documentation = "https://github.com/hibira-io/python#readme" +Source = "https://github.com/hibira-io/python" +Changelog = "https://github.com/hibira-io/python/blob/main/CHANGELOG.md" +Issues = "https://github.com/hibira-io/python/issues" + +[dependency-groups] +dev = ["pytest>=8", "fastapi>=0.100", "ruff>=0.6", "mypy>=1.10"] + +[tool.hatch.version] +path = "src/hibira/_version.py" + +[tool.hatch.build.targets.wheel] +packages = ["src/hibira"] + +[tool.hatch.build.targets.sdist] +include = ["src", "tests", "README.md", "CHANGELOG.md", "LICENSE"] + +[tool.pytest.ini_options] +testpaths = ["tests"] + +[tool.ruff] +line-length = 120 +target-version = "py39" + +[tool.ruff.lint] +select = ["E", "F", "W", "I", "B", "UP", "BLE"] +ignore = ["UP006", "UP007", "UP035", "UP045"] + +[tool.ruff.lint.flake8-bugbear] +extend-immutable-calls = ["fastapi.Depends"] + +[tool.ruff.lint.per-file-ignores] +"tests/*" = ["B008"] + +[tool.mypy] +python_version = "3.10" +strict = true +files = ["src/hibira"] diff --git a/src/hibira/__init__.py b/src/hibira/__init__.py new file mode 100644 index 0000000..248f55a --- /dev/null +++ b/src/hibira/__init__.py @@ -0,0 +1,41 @@ +"""Hibira for Python: publish realtime events, authorize private and presence +channels, verify webhooks, send push notifications, ask people for approvals +and stream LLM output to browsers. + +Hibira speaks the Pusher Channels protocol, so pusher-js, Laravel Echo and +the @hibira-io/* front-end packages connect to it unchanged. +""" + +from ._client import DEFAULT_HOST, AsyncClient, Client +from ._errors import ( + AuthenticationError, + ConfigError, + HibiraError, + HTTPError, + InvalidWebhook, + NotFound, + PayloadTooLarge, + RateLimited, +) +from ._version import __version__ +from .stream import STREAM_EVENT, StreamResult, astream_to_channel, stream_to_channel, text_of + +__all__ = [ + "DEFAULT_HOST", + "STREAM_EVENT", + "AsyncClient", + "AuthenticationError", + "Client", + "ConfigError", + "HTTPError", + "HibiraError", + "InvalidWebhook", + "NotFound", + "PayloadTooLarge", + "RateLimited", + "StreamResult", + "__version__", + "astream_to_channel", + "stream_to_channel", + "text_of", +] diff --git a/src/hibira/_client.py b/src/hibira/_client.py new file mode 100644 index 0000000..94cea22 --- /dev/null +++ b/src/hibira/_client.py @@ -0,0 +1,472 @@ +from __future__ import annotations + +import os +import time +from dataclasses import dataclass, field +from typing import Any, Iterable, Mapping, Sequence +from urllib.parse import quote, unquote + +import httpx + +from . import _signing +from ._errors import ConfigError, error_for +from ._version import __version__ + +DEFAULT_HOST = "ws.hibira.io" +MAX_TRIGGER_CHANNELS = 100 +MAX_APPROVAL_WAIT = 25 + +_USER_AGENT = f"hibira-python/{__version__}" + + +@dataclass +class _Request: + method: str + url: str + params: dict[str, str] = field(default_factory=dict) + content: bytes | None = None + headers: dict[str, str] = field(default_factory=dict) + + +def _env_bool(value: str | None) -> bool | None: + if value is None or value == "": + return None + return value.strip().lower() not in ("0", "false", "no", "off") + + +def _channels(channels: str | Iterable[str]) -> list[str]: + out = [channels] if isinstance(channels, str) else list(channels) + if not out: + raise ValueError("give at least one channel") + if len(out) > MAX_TRIGGER_CHANNELS: + raise ValueError(f"trigger takes at most {MAX_TRIGGER_CHANNELS} channels") + for ch in out: + _signing.check_channel(ch) + return out + + +def _data(data: Any) -> str: + return data if isinstance(data, str) else _signing.compact(data) + + +def _parse(response: httpx.Response) -> Any: + if response.status_code >= 400: + raise error_for(response.status_code, response.text) + if not response.content: + return {} + try: + return response.json() + except ValueError: + return {} + + +class _Base: + def __init__( + self, + app_id: str, + key: str, + secret: str, + host: str = DEFAULT_HOST, + port: int | None = None, + use_tls: bool = True, + timeout: float = 10.0, + push_instance_id: str | None = None, + push_secret_key: str | None = None, + push_endpoint: str | None = None, + ) -> None: + missing = [n for n, v in (("app_id", app_id), ("key", key), ("secret", secret)) if not v] + if missing: + raise ConfigError(f"missing {', '.join(missing)} (set HIBIRA_APP_ID, HIBIRA_APP_KEY, HIBIRA_APP_SECRET)") + self.app_id = str(app_id) + self.key = key + self.secret = secret + self.host = host or DEFAULT_HOST + self.use_tls = use_tls + self.port = port + self.timeout = timeout + scheme = "https" if use_tls else "http" + default_port = 443 if use_tls else 80 + self.base_url = f"{scheme}://{self.host}" + (f":{port}" if port and port != default_port else "") + self.push_instance_id = push_instance_id + self.push_secret_key = push_secret_key + self.push_endpoint = (push_endpoint or self.base_url).rstrip("/") + + @classmethod + def from_env(cls, **overrides: Any) -> Any: + """Settings from HIBIRA_* environment variables; keyword arguments win.""" + env = os.environ + port = env.get("HIBIRA_PORT") + values: dict[str, Any] = { + "app_id": env.get("HIBIRA_APP_ID", ""), + "key": env.get("HIBIRA_APP_KEY", ""), + "secret": env.get("HIBIRA_APP_SECRET", ""), + "host": env.get("HIBIRA_HOST") or DEFAULT_HOST, + "port": int(port) if port else None, + "use_tls": _env_bool(env.get("HIBIRA_USE_TLS")), + "push_instance_id": env.get("HIBIRA_PUSH_INSTANCE_ID") or None, + "push_secret_key": env.get("HIBIRA_PUSH_SECRET_KEY") or None, + "push_endpoint": env.get("HIBIRA_PUSH_ENDPOINT") or None, + } + if values["use_tls"] is None: + values["use_tls"] = True + values.update(overrides) + return cls(**values) + + def __repr__(self) -> str: + return f"{type(self).__name__}(app_id={self.app_id!r}, host={self.host!r})" + + # Signing (no network) + + def authenticate_channel( + self, channel: str, socket_id: str, user_data: Mapping[str, Any] | None = None + ) -> dict[str, str]: + """Authorization for a private- or presence- channel subscription. + + Return it as JSON from your auth endpoint. Presence channels need + user_data with a user_id (and optional user_info). + """ + return _signing.authenticate_channel(self.key, self.secret, channel, socket_id, user_data) + + def authenticate_user(self, socket_id: str, user_data: Mapping[str, Any]) -> dict[str, str]: + """Response for the client's user sign-in (pusher:signin); user_data needs an id.""" + return _signing.authenticate_user(self.key, self.secret, socket_id, user_data) + + def verify_webhook(self, key: str | None, signature: str | None, body: bytes | str) -> Any: + """Check a webhook's X-Pusher-Key and X-Pusher-Signature headers and return its parsed body. + + Pass the raw request body. Raises InvalidWebhook when they do not match. + """ + return _signing.verify_webhook(self.key, self.secret, key, signature, body) + + def push_token(self, user_id: str, ttl: int = 86400) -> dict[str, str]: + """A Hibira Push user token, for setUserId in the web, Android and iOS SDKs.""" + instance, secret = self._push_settings() + if not user_id or len(str(user_id)) > 164: + raise ValueError("user_id must be 1 to 164 characters") + return {"token": _signing.push_token(instance, secret, str(user_id), ttl)} + + # Request builders + + def _signed(self, method: str, path: str, params: Mapping[str, Any] | None = None, body: Any = None) -> _Request: + content = _signing.compact(body).encode() if body is not None else None + full = f"/apps/{quote(self.app_id, safe='')}{path}" + q = _signing.sign_query(self.key, self.secret, method, unquote(full), params, content) + headers = {"User-Agent": _USER_AGENT} + if content is not None: + headers["Content-Type"] = "application/json" + return _Request(method, self.base_url + full, q, content, headers) + + def _trigger_req(self, channels: str | Iterable[str], event: str, data: Any, socket_id: str | None) -> _Request: + if not event or len(event) > 200: + raise ValueError("event name must be 1 to 200 characters") + body: dict[str, Any] = {"name": event, "data": _data(data), "channels": _channels(channels)} + if socket_id: + _signing.check_socket_id(socket_id) + body["socket_id"] = socket_id + return self._signed("POST", "/events", body=body) + + def _batch_req(self, events: Sequence[Mapping[str, Any]]) -> _Request: + if not events: + raise ValueError("give at least one event") + batch = [] + for e in events: + _signing.check_channel(e["channel"]) + item = {"channel": e["channel"], "name": e["name"], "data": _data(e.get("data", ""))} + if e.get("socket_id"): + _signing.check_socket_id(e["socket_id"]) + item["socket_id"] = e["socket_id"] + if e.get("info"): + item["info"] = e["info"] + batch.append(item) + return self._signed("POST", "/batch_events", body={"batch": batch}) + + def _channels_req(self, prefix: str | None, info: str | Iterable[str] | None) -> _Request: + return self._signed("GET", "/channels", {"filter_by_prefix": prefix, "info": _info(info)}) + + def _channel_req(self, channel: str, info: str | Iterable[str] | None) -> _Request: + _signing.check_channel(channel) + return self._signed("GET", f"/channels/{channel}", {"info": _info(info)}) + + def _users_req(self, channel: str) -> _Request: + _signing.check_channel(channel) + if not channel.startswith("presence-"): + raise ValueError("users() works on presence- channels") + return self._signed("GET", f"/channels/{channel}/users") + + def _terminate_req(self, user_id: str) -> _Request: + if not user_id: + raise ValueError("user_id is required") + return self._signed("POST", f"/users/{quote(str(user_id), safe='')}/terminate_connections", body={}) + + def _approval_req(self, **fields: Any) -> _Request: + body = {k: v for k, v in fields.items() if v is not None} + if not body.get("title"): + raise ValueError("title is required") + return self._signed("POST", "/approvals", body=body) + + def _get_approval_req(self, approval_id: str, wait: int) -> _Request: + if not approval_id: + raise ValueError("approval_id is required") + if wait < 0 or wait > MAX_APPROVAL_WAIT: + raise ValueError(f"wait must be 0 to {MAX_APPROVAL_WAIT} seconds") + params = {"wait": wait} if wait else None + return self._signed("GET", f"/approvals/{quote(approval_id, safe='')}", params) + + def _push_settings(self) -> tuple[str, str]: + if not self.push_instance_id or not self.push_secret_key: + raise ConfigError( + "push needs push_instance_id and push_secret_key (HIBIRA_PUSH_INSTANCE_ID, HIBIRA_PUSH_SECRET_KEY)" + ) + return self.push_instance_id, self.push_secret_key + + def _push_req(self, method: str, path: str, body: Any = None) -> _Request: + instance, secret = self._push_settings() + headers = {"Authorization": f"Bearer {secret}", "User-Agent": _USER_AGENT} + content = None + if body is not None: + content = _signing.compact(body).encode() + headers["Content-Type"] = "application/json" + url = f"{self.push_endpoint}{path.format(instance=quote(instance, safe=''))}" + return _Request(method, url, {}, content, headers) + + def _push_interests_req(self, interests: Iterable[str], body: Mapping[str, Any]) -> _Request: + names = [interests] if isinstance(interests, str) else list(interests) + if not names: + raise ValueError("give at least one interest") + return self._push_req( + "POST", "/publish_api/v1/instances/{instance}/publishes/interests", {**body, "interests": names} + ) + + def _push_users_req(self, user_ids: Iterable[str], body: Mapping[str, Any]) -> _Request: + users = [user_ids] if isinstance(user_ids, str) else [str(u) for u in user_ids] + if not users: + raise ValueError("give at least one user id") + return self._push_req("POST", "/publish_api/v1/instances/{instance}/publishes/users", {**body, "users": users}) + + def _delete_push_user_req(self, user_id: str) -> _Request: + return self._push_req("DELETE", f"/customer_api/v1/instances/{{instance}}/users/{quote(str(user_id), safe='')}") + + +def _info(info: str | Iterable[str] | None) -> str | None: + if info is None or isinstance(info, str): + return info or None + return ",".join(info) or None + + +def _approval_fields( + title: str, + details: str | None, + options: Sequence[str] | None, + channel: str | None, + expires_in_minutes: int | None, + push_interests: Sequence[str] | None, + push_users: Sequence[str] | None, + agent: str | None, +) -> dict[str, Any]: + return { + "title": title, + "details": details, + "options": list(options) if options else None, + "channel": channel, + "expires_in_minutes": expires_in_minutes, + "push_interests": list(push_interests) if push_interests else None, + "push_users": list(push_users) if push_users else None, + "agent": agent, + } + + +class Client(_Base): + """Server client for a Hibira app: publish events, authorize channels, + verify webhooks, send push notifications and ask for approvals. + + >>> hibira = Client.from_env() + >>> hibira.trigger("orders", "created", {"id": 1042}) + """ + + def __init__( + self, app_id: str, key: str, secret: str, *, http_client: httpx.Client | None = None, **kwargs: Any + ) -> None: + super().__init__(app_id, key, secret, **kwargs) + self._own_http = http_client is None + self._http = http_client or httpx.Client(timeout=self.timeout) + + def _send(self, req: _Request, timeout: float | None = None) -> Any: + r = self._http.request( + req.method, + req.url, + params=req.params or None, + content=req.content, + headers=req.headers, + timeout=timeout if timeout is not None else self.timeout, + ) + return _parse(r) + + def close(self) -> None: + if self._own_http: + self._http.close() + + def __enter__(self) -> Client: + return self + + def __exit__(self, *exc: Any) -> None: + self.close() + + def trigger(self, channels: str | Iterable[str], event: str, data: Any, socket_id: str | None = None) -> Any: + """Publish an event to one channel or up to 100. Data that is not a str is sent as JSON. + + socket_id leaves out that connection (the sender's own tab). + """ + return self._send(self._trigger_req(channels, event, data, socket_id)) + + def trigger_batch(self, events: Sequence[Mapping[str, Any]]) -> Any: + """Publish several events in one request (10 at most by default). + + events: [{"channel", "name", "data", "socket_id"?}] + """ + return self._send(self._batch_req(events)) + + def channels(self, prefix: str | None = None, info: str | Iterable[str] | None = None) -> Any: + return self._send(self._channels_req(prefix, info)) + + def channel_info(self, channel: str, info: str | Iterable[str] | None = None) -> Any: + return self._send(self._channel_req(channel, info)) + + def users(self, channel: str) -> Any: + return self._send(self._users_req(channel)) + + def terminate_user_connections(self, user_id: str) -> Any: + return self._send(self._terminate_req(user_id)) + + def push_to_interests(self, interests: str | Iterable[str], body: Mapping[str, Any]) -> Any: + """Send a push notification to devices subscribed to any of interests. + + body is a Beams publish body: {"web": {...}, "fcm": {...}, "apns": {...}}. + """ + return self._send(self._push_interests_req(interests, body)) + + def push_to_users(self, user_ids: str | Iterable[str], body: Mapping[str, Any]) -> Any: + """Send a push notification to the devices of up to 1000 users.""" + return self._send(self._push_users_req(user_ids, body)) + + def delete_push_user(self, user_id: str) -> Any: + return self._send(self._delete_push_user_req(user_id)) + + def request_approval( + self, + title: str, + details: str | None = None, + options: Sequence[str] | None = None, + channel: str | None = None, + expires_in_minutes: int | None = None, + push_interests: Sequence[str] | None = None, + push_users: Sequence[str] | None = None, + agent: str | None = None, + ) -> Any: + """Ask a person for a decision; share the returned "url" with them.""" + fields = _approval_fields( + title, details, options, channel, expires_in_minutes, push_interests, push_users, agent + ) + return self._send(self._approval_req(**fields)) + + def get_approval(self, approval_id: str, wait: int = 0) -> Any: + """An approval's status; wait (up to 25 s) holds the request until it is answered.""" + return self._send(self._get_approval_req(approval_id, wait), timeout=self.timeout + wait) + + def wait_for_approval(self, approval_id: str, timeout: float = 3600) -> Any: + """Poll until the approval is no longer pending (or timeout seconds pass) and return it.""" + deadline = time.monotonic() + timeout + while True: + left = deadline - time.monotonic() + out = self.get_approval(approval_id, wait=max(0, min(MAX_APPROVAL_WAIT, int(left)))) + if out.get("status") != "pending" or time.monotonic() >= deadline: + return out + + +class AsyncClient(_Base): + """asyncio version of Client: the same methods, HTTP ones are awaitable. + + authenticate_channel, authenticate_user, verify_webhook and push_token + need no network and stay synchronous. + """ + + def __init__( + self, app_id: str, key: str, secret: str, *, http_client: httpx.AsyncClient | None = None, **kwargs: Any + ) -> None: + super().__init__(app_id, key, secret, **kwargs) + self._own_http = http_client is None + self._http = http_client or httpx.AsyncClient(timeout=self.timeout) + + async def _send(self, req: _Request, timeout: float | None = None) -> Any: + r = await self._http.request( + req.method, + req.url, + params=req.params or None, + content=req.content, + headers=req.headers, + timeout=timeout if timeout is not None else self.timeout, + ) + return _parse(r) + + async def aclose(self) -> None: + if self._own_http: + await self._http.aclose() + + async def __aenter__(self) -> AsyncClient: + return self + + async def __aexit__(self, *exc: Any) -> None: + await self.aclose() + + async def trigger(self, channels: str | Iterable[str], event: str, data: Any, socket_id: str | None = None) -> Any: + return await self._send(self._trigger_req(channels, event, data, socket_id)) + + async def trigger_batch(self, events: Sequence[Mapping[str, Any]]) -> Any: + return await self._send(self._batch_req(events)) + + async def channels(self, prefix: str | None = None, info: str | Iterable[str] | None = None) -> Any: + return await self._send(self._channels_req(prefix, info)) + + async def channel_info(self, channel: str, info: str | Iterable[str] | None = None) -> Any: + return await self._send(self._channel_req(channel, info)) + + async def users(self, channel: str) -> Any: + return await self._send(self._users_req(channel)) + + async def terminate_user_connections(self, user_id: str) -> Any: + return await self._send(self._terminate_req(user_id)) + + async def push_to_interests(self, interests: str | Iterable[str], body: Mapping[str, Any]) -> Any: + return await self._send(self._push_interests_req(interests, body)) + + async def push_to_users(self, user_ids: str | Iterable[str], body: Mapping[str, Any]) -> Any: + return await self._send(self._push_users_req(user_ids, body)) + + async def delete_push_user(self, user_id: str) -> Any: + return await self._send(self._delete_push_user_req(user_id)) + + async def request_approval( + self, + title: str, + details: str | None = None, + options: Sequence[str] | None = None, + channel: str | None = None, + expires_in_minutes: int | None = None, + push_interests: Sequence[str] | None = None, + push_users: Sequence[str] | None = None, + agent: str | None = None, + ) -> Any: + fields = _approval_fields( + title, details, options, channel, expires_in_minutes, push_interests, push_users, agent + ) + return await self._send(self._approval_req(**fields)) + + async def get_approval(self, approval_id: str, wait: int = 0) -> Any: + return await self._send(self._get_approval_req(approval_id, wait), timeout=self.timeout + wait) + + async def wait_for_approval(self, approval_id: str, timeout: float = 3600) -> Any: + deadline = time.monotonic() + timeout + while True: + left = deadline - time.monotonic() + out = await self.get_approval(approval_id, wait=max(0, min(MAX_APPROVAL_WAIT, int(left)))) + if out.get("status") != "pending" or time.monotonic() >= deadline: + return out diff --git a/src/hibira/_errors.py b/src/hibira/_errors.py new file mode 100644 index 0000000..2725c0c --- /dev/null +++ b/src/hibira/_errors.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +import json +from typing import Any + + +class HibiraError(Exception): + """Base class of every error this package raises.""" + + +class ConfigError(HibiraError): + """A setting is missing or invalid (app id, key, secret, push instance...).""" + + +class InvalidWebhook(HibiraError): + """A webhook's key or signature does not match this app.""" + + +class HTTPError(HibiraError): + """The Hibira HTTP API answered with an error status.""" + + def __init__(self, status: int, message: str, body: Any = None) -> None: + super().__init__(f"{status}: {message}") + self.status = status + self.message = message + self.body = body + + +class AuthenticationError(HTTPError): + """401 or 403: bad key, secret or signature, or a suspended account.""" + + +class NotFound(HTTPError): + """404: unknown app, channel, approval or push instance.""" + + +class PayloadTooLarge(HTTPError): + """413: the event is over the plan's message size (4 KB on Free).""" + + +class RateLimited(HTTPError): + """429: too many requests or messages for the plan.""" + + +def error_for(status: int, text: str) -> HTTPError: + body: Any = text + message = text.strip() or "request failed" + try: + body = json.loads(text) + except ValueError: + pass + else: + if isinstance(body, dict): + for k in ("error", "message", "description"): + if isinstance(body.get(k), str): + message = body[k] + break + cls: type[HTTPError] = HTTPError + if status in (401, 403): + cls = AuthenticationError + elif status == 404: + cls = NotFound + elif status == 413: + cls = PayloadTooLarge + elif status == 429: + cls = RateLimited + return cls(status, message, body) diff --git a/src/hibira/_signing.py b/src/hibira/_signing.py new file mode 100644 index 0000000..8083755 --- /dev/null +++ b/src/hibira/_signing.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import re +import time +from typing import Any, Mapping + +from ._errors import InvalidWebhook + +_SOCKET_ID = re.compile(r"\A\d+\.\d+\Z") +_CHANNEL = re.compile(r"\A[-a-zA-Z0-9_=@,.;]+\Z") + + +def compact(value: Any) -> str: + return json.dumps(value, separators=(",", ":"), ensure_ascii=False) + + +def hmac_hex(secret: str, message: str | bytes) -> str: + if isinstance(message, str): + message = message.encode() + return hmac.new(secret.encode(), message, hashlib.sha256).hexdigest() + + +def check_socket_id(socket_id: str) -> None: + if not isinstance(socket_id, str) or not _SOCKET_ID.match(socket_id): + raise ValueError(f"invalid socket_id: {socket_id!r}") + + +def check_channel(channel: str) -> None: + if not isinstance(channel, str) or not _CHANNEL.match(channel) or len(channel) > 200: + raise ValueError(f"invalid channel name: {channel!r}") + + +def sign_query( + key: str, + secret: str, + method: str, + path: str, + params: Mapping[str, Any] | None = None, + body: bytes | None = None, + now: float | None = None, +) -> dict[str, str]: + """Query parameters of a signed HTTP API request (Pusher auth_version 1.0).""" + q = {k.lower(): str(v) for k, v in (params or {}).items() if v is not None} + q["auth_key"] = key + q["auth_timestamp"] = str(int(now if now is not None else time.time())) + q["auth_version"] = "1.0" + if body: + q["body_md5"] = hashlib.md5(body).hexdigest() + string = "&".join(f"{k}={q[k]}" for k in sorted(q)) + q["auth_signature"] = hmac_hex(secret, f"{method.upper()}\n{path}\n{string}") + return q + + +def authenticate_channel( + key: str, secret: str, channel: str, socket_id: str, user_data: Mapping[str, Any] | None = None +) -> dict[str, str]: + check_channel(channel) + check_socket_id(socket_id) + if channel.startswith("private-encrypted-"): + raise ValueError("encrypted channels are not supported") + if channel.startswith("presence-"): + if not user_data or user_data.get("user_id") in (None, ""): + raise ValueError("presence channels need user_data with a user_id") + channel_data = compact(dict(user_data)) + sig = hmac_hex(secret, f"{socket_id}:{channel}:{channel_data}") + return {"auth": f"{key}:{sig}", "channel_data": channel_data} + if not channel.startswith("private-"): + raise ValueError("only private- and presence- channels need authorization") + return {"auth": f"{key}:{hmac_hex(secret, f'{socket_id}:{channel}')}"} + + +def authenticate_user(key: str, secret: str, socket_id: str, user_data: Mapping[str, Any]) -> dict[str, str]: + check_socket_id(socket_id) + if not user_data or user_data.get("id") in (None, ""): + raise ValueError("user_data needs an id") + encoded = compact(dict(user_data)) + return {"auth": f"{key}:{hmac_hex(secret, f'{socket_id}::user::{encoded}')}", "user_data": encoded} + + +def verify_webhook(key: str, secret: str, got_key: str | None, signature: str | None, body: bytes | str) -> Any: + raw = body.encode() if isinstance(body, str) else bytes(body) + if not got_key or not hmac.compare_digest(got_key, key): + raise InvalidWebhook("webhook key does not match this app") + if not signature or not hmac.compare_digest(signature.lower(), hmac_hex(secret, raw)): + raise InvalidWebhook("webhook signature does not match") + try: + return json.loads(raw) + except ValueError as err: + raise InvalidWebhook("webhook body is not JSON") from err + + +def _b64(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode() + + +def push_token(instance_id: str, secret_key: str, user_id: str, ttl: int, now: float | None = None) -> str: + t = int(now if now is not None else time.time()) + header = _b64(compact({"alg": "HS256", "typ": "JWT"}).encode()) + claims = {"sub": user_id, "iss": f"https://{instance_id}.pushnotifications.pusher.com", "iat": t, "exp": t + ttl} + payload = _b64(compact(claims).encode()) + sig = hmac.new(secret_key.encode(), f"{header}.{payload}".encode(), hashlib.sha256).digest() + return f"{header}.{payload}.{_b64(sig)}" diff --git a/src/hibira/_version.py b/src/hibira/_version.py new file mode 100644 index 0000000..3dc1f76 --- /dev/null +++ b/src/hibira/_version.py @@ -0,0 +1 @@ +__version__ = "0.1.0" diff --git a/src/hibira/fastapi.py b/src/hibira/fastapi.py new file mode 100644 index 0000000..97e04cf --- /dev/null +++ b/src/hibira/fastapi.py @@ -0,0 +1,128 @@ +"""FastAPI (and Starlette) helpers: the channel and user auth endpoints +pusher-js calls, and a dependency that verifies webhooks. + + from hibira import AsyncClient + from hibira.fastapi import auth_router, webhook + + hibira = AsyncClient.from_env() + + async def authorize(request, channel): + user = await current_user(request) + if channel == f"private-user-{user.id}": + return True + if channel.startswith("presence-room-"): + return {"user_id": user.id, "user_info": {"name": user.name}} + return False + + app.include_router(auth_router(hibira, authorize)) + + @app.post("/hibira/webhook") + async def on_webhook(payload: dict = Depends(webhook(hibira))): + ... +""" + +from __future__ import annotations + +import inspect +import json +from typing import Any, Awaitable, Callable, Mapping, Optional, Union +from urllib.parse import parse_qs + +try: + from fastapi import APIRouter, HTTPException, Request + from fastapi.responses import JSONResponse +except ImportError as err: # pragma: no cover + raise ImportError('hibira.fastapi needs FastAPI: pip install "hibira[fastapi]"') from err + +from ._client import _Base +from ._errors import InvalidWebhook + +Authorize = Callable[[Request, str], Union[Any, Awaitable[Any]]] +Identify = Callable[[Request], Union[Optional[Mapping[str, Any]], Awaitable[Optional[Mapping[str, Any]]]]] + + +async def _maybe(value: Any) -> Any: + return await value if inspect.isawaitable(value) else value + + +async def _form(request: Request) -> dict[str, str]: + raw = await request.body() + if request.headers.get("content-type", "").startswith("application/json"): + try: + data = json.loads(raw or b"{}") + except ValueError: + data = {} + return {k: str(v) for k, v in data.items()} if isinstance(data, dict) else {} + return {k: v[0] for k, v in parse_qs(raw.decode("utf-8", "replace")).items()} + + +def auth_router( + client: _Base, + authorize: Authorize, + *, + path: str = "/hibira/auth", + user: Optional[Identify] = None, + user_path: str = "/hibira/user-auth", +) -> APIRouter: + """Router with the endpoint pusher-js calls to authorize private- and presence- channels. + + authorize(request, channel) (sync or async) returns False/None to refuse, + True to allow a private channel, or the member's user_data + ({"user_id", "user_info"?}) for a presence channel. Point the client's + channelAuthorization.endpoint at path. + + With user (request -> {"id", ...} or None), the router also answers user + sign-in at user_path (userAuthentication.endpoint). + """ + router = APIRouter() + + @router.post(path, include_in_schema=False) + async def channel_auth(request: Request) -> JSONResponse: + form = await _form(request) + socket_id, channel = form.get("socket_id", ""), form.get("channel_name", "") + if not socket_id or not channel: + raise HTTPException(400, "socket_id and channel_name are required") + verdict = await _maybe(authorize(request, channel)) + if not verdict: + raise HTTPException(403, "forbidden") + user_data = verdict if isinstance(verdict, Mapping) else None + if channel.startswith("presence-") and user_data is None: + raise HTTPException(500, "authorize must return user_data with a user_id for presence channels") + try: + return JSONResponse(client.authenticate_channel(channel, socket_id, user_data)) + except ValueError as err: + raise HTTPException(400, str(err)) from err + + if user is not None: + + @router.post(user_path, include_in_schema=False) + async def user_auth(request: Request) -> JSONResponse: + form = await _form(request) + socket_id = form.get("socket_id", "") + if not socket_id: + raise HTTPException(400, "socket_id is required") + data = await _maybe(user(request)) + if not data: + raise HTTPException(403, "forbidden") + try: + return JSONResponse(client.authenticate_user(socket_id, data)) + except ValueError as err: + raise HTTPException(400, str(err)) from err + + return router + + +def webhook(client: _Base) -> Callable[[Request], Awaitable[Any]]: + """Dependency that verifies a Hibira webhook and returns its parsed body + ({"time_ms", "events": [...]}); a bad signature answers 401.""" + + async def verified(request: Request) -> Any: + body = await request.body() + try: + return client.verify_webhook( + request.headers.get("x-pusher-key"), request.headers.get("x-pusher-signature"), body + ) + except InvalidWebhook as err: + raise HTTPException(401, str(err)) from err + + return verified diff --git a/src/hibira/py.typed b/src/hibira/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/src/hibira/stream.py b/src/hibira/stream.py new file mode 100644 index 0000000..c284ce4 --- /dev/null +++ b/src/hibira/stream.py @@ -0,0 +1,419 @@ +"""Server side of the Hibira stream protocol: send an LLM's output (or any +stream of text) to a channel as it is produced. The useStream() hook in +@hibira-io/react and composable in @hibira-io/vue put it back together. + +Same wire format as streamToChannel in those packages: event "ai-stream", +payloads {id, seq, delta, meta (first event only)}, then a final +{id, seq, done: true, text?, error?}. Tokens are batched (one event per +interval rather than one per token), each event stays under max_bytes (Free +plans allow 4 KB per message) and events are sent one after another, in order. +""" + +from __future__ import annotations + +import asyncio +import codecs +import inspect +import json +import threading +import uuid +from dataclasses import dataclass +from typing import Any, AsyncIterable, Callable, Iterable, Optional, Union + +STREAM_EVENT = "ai-stream" + +_MISSING = object() + + +@dataclass(frozen=True) +class StreamResult: + id: str + text: str + events: int + + +def _get(obj: Any, name: str) -> Any: + if isinstance(obj, dict): + return obj.get(name) + return getattr(obj, name, None) + + +def _content_text(content: Any) -> str: + if isinstance(content, str): + return content + if isinstance(content, list): + out = [] + for block in content: + if isinstance(block, str): + out.append(block) + elif _get(block, "type") in ("text", "text_delta", "output_text") and isinstance(_get(block, "text"), str): + out.append(_get(block, "text")) + return "".join(out) + return "" + + +def text_of(chunk: Any, decoder: Optional[codecs.IncrementalDecoder] = None) -> str: + """Text of one chunk from the common LLM SDK streams, or "" when it has none. + + Handles str and bytes, Anthropic (content_block_delta events), OpenAI chat + completions (choices[0].delta.content) and responses + (response.output_text.delta), the OpenAI Agents SDK (raw_response_event), + LangChain / LangGraph message chunks (stream_mode="messages", also as + (mode, chunk) tuples), Gemini (chunk.text) and dicts of the same shapes. + """ + if chunk is None: + return "" + if isinstance(chunk, str): + return chunk + if isinstance(chunk, (bytes, bytearray, memoryview)): + if decoder is None: + return bytes(chunk).decode("utf-8", errors="replace") + return decoder.decode(bytes(chunk)) + if isinstance(chunk, tuple) and chunk: + if len(chunk) >= 2 and isinstance(chunk[0], str): + return text_of(chunk[-1], decoder) + return text_of(chunk[0], decoder) + kind = _get(chunk, "type") + if kind == "content_block_delta": + delta = _get(chunk, "delta") + return (_get(delta, "text") or "") if _get(delta, "type") == "text_delta" else "" + if kind == "response.output_text.delta": + return _get(chunk, "delta") or "" + if kind == "raw_response_event": + return text_of(_get(chunk, "data"), decoder) + if kind == "text-delta": + for name in ("text", "textDelta", "delta"): + v = _get(chunk, name) + if isinstance(v, str): + return v + return "" + if kind in ("AIMessageChunk", "ai"): + return _content_text(_get(chunk, "content")) + choices = _get(chunk, "choices") + if isinstance(choices, list): + if not choices: + return "" + return _get(_get(choices[0], "delta"), "content") or "" + if kind is None or isinstance(kind, str): + try: + text = _get(chunk, "text") + except Exception: # noqa: BLE001 + return "" + if isinstance(text, str): + return text + return "" + + +def _size(s: str) -> int: + return len(json.dumps(s, ensure_ascii=False).encode()) - 2 + + +def _bytes(s: str) -> int: + return len(s.encode()) + + +def split(text: str, budget: int) -> list[str]: + """Split text so each piece's JSON-escaped UTF-8 size fits budget, never inside a character.""" + out: list[str] = [] + piece: list[str] = [] + size = 0 + for ch in text: + n = _size(ch) + if size + n > budget and piece: + out.append("".join(piece)) + piece, size = [], 0 + piece.append(ch) + size += n + if piece: + out.append("".join(piece)) + return out + + +def _payload_bytes(payload: dict[str, Any]) -> int: + return len(json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode()) + + +class _Encoder: + def __init__(self, id: str, meta: Any, max_bytes: int) -> None: + self.id = id + self.meta = meta + self.max_bytes = max_bytes + envelope = {"id": id, "seq": 1_000_000_000, "delta": ""} + if meta is not _MISSING: + envelope["meta"] = meta + self.budget = max_bytes - _payload_bytes(envelope) + if self.budget < 16: + raise ValueError("stream_to_channel: max_bytes is too small for the id and meta") + self.seq = 0 + + def deltas(self, buffer: str) -> list[dict[str, Any]]: + out = [] + for delta in split(buffer, self.budget): + payload: dict[str, Any] = {"id": self.id, "seq": self.seq, "delta": delta} + if self.seq == 0 and self.meta is not _MISSING: + payload["meta"] = self.meta + self.seq += 1 + out.append(payload) + return out + + def done(self, text: str, error: Optional[str]) -> dict[str, Any]: + payload: dict[str, Any] = {"id": self.id, "seq": self.seq, "done": True} + if self.seq == 0 and self.meta is not _MISSING: + payload["meta"] = self.meta + if error is not None: + payload["error"] = error + if _payload_bytes({**payload, "text": text}) <= self.max_bytes: + payload["text"] = text + self.seq += 1 + return payload + + +def _error_text(err: BaseException) -> str: + if isinstance(err, (asyncio.CancelledError, KeyboardInterrupt)): + return "aborted" + return str(err) or type(err).__name__ + + +def stream_to_channel( + client: Any, + channel: str, + source: Iterable[Any], + *, + id: Optional[str] = None, + event: str = STREAM_EVENT, + interval: float = 0.1, + max_bytes: int = 3500, + meta: Any = _MISSING, + pick: Optional[Callable[[Any], Optional[str]]] = None, + socket_id: Optional[str] = None, + stop: Optional[threading.Event] = None, +) -> StreamResult: + """Send a stream of text to a channel as events useStream() puts back together. + + client is a hibira.Client (anything with trigger(channel, event, data, + socket_id=None) works). source yields text chunks or LLM SDK chunks (see + text_of; pass pick for other shapes). meta is sent with the first event. + Setting stop ends the stream with error "aborted". Returns after the last + event is sent; if source raises, the final event carries the error and it + is raised again. + """ + enc = _Encoder(id or str(uuid.uuid4()), meta, max_bytes) + decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") + lock = threading.Lock() + send_lock = threading.Lock() + state: dict[str, Any] = {"buffer": "", "timer": None, "send_error": None} + + def send(payloads: list[dict[str, Any]]) -> None: + with send_lock: + for p in payloads: + if state["send_error"] is not None: + return + try: + client.trigger(channel, event, p, socket_id=socket_id) + except Exception as err: # noqa: BLE001 + state["send_error"] = err + + def flush() -> None: + with lock: + timer = state["timer"] + if timer is not None: + timer.cancel() + state["timer"] = None + buffer, state["buffer"] = state["buffer"], "" + if not buffer: + return + # Taking send_lock before releasing lock keeps events in seq order. + send_lock.acquire() + try: + payloads = enc.deltas(buffer) + except BaseException: + send_lock.release() + raise + try: + for p in payloads: + if state["send_error"] is not None: + break + try: + client.trigger(channel, event, p, socket_id=socket_id) + except Exception as err: # noqa: BLE001 + state["send_error"] = err + finally: + send_lock.release() + + text_parts: list[str] = [] + error: Optional[BaseException] = None + try: + for chunk in source: + if stop is not None and stop.is_set(): + raise _Aborted() + if state["send_error"] is not None: + raise state["send_error"] + delta = (pick(chunk) or "") if pick else text_of(chunk, decoder) + if not delta: + continue + text_parts.append(delta) + with lock: + state["buffer"] += delta + full = _bytes(state["buffer"]) >= enc.budget + if not full and state["timer"] is None: + t = threading.Timer(interval, flush) + t.daemon = True + state["timer"] = t + t.start() + if full: + flush() + tail = decoder.decode(b"", final=True) + if tail: + text_parts.append(tail) + with lock: + state["buffer"] += tail + if stop is not None and stop.is_set(): + raise _Aborted() + except BaseException as err: # noqa: BLE001 + error = err + flush() + + text = "".join(text_parts) + message = None + if error is not None: + message = "aborted" if isinstance(error, _Aborted) else _error_text(error) + send([enc.done(text, message)]) + + if error is not None and not isinstance(error, _Aborted): + raise error + if state["send_error"] is not None: + raise state["send_error"] + return StreamResult(enc.id, text, enc.seq) + + +class _Aborted(Exception): + pass + + +async def _aiter(source: Union[AsyncIterable[Any], Iterable[Any]]) -> Any: + if hasattr(source, "__aiter__"): + async for item in source: + yield item + else: + for item in source: + yield item + + +async def astream_to_channel( + client: Any, + channel: str, + source: Union[AsyncIterable[Any], Iterable[Any]], + *, + id: Optional[str] = None, + event: str = STREAM_EVENT, + interval: float = 0.1, + max_bytes: int = 3500, + meta: Any = _MISSING, + pick: Optional[Callable[[Any], Optional[str]]] = None, + socket_id: Optional[str] = None, +) -> StreamResult: + """asyncio version of stream_to_channel, for a hibira.AsyncClient (or a + sync Client, whose trigger then runs in a thread) and an async iterable + such as an OpenAI / Anthropic async stream or graph.astream(...). + + Cancelling the task sends the final event with error "aborted" and + re-raises CancelledError. + """ + enc = _Encoder(id or str(uuid.uuid4()), meta, max_bytes) + decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") + send_lock = asyncio.Lock() + buffer = "" + timer: Optional[asyncio.TimerHandle] = None + pending: set[asyncio.Task[None]] = set() + send_error: Optional[BaseException] = None + loop = asyncio.get_running_loop() + + async def trigger(payload: dict[str, Any]) -> None: + result = client.trigger(channel, event, payload, socket_id=socket_id) + if inspect.isawaitable(result): + await result + + async def send(payloads: list[dict[str, Any]]) -> None: + nonlocal send_error + async with send_lock: + for p in payloads: + if send_error is not None: + return + try: + await trigger(p) + except Exception as err: # noqa: BLE001 + send_error = err + + def flush() -> None: + nonlocal buffer, timer + if timer is not None: + timer.cancel() + timer = None + if not buffer: + return + payloads = enc.deltas(buffer) + buffer = "" + task = loop.create_task(send(payloads)) + pending.add(task) + task.add_done_callback(pending.discard) + + if not _is_async_client(client): + sync_client = client + + class _Threaded: + def trigger(self, *a: Any, **kw: Any) -> Any: + return asyncio.to_thread(sync_client.trigger, *a, **kw) + + client = _Threaded() + + text_parts: list[str] = [] + error: Optional[BaseException] = None + try: + async for chunk in _aiter(source): + if send_error is not None: + raise send_error + delta = (pick(chunk) or "") if pick else text_of(chunk, decoder) + if not delta: + continue + text_parts.append(delta) + buffer += delta + if _bytes(buffer) >= enc.budget: + flush() + elif timer is None: + timer = loop.call_later(interval, flush) + tail = decoder.decode(b"", final=True) + if tail: + text_parts.append(tail) + buffer += tail + except BaseException as err: # noqa: BLE001 + error = err + flush() + + text = "".join(text_parts) + message = _error_text(error) if error is not None else None + final = enc.done(text, message) + + async def finish() -> None: + # Tasks run in creation order, but each waits on send_lock (FIFO), so + # the final event goes last. + await send([final]) + if pending: + await asyncio.gather(*list(pending), return_exceptions=True) + + finisher = asyncio.ensure_future(finish()) + try: + await asyncio.shield(finisher) + except asyncio.CancelledError: + await finisher + raise + + if error is not None: + raise error + if send_error is not None: + raise send_error + return StreamResult(enc.id, text, enc.seq) + + +def _is_async_client(client: Any) -> bool: + trigger = getattr(client, "trigger", None) + return inspect.iscoroutinefunction(trigger) diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..aa2af35 --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,184 @@ +import asyncio +import json +from urllib.parse import parse_qs, urlsplit + +import httpx +import pytest + +import hibira +from hibira import _signing + + +def recorder(status=200, body=b"{}"): + seen = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + return httpx.Response(status, content=body, headers={"content-type": "application/json"}) + + return seen, handler + + +def make(handler, **kw): + return hibira.Client( + "app1", "key1", "secret1", http_client=httpx.Client(transport=httpx.MockTransport(handler)), **kw + ) + + +def check_signature(req: httpx.Request): + q = {k: v[0] for k, v in parse_qs(urlsplit(str(req.url)).query).items()} + sig = q.pop("auth_signature") + signed = _signing.sign_query( + "key1", + "secret1", + req.method, + req.url.path, + {k: v for k, v in q.items() if not k.startswith("auth_") and k != "body_md5"}, + req.content or None, + now=int(q["auth_timestamp"]), + ) + assert signed["auth_signature"] == sig + return q + + +def test_trigger(): + seen, handler = recorder() + make(handler).trigger(["a", "b"], "created", {"id": 1}, socket_id="1.2") + req = seen[0] + assert req.method == "POST" + assert str(req.url).startswith("https://ws.hibira.io/apps/app1/events?") + assert json.loads(req.content) == { + "name": "created", + "data": '{"id":1}', + "channels": ["a", "b"], + "socket_id": "1.2", + } + check_signature(req) + + +def test_trigger_string_data_and_validation(): + seen, handler = recorder() + c = make(handler) + c.trigger("a", "e", "raw") + assert json.loads(seen[0].content)["data"] == "raw" + with pytest.raises(ValueError): + c.trigger([], "e", 1) + with pytest.raises(ValueError): + c.trigger([f"c{i}" for i in range(101)], "e", 1) + with pytest.raises(ValueError): + c.trigger("a", "e", 1, socket_id="bad") + + +def test_batch_and_queries(): + seen, handler = recorder(body=b'{"channels":{}}') + c = make(handler, host="localhost", port=8080, use_tls=False) + c.trigger_batch([{"channel": "a", "name": "e", "data": {"x": 1}}]) + assert json.loads(seen[0].content) == {"batch": [{"channel": "a", "name": "e", "data": '{"x":1}'}]} + assert c.channels(prefix="presence-", info=["user_count"]) == {"channels": {}} + q = check_signature(seen[1]) + assert q["filter_by_prefix"] == "presence-" and q["info"] == "user_count" + assert str(seen[1].url).startswith("http://localhost:8080/apps/app1/channels?") + c.channel_info("presence-x", info="user_count,subscription_count") + assert seen[2].url.path == "/apps/app1/channels/presence-x" + c.users("presence-x") + assert seen[3].url.path == "/apps/app1/channels/presence-x/users" + with pytest.raises(ValueError): + c.users("private-x") + c.terminate_user_connections("u 1") + assert seen[4].url.raw_path.startswith(b"/apps/app1/users/u%201/terminate_connections") + for r in seen: + check_signature(r) + + +@pytest.mark.parametrize( + ("status", "cls"), + [ + (413, hibira.PayloadTooLarge), + (401, hibira.AuthenticationError), + (403, hibira.AuthenticationError), + (404, hibira.NotFound), + (429, hibira.RateLimited), + (500, hibira.HTTPError), + ], +) +def test_errors(status, cls): + _, handler = recorder(status, b'{"error":"nope"}') + with pytest.raises(cls) as info: + make(handler).trigger("a", "e", 1) + assert info.value.status == status + assert info.value.message == "nope" + assert isinstance(info.value, hibira.HibiraError) + + +def test_push(): + seen, handler = recorder(body=b'{"publishId":"p1"}') + c = make(handler, push_instance_id="inst", push_secret_key="sk") + assert c.push_to_interests("news", {"web": {"notification": {"title": "Hi"}}}) == {"publishId": "p1"} + assert seen[0].url == "https://ws.hibira.io/publish_api/v1/instances/inst/publishes/interests" + assert seen[0].headers["authorization"] == "Bearer sk" + assert json.loads(seen[0].content)["interests"] == ["news"] + c.push_to_users(["u1", 2], {"fcm": {}}) + assert json.loads(seen[1].content)["users"] == ["u1", "2"] + c.delete_push_user("u1") + assert seen[2].method == "DELETE" and seen[2].url.path == "/customer_api/v1/instances/inst/users/u1" + with pytest.raises(hibira.ConfigError): + make(handler).push_to_users("u1", {}) + + +def test_approvals(): + answers = iter( + [ + {"approval_id": "apr_1", "status": "pending", "url": "https://hibira.io/approve/apr_1/t"}, + {"approval_id": "apr_1", "status": "pending"}, + {"approval_id": "apr_1", "status": "approved", "choice": "Refund"}, + ] + ) + seen = [] + + def handler(req): + seen.append(req) + return httpx.Response(201 if req.method == "POST" else 200, json=next(answers)) + + c = make(handler) + out = c.request_approval("Refund #1042?", options=["Refund", "Decline"], push_interests=["ops"]) + assert out["url"].startswith("https://hibira.io/approve/") + assert json.loads(seen[0].content) == { + "title": "Refund #1042?", + "options": ["Refund", "Decline"], + "push_interests": ["ops"], + } + assert c.wait_for_approval("apr_1", timeout=60)["choice"] == "Refund" + assert seen[1].url.path == "/apps/app1/approvals/apr_1" + assert check_signature(seen[1])["wait"] == "25" + with pytest.raises(ValueError): + c.get_approval("apr_1", wait=26) + + +def test_from_env(monkeypatch): + monkeypatch.setenv("HIBIRA_APP_ID", "1") + monkeypatch.setenv("HIBIRA_APP_KEY", "k") + monkeypatch.setenv("HIBIRA_APP_SECRET", "s") + monkeypatch.setenv("HIBIRA_HOST", "ws-tyo.hibira.io") + monkeypatch.setenv("HIBIRA_USE_TLS", "false") + monkeypatch.setenv("HIBIRA_PORT", "6001") + c = hibira.Client.from_env(app_id="2") + assert c.app_id == "2" and c.base_url == "http://ws-tyo.hibira.io:6001" + assert c.push_endpoint == "http://ws-tyo.hibira.io:6001" + monkeypatch.delenv("HIBIRA_APP_SECRET") + with pytest.raises(hibira.ConfigError): + hibira.Client.from_env() + + +def test_async_client(): + seen, handler = recorder() + + async def main(): + async with hibira.AsyncClient( + "app1", "key1", "secret1", http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)) + ) as c: + await c.trigger("a", "e", {"x": 1}) + assert c.authenticate_channel("private-a", "1.1")["auth"].startswith("key1:") + + asyncio.run(main()) + assert json.loads(seen[0].content)["data"] == '{"x":1}' + check_signature(seen[0]) diff --git a/tests/test_e2e.py b/tests/test_e2e.py new file mode 100644 index 0000000..bbac7f9 --- /dev/null +++ b/tests/test_e2e.py @@ -0,0 +1,166 @@ +"""End-to-end tests against a running Hibira (pusher-server) node. + + HIBIRA_E2E=1 HIBIRA_HOST=127.0.0.1 HIBIRA_PORT=8080 HIBIRA_USE_TLS=false \ + HIBIRA_APP_ID=... HIBIRA_APP_KEY=... HIBIRA_APP_SECRET=... pytest tests/test_e2e.py + +The app needs channel history on (history: true) for the rewind test. +""" + +import asyncio +import json +import os +import uuid + +import pytest + +import hibira + +pytestmark = pytest.mark.skipif(os.environ.get("HIBIRA_E2E") != "1", reason="set HIBIRA_E2E=1 to run against a server") + +websockets = pytest.importorskip("websockets") + + +class Socket: + def __init__(self, client): + self.client = client + self.events = [] + + async def __aenter__(self): + c = self.client + scheme = "wss" if c.use_tls else "ws" + port = f":{c.port}" if c.port else "" + self.ws = await websockets.connect(f"{scheme}://{c.host}{port}/app/{c.key}?protocol=7&client=python-e2e") + msg = json.loads(await self.ws.recv()) + assert msg["event"] == "pusher:connection_established" + self.socket_id = json.loads(msg["data"])["socket_id"] + self.reader = asyncio.create_task(self._read()) + return self + + async def __aexit__(self, *exc): + self.reader.cancel() + await self.ws.close() + + async def _read(self): + async for raw in self.ws: + msg = json.loads(raw) + data = msg.get("data") + if isinstance(data, str): + try: + data = json.loads(data) + except ValueError: + pass + self.events.append((msg.get("channel"), msg["event"], data)) + + async def send(self, event, data): + await self.ws.send(json.dumps({"event": event, "data": data})) + + async def subscribe(self, channel, user_data=None): + data = {"channel": channel} + if channel.startswith(("private-", "presence-")): + data.update(self.client.authenticate_channel(channel, self.socket_id, user_data)) + await self.send("pusher:subscribe", data) + await self.wait(lambda e: e[0] == channel and e[1].endswith("subscription_succeeded")) + + async def wait(self, pred, timeout=5.0): + for _ in range(int(timeout / 0.02)): + for e in self.events: + if pred(e): + return e + await asyncio.sleep(0.02) + raise AssertionError(f"timed out; got {self.events}") + + +def rebuild(events, stream_id): + parts = sorted( + (e[2] for e in events if e[1] == "ai-stream" and e[2].get("id") == stream_id), key=lambda p: p["seq"] + ) + assert [p["seq"] for p in parts] == list(range(len(parts))) + assert parts[-1]["done"] + return "".join(p.get("delta", "") for p in parts), parts + + +@pytest.fixture +def channel(): + return f"private-py-{uuid.uuid4().hex[:8]}" + + +def test_trigger_and_query(channel): + async def main(): + async with hibira.AsyncClient.from_env() as c, Socket(c) as ws: + await ws.subscribe(channel) + await c.trigger([channel, "public-other"], "hello", {"n": 1}) + assert (await ws.wait(lambda e: e[1] == "hello"))[2] == {"n": 1} + await c.trigger(channel, "skip-me", "x", socket_id=ws.socket_id) + await c.trigger_batch([{"channel": channel, "name": "batched", "data": {"n": 2}}]) + await ws.wait(lambda e: e[1] == "batched") + assert not any(e[1] == "skip-me" for e in ws.events) + info = await c.channel_info(channel, info="subscription_count") + assert info["occupied"] is True + assert channel in (await c.channels(prefix="private-py-"))["channels"] + + asyncio.run(main()) + + +def test_presence_users(): + room = f"presence-py-{uuid.uuid4().hex[:8]}" + + async def main(): + async with hibira.AsyncClient.from_env() as c, Socket(c) as ws: + await ws.subscribe(room, {"user_id": "ann", "user_info": {"name": "Ann"}}) + assert (await c.users(room))["users"] == [{"id": "ann"}] + + asyncio.run(main()) + + +def test_errors(): + c = hibira.Client.from_env() + with pytest.raises(hibira.HTTPError) as info: + c.trigger("public-x", "big", "x" * 200_000) + assert info.value.status in (400, 413) + bad = hibira.Client.from_env(secret="wrong-secret-0123456789") + with pytest.raises(hibira.AuthenticationError): + bad.trigger("public-x", "e", 1) + + +def test_stream_sync_and_rewind(channel): + async def main(): + c = hibira.Client.from_env() + async with Socket(c) as ws: + await ws.subscribe(channel) + words = [f"word{i} " for i in range(40)] + ["日本語 😀"] + out = await asyncio.to_thread( + hibira.stream_to_channel, c, channel, iter(words), meta={"role": "assistant"}, interval=0.01 + ) + await ws.wait(lambda e: e[1] == "ai-stream" and e[2].get("done")) + text, parts = rebuild(ws.events, out.id) + assert text == "".join(words) == out.text + assert parts[0]["meta"] == {"role": "assistant"} and parts[-1]["text"] == text + + async with Socket(c) as late: + await late.subscribe(channel) + await late.send("pusher:rewind", {"channel": channel, "seconds": 60}) + ok = await late.wait(lambda e: e[1] in ("pusher:rewind_succeeded", "pusher:rewind_error")) + if ok[1] == "pusher:rewind_error": + pytest.skip(f"channel history is off for this app: {ok[2]}") + text, _ = rebuild(late.events, out.id) + assert text == out.text + + asyncio.run(main()) + + +def test_stream_async(channel): + async def tokens(): + for i in range(20): + yield {"choices": [{"delta": {"content": f"{i} "}}]} + await asyncio.sleep(0.005) + + async def main(): + async with hibira.AsyncClient.from_env() as c, Socket(c) as ws: + await ws.subscribe(channel) + out = await hibira.astream_to_channel(c, channel, tokens(), interval=0.02, max_bytes=200) + await ws.wait(lambda e: e[1] == "ai-stream" and e[2].get("done")) + text, parts = rebuild(ws.events, out.id) + assert text == out.text == "".join(f"{i} " for i in range(20)) + assert all(len(json.dumps(p, separators=(",", ":")).encode()) <= 200 for p in parts) + + asyncio.run(main()) diff --git a/tests/test_fastapi.py b/tests/test_fastapi.py new file mode 100644 index 0000000..42c0736 --- /dev/null +++ b/tests/test_fastapi.py @@ -0,0 +1,71 @@ +import hashlib +import hmac + +import pytest + +fastapi = pytest.importorskip("fastapi") +from fastapi import Depends, FastAPI # noqa: E402 +from fastapi.testclient import TestClient # noqa: E402 + +import hibira # noqa: E402 +from hibira.fastapi import auth_router, webhook # noqa: E402 + +client = hibira.Client("1", "key", "secret") + + +async def authorize(request, channel): + if request.headers.get("x-user") != "ann": + return False + if channel.startswith("presence-"): + return {"user_id": "ann", "user_info": {"name": "Ann"}} + return channel == "private-ann" + + +app = FastAPI() +app.include_router(auth_router(client, authorize, user=lambda r: {"id": "ann"} if r.headers.get("x-user") else None)) +received = [] + + +@app.post("/hook") +def hook(payload: dict = Depends(webhook(client))): + received.append(payload) + return {"ok": True} + + +http = TestClient(app) + + +def test_private_and_presence_auth(): + r = http.post("/hibira/auth", data={"socket_id": "1.2", "channel_name": "private-ann"}, headers={"x-user": "ann"}) + assert r.status_code == 200 and r.json() == client.authenticate_channel("private-ann", "1.2") + r = http.post("/hibira/auth", json={"socket_id": "1.2", "channel_name": "presence-room"}, headers={"x-user": "ann"}) + assert r.json()["channel_data"] == '{"user_id":"ann","user_info":{"name":"Ann"}}' + assert ( + http.post( + "/hibira/auth", data={"socket_id": "1.2", "channel_name": "private-bob"}, headers={"x-user": "ann"} + ).status_code + == 403 + ) + assert http.post("/hibira/auth", data={"socket_id": "1.2", "channel_name": "private-ann"}).status_code == 403 + assert ( + http.post( + "/hibira/auth", data={"socket_id": "x", "channel_name": "private-ann"}, headers={"x-user": "ann"} + ).status_code + == 400 + ) + assert http.post("/hibira/auth", data={}).status_code == 400 + + +def test_user_auth(): + r = http.post("/hibira/user-auth", data={"socket_id": "1.2"}, headers={"x-user": "ann"}) + assert r.json() == client.authenticate_user("1.2", {"id": "ann"}) + assert http.post("/hibira/user-auth", data={"socket_id": "1.2"}).status_code == 403 + + +def test_webhook(): + body = b'{"time_ms":1,"events":[]}' + sig = hmac.new(b"secret", body, hashlib.sha256).hexdigest() + r = http.post("/hook", content=body, headers={"x-pusher-key": "key", "x-pusher-signature": sig}) + assert r.status_code == 200 and received[-1] == {"time_ms": 1, "events": []} + r = http.post("/hook", content=body, headers={"x-pusher-key": "key", "x-pusher-signature": "0" * 64}) + assert r.status_code == 401 diff --git a/tests/test_signing.py b/tests/test_signing.py new file mode 100644 index 0000000..96b612d --- /dev/null +++ b/tests/test_signing.py @@ -0,0 +1,90 @@ +import base64 +import hashlib +import hmac +import json + +import pytest + +import hibira +from hibira import _signing + +KEY = "278d425bdf160c739803" +SECRET = "7ad3773142a6692b25b8" + + +def test_sign_query_matches_pusher_reference(): + body = b'{"name":"foo","channels":["project-3"],"data":"{\\"some\\":\\"data\\"}"}' + q = _signing.sign_query(KEY, SECRET, "POST", "/apps/3/events", body=body, now=1353088179) + assert q["body_md5"] == "ec365a775a4cd0599faeb73354201b6f" + assert q["auth_signature"] == "da454824c97ba181a32ccc17a72625ba02771f50b50e1e7430e47a1f3f457e6c" + + +def test_private_channel_auth_matches_pusher_reference(): + c = hibira.Client("3", KEY, SECRET) + assert c.authenticate_channel("private-foobar", "1234.1234") == { + "auth": "278d425bdf160c739803:58df8b0c36d6982b82c3ecf6b4662e34fe8c25bba48f5369f135bf843651c3a4" + } + + +def test_presence_channel_auth(): + c = hibira.Client("3", KEY, SECRET) + out = c.authenticate_channel("presence-foobar", "1234.1234", {"user_id": 10, "user_info": {"name": "Mr. Channels"}}) + assert out["channel_data"] == '{"user_id":10,"user_info":{"name":"Mr. Channels"}}' + expected = hmac.new( + SECRET.encode(), f"1234.1234:presence-foobar:{out['channel_data']}".encode(), hashlib.sha256 + ).hexdigest() + assert out["auth"] == f"{KEY}:{expected}" + + +@pytest.mark.parametrize( + ("channel", "socket_id", "user_data"), + [ + ("public", "1.1", None), + ("private-a", "nope", None), + ("private-encrypted-a", "1.1", None), + ("presence-a", "1.1", None), + ("presence-a", "1.1", {"user_info": {}}), + ("private-a b", "1.1", None), + ], +) +def test_channel_auth_rejects(channel, socket_id, user_data): + with pytest.raises(ValueError): + hibira.Client("3", KEY, SECRET).authenticate_channel(channel, socket_id, user_data) + + +def test_user_auth(): + out = hibira.Client("3", KEY, SECRET).authenticate_user("1.2", {"id": "u1", "name": "Ann"}) + assert out["user_data"] == '{"id":"u1","name":"Ann"}' + sig = hmac.new(SECRET.encode(), b'1.2::user::{"id":"u1","name":"Ann"}', hashlib.sha256).hexdigest() + assert out["auth"] == f"{KEY}:{sig}" + with pytest.raises(ValueError): + hibira.Client("3", KEY, SECRET).authenticate_user("1.2", {"name": "x"}) + + +def test_verify_webhook(): + c = hibira.Client("3", KEY, SECRET) + body = b'{"time_ms":1,"events":[{"name":"channel_occupied","channel":"a"}]}' + sig = hmac.new(SECRET.encode(), body, hashlib.sha256).hexdigest() + assert c.verify_webhook(KEY, sig, body)["events"][0]["name"] == "channel_occupied" + assert c.verify_webhook(KEY, sig.upper(), body.decode())["time_ms"] == 1 + for key, s, b in [("other", sig, body), (KEY, "0" * 64, body), (KEY, None, body), (KEY, sig, body + b" ")]: + with pytest.raises(hibira.InvalidWebhook): + c.verify_webhook(key, s, b) + + +def test_push_token(): + c = hibira.Client("3", KEY, SECRET, push_instance_id="inst-1", push_secret_key="push-secret") + token = c.push_token("alice", ttl=60)["token"] + header, payload, sig = token.split(".") + + def dec(s): + return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4)) + + assert json.loads(dec(header)) == {"alg": "HS256", "typ": "JWT"} + claims = json.loads(dec(payload)) + assert claims["sub"] == "alice" + assert claims["iss"] == "https://inst-1.pushnotifications.pusher.com" + assert claims["exp"] - claims["iat"] == 60 + assert dec(sig) == hmac.new(b"push-secret", f"{header}.{payload}".encode(), hashlib.sha256).digest() + with pytest.raises(hibira.ConfigError): + hibira.Client("3", KEY, SECRET).push_token("alice") diff --git a/tests/test_stream.py b/tests/test_stream.py new file mode 100644 index 0000000..b179f1b --- /dev/null +++ b/tests/test_stream.py @@ -0,0 +1,218 @@ +import asyncio +import json +import threading +import time +from types import SimpleNamespace as NS + +import pytest + +import hibira +from hibira.stream import split, text_of + + +class Recorder: + def __init__(self, fail_at=None, delay=0.0): + self.events = [] + self.fail_at = fail_at + self.delay = delay + self.lock = threading.Lock() + + def trigger(self, channel, event, data, socket_id=None): + if self.delay: + time.sleep(self.delay) + with self.lock: + if self.fail_at is not None and len(self.events) == self.fail_at: + raise hibira.PayloadTooLarge(413, "too large") + self.events.append((channel, event, data, socket_id)) + + +class AsyncRecorder(Recorder): + async def trigger(self, channel, event, data, socket_id=None): + await asyncio.sleep(self.delay) + if self.fail_at is not None and len(self.events) == self.fail_at: + raise hibira.PayloadTooLarge(413, "too large") + self.events.append((channel, event, data, socket_id)) + + +def rebuild(events): + payloads = [e[2] for e in events] + assert [p["seq"] for p in payloads] == list(range(len(payloads))) + return "".join(p.get("delta", "") for p in payloads if not p.get("done")), payloads[-1] + + +def test_text_of_shapes(): + assert text_of("hi") == "hi" + assert text_of(b"hi") == "hi" + assert text_of({"type": "content_block_delta", "delta": {"type": "text_delta", "text": "a"}}) == "a" + assert text_of({"type": "content_block_delta", "delta": {"type": "input_json_delta", "partial_json": "{"}}) == "" + assert text_of(NS(type="content_block_delta", delta=NS(type="text_delta", text="b"))) == "b" + assert text_of({"type": "response.output_text.delta", "delta": "c"}) == "c" + assert text_of(NS(type="raw_response_event", data=NS(type="response.output_text.delta", delta="d"))) == "d" + assert text_of(NS(type="run_item_stream_event", item=None)) == "" + assert text_of({"choices": [{"delta": {"content": "e"}}]}) == "e" + assert text_of(NS(choices=[NS(delta=NS(content=None))])) == "" + assert text_of(NS(choices=[])) == "" + ai = NS(type="AIMessageChunk", content=[{"type": "text", "text": "f"}, {"type": "tool_use", "id": "x"}]) + assert text_of((ai, {"langgraph_node": "agent"})) == "f" + assert text_of(("messages", (NS(type="AIMessageChunk", content="g"), {}))) == "g" + assert text_of(NS(type="tool", content="secret tool output")) == "" + assert text_of(NS(text="h")) == "h" + assert text_of({"type": "text-delta", "textDelta": "i"}) == "i" + assert text_of(None) == "" and text_of(42) == "" + + +def test_split_respects_budget_and_characters(): + text = 'héllo "wörld"\n😀' * 50 + for budget in (16, 20, 33): + pieces = split(text, budget) + assert "".join(pieces) == text + for p in pieces: + assert len(json.dumps(p, ensure_ascii=False).encode()) - 2 <= budget + + +def test_stream_batches_and_finishes(): + rec = Recorder() + + def tokens(): + for word in "the quick brown fox".split(): + yield word + " " + + out = hibira.stream_to_channel(rec, "private-chat", tokens(), id="s1", meta={"role": "assistant"}, socket_id="1.2") + assert out == hibira.StreamResult("s1", "the quick brown fox ", len(rec.events)) + text, last = rebuild(rec.events) + assert text == "the quick brown fox " + assert last == {"id": "s1", "seq": len(rec.events) - 1, "done": True, "text": text} + first = rec.events[0] + assert first[0] == "private-chat" and first[1] == "ai-stream" and first[3] == "1.2" + assert first[2]["meta"] == {"role": "assistant"} + assert len(rec.events) == 2 # fast tokens are batched into one event + assert all("meta" not in e[2] for e in rec.events[1:]) + + +def test_stream_interval_flushes_slow_tokens(): + rec = Recorder() + + def tokens(): + yield "a" + time.sleep(0.15) + assert [e[2]["delta"] for e in rec.events] == ["a"] + yield "b" + + hibira.stream_to_channel(rec, "c", tokens(), interval=0.05) + assert rebuild(rec.events)[0] == "ab" + + +def test_stream_splits_large_text_under_max_bytes(): + rec = Recorder() + big = "字" * 5000 + out = hibira.stream_to_channel(rec, "c", [big], max_bytes=1000, meta={"m": 1}) + text, last = rebuild(rec.events) + assert text == big == out.text + for e in rec.events: + assert len(json.dumps(e[2], separators=(",", ":"), ensure_ascii=False).encode()) <= 1000 + assert "text" not in last + + +def test_empty_stream_sends_done_with_meta(): + rec = Recorder() + hibira.stream_to_channel(rec, "c", [], id="x", meta=None) + assert [e[2] for e in rec.events] == [{"id": "x", "seq": 0, "done": True, "meta": None, "text": ""}] + + +def test_source_error_is_sent_and_raised(): + rec = Recorder() + + def tokens(): + yield "partial" + raise RuntimeError("model overloaded") + + with pytest.raises(RuntimeError): + hibira.stream_to_channel(rec, "c", tokens()) + text, last = rebuild(rec.events) + assert text == "partial" and last["error"] == "model overloaded" and last["done"] + + +def test_stop_event_aborts(): + rec = Recorder() + stop = threading.Event() + + def tokens(): + yield "a" + stop.set() + yield "b" + + out = hibira.stream_to_channel(rec, "c", tokens(), stop=stop) + assert out.text == "a" + assert rec.events[-1][2]["error"] == "aborted" + + +def test_send_error_stops_and_raises(): + rec = Recorder(fail_at=0) + with pytest.raises(hibira.PayloadTooLarge): + hibira.stream_to_channel(rec, "c", iter(["a"] * 3), interval=10) + assert rec.events == [] + + +def test_utf8_bytes_split_across_chunks(): + rec = Recorder() + data = "日本語".encode() + out = hibira.stream_to_channel(rec, "c", [data[:1], data[1:4], data[4:]]) + assert out.text == "日本語" + + +def test_max_bytes_too_small(): + with pytest.raises(ValueError): + hibira.stream_to_channel(Recorder(), "c", ["a"], max_bytes=40, meta={"x": "y" * 20}) + + +def test_async_stream_in_order(): + rec = AsyncRecorder(delay=0.01) + + async def tokens(): + for i in range(30): + yield f"{i}," + await asyncio.sleep(0.004) + + out = asyncio.run(hibira.astream_to_channel(rec, "c", tokens(), interval=0.02, meta={"a": 1})) + text, last = rebuild(rec.events) + assert text == out.text == "".join(f"{i}," for i in range(30)) + assert last["done"] and last["text"] == text + assert 2 < len(rec.events) < 30 + + +def test_async_stream_with_sync_client_and_plain_iterable(): + rec = Recorder() + out = asyncio.run(hibira.astream_to_channel(rec, "c", ["x", "y"])) + assert out.text == "xy" and rebuild(rec.events)[0] == "xy" + + +def test_async_cancel_sends_aborted(): + rec = AsyncRecorder() + + async def tokens(): + yield "a" + await asyncio.sleep(10) + yield "b" + + async def main(): + task = asyncio.create_task(hibira.astream_to_channel(rec, "c", tokens(), interval=0.01)) + await asyncio.sleep(0.1) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + asyncio.run(main()) + text, last = rebuild(rec.events) + assert text == "a" and last["error"] == "aborted" + + +def test_async_source_error(): + rec = AsyncRecorder() + + async def tokens(): + yield "a" + raise ValueError("boom") + + with pytest.raises(ValueError): + asyncio.run(hibira.astream_to_channel(rec, "c", tokens())) + assert rec.events[-1][2]["error"] == "boom" From e39ef211d14f07bd622caf916c39116d6f3fde9b Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 1 Nov 2026 14:42:43 +0000 Subject: [PATCH 2/2] CI: install test Pythons with setup-uv (no 3.9 in setup-python on ubuntu-latest) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01SHtsCCHwKcFDPRqugEWRfU --- .github/workflows/test.yml | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 4bc6782..7be6c30 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -29,10 +29,11 @@ jobs: python: ["3.9", "3.10", "3.11", "3.12", "3.13"] steps: - uses: actions/checkout@v4 - - uses: actions/setup-python@v5 + # setup-uv installs any CPython version; setup-python has no 3.9 on newer runner images. + - uses: astral-sh/setup-uv@v6 with: python-version: ${{ matrix.python }} - - run: pip install -e . pytest fastapi + - run: uv pip install -e . pytest fastapi - run: pytest -q build: