"""HTTP client tests using `aioresponses` to mock aiohttp. What's exercised here, in roughly increasing scope: - Header construction, including signed-vs-unsigned, resume mode, content-type omission, browser-fluff toggle. - URL composition (relative path → `Config.backend_url` + path). - End-to-end request flow against a mocked endpoint, including JSON body serialisation byte-equal to the bytes we sign. - Error mapping (401 → AuthError, 429 → RateLimitError, 5xx → HTTPStatusError). - Retry behaviour: retries on 429/5xx with re-signing (fresh timestamp each attempt), no retry on 4xx, respect for `Retry-After`. - Streaming: synthetic SSE bytes parsed into `SSEEvent`s, error events, no-retry-on-2xx. Tests use a `Config` built in-memory (no discovery) with a deterministic reference secret so signatures are reproducible across runs. """ from __future__ import annotations import re from typing import Any import aiohttp import pytest from aioresponses import aioresponses from raycast_api.client import Client, RetryPolicy from raycast_api.client.streaming import SSEEvent from raycast_api.config import Config from raycast_api.errors import ( AuthError, HTTPStatusError, RateLimitError, StreamError, TransportError, ) from raycast_api.signing_spec import RotRange, SigningSpec REFERENCE_SECRET = "6bc455473576ce2cd6f70426caff867aabbe3f7291c1a79681af5e8ce0ca1408" DEVICE_ID = "20eca913cada74f879e6535304f9d44da380c28eb855065c0d71017a3d7c3099" FIXED_TIMESTAMP = 1778858809 def _make_config() -> Config: """Synthetic config with the real-world signing spec and reference secret. The secret comes from `sign.py` (public reference value); the spec mirrors what Phase 2's discovery produces for production Raycast Beta. Bundle / launcher hashes are placeholder zeros — they're not exercised here. """ spec = SigningSpec( rot_fn_name="Sur", signing_fn_name="Nkt", rot_ranges=[ RotRange(start=65, end=90, shift=13), RotRange(start=97, end=122, shift=13), RotRange(start=48, end=57, shift=5), ], ) return Config( signature_secret=REFERENCE_SECRET, signing_spec=spec, app_version="0.60.1.0", user_agent="Raycast/0.60.1.0 (x-macOS Version 26.3.1)", bundle_hash="0" * 64, launcher_hash="0" * 64, ) def _make_client(**kwargs: Any) -> Client: """Default client with a fixed clock so signatures are reproducible.""" return Client( config=_make_config(), bearer_token="rca_test_token", device_id=DEVICE_ID, clock=lambda: FIXED_TIMESTAMP, **kwargs, ) class TestHeaderBuilding: def test_signed_post_full_header_set(self) -> None: client = _make_client() body = b'{"hello":"world"}' headers = client.build_headers( sign=True, body=body, content_type="application/json" ) assert headers["X-Raycast-Timestamp"] == str(FIXED_TIMESTAMP) assert headers["X-Raycast-DeviceId"] == DEVICE_ID assert headers["X-Raycast-Experimental"] == "autoModels" assert "X-Raycast-Signature-v2" in headers assert headers["Content-Type"] == "application/json" assert headers["Authorization"] == "Bearer rca_test_token" assert headers["User-Agent"].startswith("Raycast/") def test_signed_resume_get_omits_content_type(self) -> None: client = _make_client() headers = client.build_headers( sign=True, body=b"", is_resume=True, last_event_id="42", content_type="application/json", ) assert "Content-Type" not in headers assert headers["Last-Event-ID"] == "42" assert "X-Raycast-Signature-v2" in headers def test_unsigned_omits_raycast_headers(self) -> None: client = _make_client() headers = client.build_headers(sign=False, body=b"") assert "X-Raycast-Signature-v2" not in headers assert "X-Raycast-Timestamp" not in headers assert "X-Raycast-DeviceId" not in headers assert "X-Raycast-Experimental" not in headers assert "Authorization" in headers assert "User-Agent" in headers def test_browser_headers_present_by_default(self) -> None: client = _make_client() headers = client.build_headers(sign=True, body=b"") assert headers["Accept"] == "*/*" assert headers["Origin"] == "file://" assert headers["Sec-Fetch-Site"] == "cross-site" assert headers["Sec-Fetch-Mode"] == "cors" assert headers["Sec-Fetch-Dest"] == "empty" assert headers["Accept-Language"] == "en-US" def test_browser_headers_can_be_disabled(self) -> None: client = _make_client(browser_headers=False, locale="de-DE") headers = client.build_headers(sign=True, body=b"") assert "Accept" not in headers assert "Origin" not in headers assert "Accept-Language" not in headers def test_locale_drives_accept_language(self) -> None: client = _make_client(locale="ru-RU") headers = client.build_headers(sign=True, body=b"") assert headers["Accept-Language"] == "ru-RU" def test_empty_bearer_omits_authorization(self) -> None: client = Client( config=_make_config(), bearer_token="", device_id=DEVICE_ID, clock=lambda: FIXED_TIMESTAMP, ) headers = client.build_headers(sign=False, body=b"") assert "Authorization" not in headers def test_signature_matches_reference_signer(self) -> None: """The header builder uses the same Signer as `sign.py`, so byte-match.""" from raycast_api.signing import Signer client = _make_client() signer = Signer(spec=_make_config().signing_spec, secret=REFERENCE_SECRET) body = b'{"buffer_id":"00000000-0000-0000-0000-000000000000"}' headers = client.build_headers( sign=True, body=body, content_type="application/json" ) expected = signer.sign( timestamp=str(FIXED_TIMESTAMP), device_id=DEVICE_ID, body=body ) assert headers["X-Raycast-Signature-v2"] == expected def test_extra_headers_merged_last(self) -> None: client = _make_client() headers = client.build_headers( sign=True, body=b"", extra={"X-Trace-Id": "abc", "User-Agent": "Overridden"} ) assert headers["X-Trace-Id"] == "abc" assert headers["User-Agent"] == "Overridden" class TestUrl: def test_relative_path(self) -> None: client = _make_client() assert client._url("/api/v1/me") == "https://backend.raycast.com/api/v1/me" def test_path_without_leading_slash(self) -> None: client = _make_client() assert client._url("api/v1/me") == "https://backend.raycast.com/api/v1/me" def test_absolute_url_passthrough(self) -> None: client = _make_client() assert ( client._url("https://other.example.com/x") == "https://other.example.com/x" ) def test_custom_backend_url(self) -> None: cfg = _make_config() cfg.backend_url = "http://localhost:5001" client = Client( config=cfg, bearer_token="x", device_id=DEVICE_ID, clock=lambda: 0 ) assert client._url("/api/v1/me") == "http://localhost:5001/api/v1/me" URL_ME = "https://backend.raycast.com/api/v1/me" URL_CHAT = "https://backend.raycast.com/api/v1/ai/chat_completions" URL_FILES = "https://backend.raycast.com/api/v1/ai/files" @pytest.fixture def mock_aiohttp(): with aioresponses() as m: yield m class TestRequestHappyPath: async def test_unsigned_get_returns_response( self, mock_aiohttp: aioresponses ) -> None: mock_aiohttp.get(URL_ME, status=200, payload={"id": "user_1"}) async with _make_client() as client: async with client.request("GET", "/api/v1/me", sign=False) as resp: data = await resp.json() assert data == {"id": "user_1"} async def test_signed_post_with_json_body(self, mock_aiohttp: aioresponses) -> None: captured: dict[str, Any] = {} def callback(url: object, **kwargs: Any) -> Any: from aioresponses import CallbackResult captured["headers"] = dict(kwargs["headers"]) captured["body"] = kwargs.get("data") return CallbackResult(status=200, payload={"ok": True}) mock_aiohttp.post(URL_CHAT, callback=callback) async with _make_client() as client: async with client.request( "POST", "/api/v1/ai/chat_completions", json_body={"messages": [{"role": "user", "content": {"text": "hi"}}]}, ) as resp: assert resp.status == 200 assert "X-Raycast-Signature-v2" in captured["headers"] assert captured["headers"]["X-Raycast-Timestamp"] == str(FIXED_TIMESTAMP) assert captured["headers"]["X-Raycast-DeviceId"] == DEVICE_ID assert captured["headers"]["Content-Type"] == "application/json" body = captured["body"] assert isinstance(body, (bytes, bytearray)) assert b'"messages"' in body assert b": " not in body async def test_delete_with_body(self, mock_aiohttp: aioresponses) -> None: captured: dict[str, Any] = {} def callback(url: object, **kwargs: Any) -> Any: from aioresponses import CallbackResult captured["body"] = kwargs.get("data") return CallbackResult(status=204) mock_aiohttp.delete(URL_FILES, callback=callback) async with _make_client() as client: async with client.request( "DELETE", "/api/v1/ai/files", json_body={"chat_ids": ["abc"]} ) as resp: assert resp.status == 204 assert captured["body"] == b'{"chat_ids":["abc"]}' async def test_resume_get_omits_content_type_in_request( self, mock_aiohttp: aioresponses ) -> None: captured: dict[str, Any] = {} def callback(url: object, **kwargs: Any) -> Any: from aioresponses import CallbackResult captured["headers"] = dict(kwargs["headers"]) return CallbackResult(status=200, body=b"") url_resume_re = re.compile( r"^https://backend\.raycast\.com/api/v1/ai/chat_completions/resume.*$" ) mock_aiohttp.get(url_resume_re, callback=callback) async with _make_client() as client: async with client.request( "GET", "/api/v1/ai/chat_completions/resume", sign=True, is_resume=True, last_event_id="9", params={"buffer_id": "abc"}, ): pass assert "Content-Type" not in captured["headers"] assert captured["headers"]["Last-Event-ID"] == "9" assert "X-Raycast-Signature-v2" in captured["headers"] async def test_body_passthrough_str(self, mock_aiohttp: aioresponses) -> None: captured: dict[str, Any] = {} def callback(url: object, **kwargs: Any) -> Any: from aioresponses import CallbackResult captured["body"] = kwargs.get("data") return CallbackResult(status=200, body=b"") mock_aiohttp.post(URL_FILES, callback=callback) async with _make_client() as client: async with client.request("POST", "/api/v1/ai/files", body='{"raw":1}'): pass assert captured["body"] == b'{"raw":1}' async def test_body_passthrough_bytes(self, mock_aiohttp: aioresponses) -> None: captured: dict[str, Any] = {} def callback(url: object, **kwargs: Any) -> Any: from aioresponses import CallbackResult captured["body"] = kwargs.get("data") return CallbackResult(status=200, body=b"") mock_aiohttp.post(URL_FILES, callback=callback) async with _make_client() as client: async with client.request("POST", "/api/v1/ai/files", body=b"\x00raw"): pass assert captured["body"] == b"\x00raw" async def test_both_body_and_json_rejected(self) -> None: async with _make_client() as client: with pytest.raises(ValueError, match="not both"): async with client.request( "POST", "/api/v1/ai/files", body=b"x", json_body={"a": 1} ): pass async def test_external_session_not_closed(self) -> None: external = aiohttp.ClientSession() try: client = Client( config=_make_config(), bearer_token="t", device_id=DEVICE_ID, session=external, clock=lambda: 0, ) await client.close() assert not external.closed finally: await external.close() class TestErrorMapping: async def test_401_becomes_auth_error(self, mock_aiohttp: aioresponses) -> None: mock_aiohttp.get(URL_ME, status=401, body=b"unauthorized") async with _make_client(retry=RetryPolicy(max_attempts=1)) as client: with pytest.raises(AuthError) as ei: async with client.request("GET", "/api/v1/me", sign=False): pass assert ei.value.status == 401 assert ei.value.body == "unauthorized" async def test_429_becomes_rate_limit_error( self, mock_aiohttp: aioresponses ) -> None: mock_aiohttp.get( URL_ME, status=429, headers={"Retry-After": "12"}, body=b"slow down" ) async with _make_client(retry=RetryPolicy(max_attempts=1)) as client: with pytest.raises(RateLimitError) as ei: async with client.request("GET", "/api/v1/me", sign=False): pass assert ei.value.status == 429 assert ei.value.retry_after == 12.0 async def test_500_becomes_status_error(self, mock_aiohttp: aioresponses) -> None: mock_aiohttp.get(URL_ME, status=500, body=b"oops") async with _make_client(retry=RetryPolicy(max_attempts=1)) as client: with pytest.raises(HTTPStatusError) as ei: async with client.request("GET", "/api/v1/me", sign=False): pass assert ei.value.status == 500 assert not isinstance(ei.value, (AuthError, RateLimitError)) async def test_404_not_retried(self, mock_aiohttp: aioresponses) -> None: mock_aiohttp.get(URL_ME, status=404) async with _make_client() as client: with pytest.raises(HTTPStatusError) as ei: async with client.request("GET", "/api/v1/me", sign=False): pass assert ei.value.status == 404 class TestRetry: async def test_retries_503_then_succeeds(self, mock_aiohttp: aioresponses) -> None: sleeps: list[float] = [] async def fake_sleep(d: float) -> None: sleeps.append(d) mock_aiohttp.get(URL_ME, status=503) mock_aiohttp.get(URL_ME, status=200, payload={"ok": True}) client = Client( config=_make_config(), bearer_token="t", device_id=DEVICE_ID, clock=lambda: 0, sleep=fake_sleep, retry=RetryPolicy(max_attempts=3, initial_delay=0.1, max_delay=1.0), ) async with client: async with client.request("GET", "/api/v1/me", sign=False) as resp: data = await resp.json() assert data == {"ok": True} assert sleeps == [0.1] async def test_respects_retry_after_on_429( self, mock_aiohttp: aioresponses ) -> None: sleeps: list[float] = [] async def fake_sleep(d: float) -> None: sleeps.append(d) mock_aiohttp.get(URL_ME, status=429, headers={"Retry-After": "2"}) mock_aiohttp.get(URL_ME, status=200, payload={"ok": True}) client = Client( config=_make_config(), bearer_token="t", device_id=DEVICE_ID, clock=lambda: 0, sleep=fake_sleep, retry=RetryPolicy(max_attempts=3, initial_delay=0.5, max_delay=10), ) async with client: async with client.request("GET", "/api/v1/me", sign=False): pass assert sleeps == [2.0] async def test_gives_up_after_max_attempts( self, mock_aiohttp: aioresponses ) -> None: async def fake_sleep(_d: float) -> None: pass for _ in range(4): mock_aiohttp.get(URL_ME, status=503) client = Client( config=_make_config(), bearer_token="t", device_id=DEVICE_ID, clock=lambda: 0, sleep=fake_sleep, retry=RetryPolicy(max_attempts=3, initial_delay=0.01), ) async with client: with pytest.raises(HTTPStatusError) as ei: async with client.request("GET", "/api/v1/me", sign=False): pass assert ei.value.status == 503 async def test_resigns_on_retry(self, mock_aiohttp: aioresponses) -> None: """Each attempt re-signs with a fresh timestamp, not the original one.""" clock = iter([1000, 1001]) captured: list[dict[str, str]] = [] def cb(url: object, **kwargs: Any) -> Any: from aioresponses import CallbackResult captured.append(dict(kwargs["headers"])) status = 503 if len(captured) == 1 else 200 return CallbackResult(status=status, payload={}) mock_aiohttp.post(URL_CHAT, callback=cb) mock_aiohttp.post(URL_CHAT, callback=cb) async def fake_sleep(_d: float) -> None: pass client = Client( config=_make_config(), bearer_token="t", device_id=DEVICE_ID, clock=lambda: next(clock), sleep=fake_sleep, retry=RetryPolicy(max_attempts=2, initial_delay=0.01), ) async with client: async with client.request( "POST", "/api/v1/ai/chat_completions", json_body={"x": 1} ): pass assert captured[0]["X-Raycast-Timestamp"] == "1000" assert captured[1]["X-Raycast-Timestamp"] == "1001" assert ( captured[0]["X-Raycast-Signature-v2"] != captured[1]["X-Raycast-Signature-v2"] ) class TestStreaming: async def test_error_event_raises(self, mock_aiohttp: aioresponses) -> None: body = b'event: error\ndata: {"message":"boom"}\n\n' mock_aiohttp.post(URL_CHAT, status=200, body=body) async with _make_client() as client: events: list[SSEEvent] = [] with pytest.raises(StreamError) as ei: async for evt in client.stream( "POST", "/api/v1/ai/chat_completions", json_body={"x": 1} ): events.append(evt) assert ei.value.payload == {"message": "boom"} assert len(events) == 1 assert events[0].is_error async def test_stream_does_not_retry_on_2xx( self, mock_aiohttp: aioresponses ) -> None: body = b'data: a\n\nevent: complete\ndata: {"complete":true}\n\n' mock_aiohttp.post(URL_CHAT, status=200, body=body) async with _make_client() as client: events = [e async for e in client.stream("POST", URL_CHAT, json_body={})] assert [e.data for e in events] == ["a", '{"complete":true}'] class TestSessionLifecycle: async def test_request_without_session_raises_clear_error(self) -> None: client = Client( config=_make_config(), bearer_token="t", device_id=DEVICE_ID, clock=lambda: 0, ) with pytest.raises(RuntimeError, match="not initialised"): async with client.request("GET", "/api/v1/me", sign=False): pass async def test_owned_session_closed_on_exit(self) -> None: client = _make_client() async with client: assert client._session is not None and not client._session.closed assert client._session is None