"""Structured error reporting for `ria-agent register` (T2).""" from __future__ import annotations import json import sys import urllib.error from io import BytesIO from unittest.mock import patch import pytest from ria_toolkit_oss.agent import cli as agent_cli def _structured(reason: str) -> bytes: return json.dumps({"detail": {"reason": reason}}).encode() @pytest.mark.parametrize( "reason", ["invalid_key", "expired", "revoked", "already_consumed"], ) def test_explain_maps_known_reasons(reason): msg = agent_cli._explain_registration_failure(403, _structured(reason)) assert msg == agent_cli.REGISTRATION_REASON_MESSAGES[reason] def test_explain_unknown_reason_falls_through_with_code(): msg = agent_cli._explain_registration_failure(403, _structured("brand_new_thing")) assert "brand_new_thing" in msg assert "rejected" in msg.lower() def test_explain_string_detail(): body = json.dumps({"detail": "Forbidden"}).encode() msg = agent_cli._explain_registration_failure(403, body) assert msg == "Registration rejected: Forbidden" def test_explain_429_with_string_detail(): body = json.dumps({"detail": "Too many attempts; try again shortly"}).encode() msg = agent_cli._explain_registration_failure(429, body) assert "rate-limited" in msg assert "Too many attempts" in msg def test_explain_429_with_no_body(): msg = agent_cli._explain_registration_failure(429, b"") assert "rate-limited" in msg def test_explain_malformed_json(): msg = agent_cli._explain_registration_failure(500, b"boom") assert msg.startswith("HTTP 500") assert "boom" in msg def test_explain_empty_body(): msg = agent_cli._explain_registration_failure(502, b"") assert msg == "HTTP 502: no body" def _http_error(status: int, body: bytes) -> urllib.error.HTTPError: return urllib.error.HTTPError( url="http://hub/screens/agents/register", code=status, msg="", hdrs=None, # type: ignore[arg-type] fp=BytesIO(body), ) def test_user_agent_is_set_and_not_python_default(): """Cloudflare on `riahub.ai` returns 403 code 1010 to `Python-urllib/*`. Guarding the UA explicitly is the entire point of the register-flow fix; if this test ever breaks, the production bug is back. """ ua = agent_cli._user_agent() assert ua, "User-Agent must not be empty" assert not ua.lower().startswith("python-urllib"), ( f"User-Agent must not be Python's default (got {ua!r}) — Cloudflare blocks it" ) assert ua.startswith("ria-agent/") def test_register_request_carries_explicit_user_agent(tmp_path): """Capture the outbound urllib Request and verify the UA header is set.""" cfg_path = tmp_path / "agent.json" captured: dict = {} def _fake_urlopen(req, *args, **kwargs): # urllib normalizes header names; get_header takes the title-cased form. captured["ua"] = req.get_header("User-agent") captured["api_key"] = req.get_header("X-api-key") captured["timeout"] = kwargs.get("timeout") raise urllib.error.HTTPError( url=req.full_url, code=403, msg="", hdrs=None, # type: ignore[arg-type] fp=BytesIO(_structured("invalid_key")), ) with ( patch.dict("os.environ", {"RIA_AGENT_CONFIG": str(cfg_path)}, clear=False), patch("urllib.request.urlopen", side_effect=_fake_urlopen), patch.object( sys, "argv", ["ria-agent", "register", "--hub", "http://hub:3005", "--api-key", "ria_reg_x"], ), ): with pytest.raises(SystemExit): agent_cli.main() assert captured["ua"], "User-Agent header was not sent" assert not captured["ua"].lower().startswith("python-urllib") assert captured["api_key"] == "ria_reg_x" assert captured["timeout"] is not None, "register must pass a timeout to urlopen" def test_register_surfaces_reason_on_http_error(tmp_path, capsys): cfg_path = tmp_path / "agent.json" err = _http_error(403, _structured("revoked")) with ( patch.dict("os.environ", {"RIA_AGENT_CONFIG": str(cfg_path)}, clear=False), patch("urllib.request.urlopen", side_effect=err), patch.object( sys, "argv", ["ria-agent", "register", "--hub", "http://hub:3005", "--api-key", "ria_reg_x"], ), ): with pytest.raises(SystemExit) as exc: agent_cli.main() assert exc.value.code == 1 captured = capsys.readouterr() assert "revoked" in captured.err.lower() assert "Settings → RIA Agents" in captured.err # Config must NOT be written on failure. assert not cfg_path.exists() def test_default_hub_url_is_production(): """Lock in the constant so a future typo doesn't silently redirect users.""" assert agent_cli.DEFAULT_HUB_URL == "https://riahub.ai" def test_register_defaults_hub_to_production(tmp_path): """Omitting --hub uses the production hub URL constant.""" cfg_path = tmp_path / "agent.json" captured: dict = {} def _fake_urlopen(req, *args, **kwargs): captured["url"] = req.full_url raise urllib.error.HTTPError( url=req.full_url, code=403, msg="", hdrs=None, # type: ignore[arg-type] fp=BytesIO(_structured("invalid_key")), ) with ( patch.dict("os.environ", {"RIA_AGENT_CONFIG": str(cfg_path)}, clear=False), patch("urllib.request.urlopen", side_effect=_fake_urlopen), patch.object(sys, "argv", ["ria-agent", "register", "--api-key", "ria_reg_x"]), ): with pytest.raises(SystemExit): agent_cli.main() assert captured["url"] == f"{agent_cli.DEFAULT_HUB_URL}/screens/agents/register" def test_register_hub_override_wins_over_default(tmp_path): """Explicit --hub still wins; default is only a fallback.""" cfg_path = tmp_path / "agent.json" captured: dict = {} def _fake_urlopen(req, *args, **kwargs): captured["url"] = req.full_url raise urllib.error.HTTPError( url=req.full_url, code=403, msg="", hdrs=None, # type: ignore[arg-type] fp=BytesIO(_structured("invalid_key")), ) with ( patch.dict("os.environ", {"RIA_AGENT_CONFIG": str(cfg_path)}, clear=False), patch("urllib.request.urlopen", side_effect=_fake_urlopen), patch.object( sys, "argv", ["ria-agent", "register", "--hub", "http://whitehorse:3005", "--api-key", "ria_reg_x"], ), ): with pytest.raises(SystemExit): agent_cli.main() assert captured["url"] == "http://whitehorse:3005/screens/agents/register" assert agent_cli.DEFAULT_HUB_URL not in captured["url"]