1"""Locks for the baseline runner's request shapes and key hygiene.
2
3The request builders are pure (no network), so the provider API shapes and the
4never-key-in-URL rule are pinned here without mocking urllib.
5"""
6import os
7import pathlib
8
9import source_baseline as sb
10
11
12def _with_keys(**keys):
13    for k in (

test_source_baseline.py on GitHub

"OPENAI_API_KEY"

,

"ANTHROPIC_API_KEY"

,

"GEMINI_API_KEY"

):

os.environ.pop(k, None)

os.environ.update(keys)

def test_openai_shape_uses_max_completion_tokens():

_with_keys(OPENAI_API_KEY="sk-test")

url, headers, body = sb.build_openai_request("gpt-4o", "sys", "hi")

assert url == "https://api.openai.com/v1/chat/completions"

assert body["max_completion_tokens"] == sb.MAX_TOKENS

assert "max_tokens" not in body, "gpt-5.x rejects max_tokens (HTTP 400)"

assert body["messages"][0] == {"role": "system", "content": "sys"}

def test_openai_no_system_message_when_empty():

_with_keys(OPENAI_API_KEY="sk-test")

_, _, body = sb.build_openai_request("gpt-4o", "", "hi")

assert [m["role"] for m in body["messages"]] == ["user"]

def test_anthropic_shape_uses_top_level_system():

_with_keys(ANTHROPIC_API_KEY="sk-ant")

url, headers, body = sb.build_anthropic_request("claude-3-5-sonnet", "sys", "hi")

assert url == "https://api.anthropic.com/v1/messages"

assert body["system"] == "sys"

assert headers["anthropic-version"] == "2023-06-01"

assert body["max_tokens"] == sb.MAX_TOKENS

def test_gemini_key_in_header_never_in_url():

_with_keys(GEMINI_API_KEY="AIza-test")

url, headers, body = sb.build_gemini_request("gemini-1.5-pro", "sys", "hi")

assert "AIza-test" not in url, "key must never be a URL query parameter"

assert "key=" not in url

assert headers["x-goog-api-key"] == "AIza-test"

assert body["systemInstruction"] == {"parts": [{"text": "sys"}]}

assert body["generationConfig"]["maxOutputTokens"] == sb.MAX_TOKENS

def test_all_provider_urls_are_official_hosts():

# Security note in the skill: the key only ever travels to its own

# provider's official endpoint.

_with_keys(OPENAI_API_KEY="a", ANTHROPIC_API_KEY="b", GEMINI_API_KEY="c")

urls = [

sb.build_openai_request("m", "", "x")[0],

sb.build_anthropic_request("m", "", "x")[0],

sb.build_gemini_request("m", "", "x")[0],

]

allowed = ("https://api.openai.com/", "https://api.anthropic.com/",

"https://generativelanguage.googleapis.com/")

for u in urls:

assert u.startswith(allowed), u

def test_env_file_loader_ignores_blank_and_malformed_lines(tmp_path: pathlib.Path):

p = tmp_path / ".source-provider-env"

p.write_text("\n# comment-ish\nOPENAI_API_KEY=sk-live\n", encoding="utf-8")

os.environ.pop("OPENAI_API_KEY", None)

sb.load_env_file(str(p))

assert os.environ["OPENAI_API_KEY"] == "sk-live"

def test_provider_selected_from_file_not_ambient_env(tmp_path):

# Review finding: file is authoritative; ambient keys must not win.

_with_keys(OPENAI_API_KEY="sk-ambient")

p = tmp_path / ".source-provider-env"

p.write_text("GEMINI_API_KEY=AIza-file\n", encoding="utf-8")

pairs = sb.load_env_file(str(p))

provider = next(k for k in sb.PROVIDERS if k in pairs)

assert provider == "GEMINI_API_KEY"

def _http_error(code, reason, body: bytes):

import io

import urllib.error

return urllib.error.HTTPError(

"https://api.openai.com/v1/chat/completions", code, reason,

hdrs=None, fp=io.BytesIO(body),

)

def test_error_detail_extracts_message_and_param():

# Review finding: the evaluator contract needs the 400 body's

# message/param — reason alone is just "Bad Request".

e = _http_error(400, "Bad Request", b'{"error": {"message": "Unsupported parameter: max_tokens", "param": "max_tokens", "type": "invalid_request_error"}}')

assert sb.error_detail(e) == "Unsupported parameter: max_tokens (param: max_tokens)"

def test_error_detail_handles_non_json_and_is_bounded():

e = _http_error(502, "Bad Gateway", b"<html>upstream error</html>" * 200)

