87 lines
2.7 KiB
Python
87 lines
2.7 KiB
Python
"""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
|