"""Regression coverage for the Leo Telegram -> Living IP HTTP chat bridge.""" import sys from pathlib import Path import pytest REPO_ROOT = Path(__file__).resolve().parents[1] TELEGRAM_DIR = REPO_ROOT / "telegram" sys.path.insert(0, str(TELEGRAM_DIR)) from agent_config import load_agent_config # noqa: E402 from http_chat_proxy import build_chat_proxy_payload, extract_chat_proxy_reply # noqa: E402 def test_leo_config_opts_into_http_chat_proxy_without_changing_default_agents(): leo = load_agent_config(str(TELEGRAM_DIR / "agents" / "leo.yaml")) rio = load_agent_config(str(TELEGRAM_DIR / "agents" / "rio.yaml")) assert leo.name == "Leo" assert leo.http_chat_proxy_url == "https://leo.livingip.xyz/api/agents/leo/chat" assert leo.respond_to_private_chats is True assert "@teLEOhuman" in leo.mention_aliases assert rio.http_chat_proxy_url is None assert rio.respond_to_private_chats is False def test_invalid_http_chat_proxy_url_fails_closed(tmp_path): config = tmp_path / "bad.yaml" config.write_text( """ name: Leo handle: "@teLEOhuman" bot_token_file: leo-telegram-bot-token pentagon_agent_id: livingip-leo domain: collective-intelligence kb_scope: primary: ["core"] voice_summary: "test" voice_definition: "test" learnings_file: agents/leo/learnings.md http_chat_proxy_url: "not-a-url" """.strip() ) with pytest.raises(ValueError, match="http_chat_proxy_url"): load_agent_config(str(config)) def test_proxy_payload_contains_no_secret_material(): payload = build_chat_proxy_payload( message="Can Leo use x402 paid research now?", source="telegram", agent="leo", chat_id=123, message_id=456, username="tester", ) assert payload == { "message": "Can Leo use x402 paid research now?", "metadata": { "source": "telegram", "agent": "leo", "chat_id": 123, "message_id": 456, "username": "tester", }, } assert "token" not in str(payload).lower() assert "secret" not in str(payload).lower() @pytest.mark.parametrize( ("payload", "expected"), [ ({"reply": "public route reply"}, "public route reply"), ({"decision": {"reply": "analysis route reply"}}, "analysis route reply"), ({"llm": {"decision": {"reply": "nested decision reply"}}}, "nested decision reply"), ], ) def test_extract_chat_proxy_reply_accepts_retained_leo_shapes(payload, expected): assert extract_chat_proxy_reply(payload) == expected def test_extract_chat_proxy_reply_fails_closed_on_missing_reply(): assert extract_chat_proxy_reply({"schema": "livingip.x402.leoChatResponse.v1"}) is None