93 lines
2.9 KiB
Python
93 lines
2.9 KiB
Python
"""Live integration tests against `backend.raycast.com`.
|
|
|
|
Opt-in: gated on the `--live` pytest flag AND on env vars supplying real
|
|
credentials. Without `--live` these are all skipped by the
|
|
`pytest_collection_modifyitems` hook in `conftest.py`.
|
|
|
|
Required env vars when running with `--live`:
|
|
RAYCAST_BEARER — a valid OAuth bearer token (rca_...)
|
|
RAYCAST_DEVICE_ID — a stable 64-hex device id
|
|
|
|
Optional:
|
|
RAYCAST_CONFIG — path to a `config.json` produced by `raycast-api init`.
|
|
Defaults to `./config.json` (repo-root one shipped with
|
|
this checkout).
|
|
|
|
Why these tests exist: the synthetic-fixture suite proves the signing math
|
|
is internally consistent. Only a real round-trip proves the bytes we send
|
|
on the wire actually match what the server expects — i.e. catches the
|
|
"refactor in `transforms.py` broke prod" class of regression.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from raycast_api.ai.types import Message
|
|
from raycast_api.client.http import Client
|
|
from raycast_api.config import Config
|
|
|
|
pytestmark = pytest.mark.live
|
|
|
|
|
|
def _bearer() -> str:
|
|
val = os.environ.get("RAYCAST_BEARER")
|
|
if not val:
|
|
pytest.skip("RAYCAST_BEARER not set")
|
|
return val
|
|
|
|
|
|
def _device_id() -> str:
|
|
val = os.environ.get("RAYCAST_DEVICE_ID")
|
|
if not val:
|
|
pytest.skip("RAYCAST_DEVICE_ID not set")
|
|
return val
|
|
|
|
|
|
def _config() -> Config:
|
|
path = Path(os.environ.get("RAYCAST_CONFIG", "config.json")).expanduser()
|
|
if not path.is_file():
|
|
pytest.skip(f"no config at {path} — run `raycast-api init` first")
|
|
return Config.load(path)
|
|
|
|
|
|
@pytest.fixture
|
|
def client() -> Client:
|
|
return Client(config=_config(), bearer_token=_bearer(), device_id=_device_id())
|
|
|
|
|
|
async def test_me(client: Client) -> None:
|
|
"""Bearer-token probe. Unsigned — proves the token is alive."""
|
|
async with client:
|
|
me = await client.me.get()
|
|
assert isinstance(me, dict)
|
|
assert me, "empty /me response — token likely invalid"
|
|
|
|
|
|
async def test_models_list(client: Client) -> None:
|
|
"""Lists `/ai/models`. Unsigned, but exercises auth + JSON decoding."""
|
|
async with client:
|
|
models = await client.models.list()
|
|
assert models.models, "no models returned"
|
|
|
|
|
|
async def test_chat_streaming(client: Client) -> None:
|
|
"""One-token chat completion. Proves signing works end-to-end.
|
|
|
|
Picks the first model from the live catalog so the test doesn't pin
|
|
a specific provider. Prompt is intentionally trivial to minimize cost.
|
|
"""
|
|
async with client:
|
|
catalog = await client.models.list()
|
|
model = catalog.models[0]
|
|
chunks: list[str] = []
|
|
async for chunk in client.chat.stream(
|
|
model=model, messages=[Message.user("Reply with the single word: OK")]
|
|
):
|
|
if chunk.text:
|
|
chunks.append(chunk.text)
|
|
assert chunks, "no text chunks streamed — signing or stream parsing broke"
|