d = sb.error_detail(e)

assert d.startswith("<html>upstream error")

assert len(d) <= 200

def test_error_detail_never_raises_on_unreadable_body():

import urllib.error

e = urllib.error.HTTPError("https://api.openai.com/x", 400, "Bad Request", None, None)

assert sb.error_detail(e) == ""

def test_status_line_keeps_contract_prefix():

# Step-3 classification greps on the "http_<code>" prefix — the appended

# detail must not break it.

e = _http_error(400, "Bad Request", b'{"error": {"message": "Unsupported parameter"}}')

detail = sb.error_detail(e)

status = f"http_{e.code}: {e.reason}" + (f" — {detail}" if detail else "")

assert status.startswith("http_400: Bad Request")

assert "Unsupported parameter" in status

def test_auth_error_bodies_are_suppressed_entirely():

# Review finding: an auth endpoint can echo the submitted credential in

# its body, and 401/403 classification uses the code alone — so auth

# bodies carry no detail at all.

for code, reason in ((401, "Unauthorized"), (403, "Forbidden")):

e = _http_error(code, reason, b'{"error": {"message": "bad key sk-live-SECRETKEY123"}}')

assert sb.error_detail(e, ["sk-live-SECRETKEY123"]) == ""

status = f"http_{e.code}: {e.reason}"

assert "SECRETKEY123" not in status

def test_echoed_key_never_reaches_detail_or_status():

# Review repro: FULL_SECRET_IN_DETAIL must be impossible. A 400 body that

# echoes the key (JSON path) and a non-JSON fallback that echoes it are

# both redacted inside error_detail itself.

secrets = ["sk-live-SECRETKEY123"]

e = _http_error(400, "Bad Request", b'{"error": {"message": "invalid key sk-live-SECRETKEY123 for model"}}')

d = sb.error_detail(e, secrets)

assert "SECRETKEY123" not in d and "***" in d

e2 = _http_error(500, "Server Error", b"upstream said: sk-live-SECRETKEY123 rejected")

d2 = sb.error_detail(e2, secrets)

assert "SECRETKEY123" not in d2 and "***" in d2

def test_redact_strips_key_from_exception_status():

# Review finding: http.client rejects an invalid header with the FULL

# header value in the ValueError message — 'Bearer sk-live-KEY' would land

# in the status JSONL, breaking the never-in-output promise.

msg = "error: ValueError: Invalid header value b'Bearer sk-live-SECRET123\\rX'"

assert "SECRET123" not in sb.redact(msg, ["sk-live-SECRET123"])

assert "***" in sb.redact(msg, ["sk-live-SECRET123"])

def test_redact_tolerates_empty_secrets():

assert sb.redact("error: boom", ["", None]) == "error: boom"

def test_source_provider_env_is_authoritative(tmp_path, monkeypatch):

# Review finding: a file carrying several provider keys must not fall back

# to dict-order guessing (OpenAI-first) — SOURCE_PROVIDER, the helper's

# declared input, decides.

pairs = {"OPENAI_API_KEY": "a", "ANTHROPIC_API_KEY": "b"}

monkeypatch.setenv("SOURCE_PROVIDER", "anthropic")

assert sb.pick_provider_key(pairs) == "ANTHROPIC_API_KEY"

monkeypatch.setenv("SOURCE_PROVIDER", "openai")

assert sb.pick_provider_key(pairs) == "OPENAI_API_KEY"

# stated provider whose key is NOT in the file → hard None, never a guess

monkeypatch.setenv("SOURCE_PROVIDER", "google")

assert sb.pick_provider_key(pairs) is None

monkeypatch.delenv("SOURCE_PROVIDER")

assert sb.pick_provider_key(pairs) == "OPENAI_API_KEY"

def test_redact_catches_escaped_control_chars_in_exception_text():

# Adjudicated finding: a raw CR inside the credential renders ESCAPED in

# exception text (one control char -> backslash-r as two characters), so

# literal replacement missed it and the key leaked in escaped form.

secret = "sk-live-SECRET\rKEY123"

escaped_msg = "error: ValueError: Invalid header value b'Bearer sk-live-SECRET\\rKEY123'"

out = sb.redact(escaped_msg, [secret])

assert "SECRET" not in out and "KEY123" not in out

def test_redact_catches_fragments_around_control_chars():

# Even a partial echo of either side of the control char must not survive.

secret = "sk-live-SECRET\nKEY123456"

out = sb.redact("provider said sk-live-SECRET then KEY123456 rejected", [secret])

assert "sk-live-SECRET" not in out and "KEY123456" not in out