free-claude-code/smoke/lib/http.py
Ali Khokhar 462a9430bb
Add local live smoke test suite (#148)
## Summary
- add an opt-in local `smoke/` pytest suite for API, auth, providers,
CLI, IDE-shaped requests, messaging, voice, tools, and thinking stream
contracts
- keep smoke tests out of normal CI collection with `testpaths =
["tests"]`
- write sanitized smoke artifacts under `.smoke-results/`

## Verification
- `uv run ruff format`
- `uv run ruff check`
- `uv run ty check`
- `uv run ty check smoke`
- `FCC_LIVE_SMOKE=1 FCC_SMOKE_TARGETS=all FCC_SMOKE_RUN_VOICE=1 uv run
pytest smoke -n 0 -m live -s --tb=short` -> 17 passed, 9 skipped
- `uv run pytest` -> 904 passed

## Notes
- Skipped live checks require local credentials/tools/services, such as
provider models, Telegram/Discord targets, voice backend, or Claude CLI.
- `claude-pick` smoke was intentionally removed.
2026-04-23 19:06:09 -07:00

70 lines
1.8 KiB
Python

"""HTTP helpers for live smoke requests."""
from __future__ import annotations
from typing import Any
import httpx
from .config import SmokeConfig, auth_headers, redacted
from .server import RunningServer
from .sse import SSEEvent, parse_sse_lines
def message_payload(
text: str,
*,
model: str = "claude-3-5-sonnet-20241022",
max_tokens: int = 128,
extra: dict[str, Any] | None = None,
) -> dict[str, Any]:
payload: dict[str, Any] = {
"model": model,
"max_tokens": max_tokens,
"messages": [{"role": "user", "content": text}],
}
if extra:
payload.update(extra)
return payload
def post_json(
server: RunningServer,
path: str,
payload: dict[str, Any],
config: SmokeConfig,
*,
headers: dict[str, str] | None = None,
) -> httpx.Response:
request_headers = headers or auth_headers()
response = httpx.post(
f"{server.base_url}{path}",
headers=request_headers,
json=payload,
timeout=config.timeout_s,
)
return response
def collect_message_stream(
server: RunningServer,
payload: dict[str, Any],
config: SmokeConfig,
*,
headers: dict[str, str] | None = None,
) -> list[SSEEvent]:
request_headers = headers or auth_headers()
with httpx.stream(
"POST",
f"{server.base_url}/v1/messages",
headers=request_headers,
json=payload,
timeout=config.timeout_s,
) as response:
if response.status_code != 200:
body = response.read().decode("utf-8", errors="replace")
raise AssertionError(
f"stream request failed: HTTP {response.status_code} "
f"{redacted(body[:1000])}"
)
return parse_sse_lines(response.iter_lines())