328 lines
12 KiB
Python
328 lines
12 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 ( # noqa: E402
|
|
build_chat_proxy_payload,
|
|
build_smart_research_proxy_payload,
|
|
extract_auto_smart_research_followup_goal,
|
|
extract_auto_smart_research_goal,
|
|
extract_paid_work_order_id,
|
|
extract_chat_proxy_reply,
|
|
extract_smart_research_goal,
|
|
should_attach_structured_market_context,
|
|
smart_research_command_names,
|
|
)
|
|
from market_data import extract_market_data_tokens, format_price_context # 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"))
|
|
leo_wallet_test = load_agent_config(str(TELEGRAM_DIR / "agents" / "leo-wallet-test.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 leo_wallet_test.name == "Leo Wallet Test"
|
|
assert leo_wallet_test.http_chat_proxy_url == "https://leo.livingip.xyz/api/agents/leo/chat"
|
|
assert leo_wallet_test.http_research_proxy_url == "https://leo.livingip.xyz/api/agents/leo/research"
|
|
assert "/smart_research" in leo_wallet_test.smart_research_command_prefixes
|
|
assert leo_wallet_test.auto_smart_research_from_chat is True
|
|
assert leo_wallet_test.respond_to_private_chats is True
|
|
assert leo_wallet_test.bot_token_file == "leo-test-telegram-bot-token"
|
|
assert "@lipleowallet0622183538bot" in leo_wallet_test.mention_aliases
|
|
assert rio.http_chat_proxy_url is None
|
|
assert rio.respond_to_private_chats is False
|
|
assert leo.auto_smart_research_from_chat 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()
|
|
|
|
|
|
def test_smart_research_payload_is_no_spend_by_default():
|
|
payload = build_smart_research_proxy_payload(
|
|
research_goal="Find x402 evidence",
|
|
source="telegram",
|
|
agent="leo",
|
|
chat_id=123,
|
|
message_id=456,
|
|
username="tester",
|
|
max_amount_usd=0.01,
|
|
)
|
|
|
|
assert payload["message"] == "Find x402 evidence"
|
|
assert payload["research_goal"] == "Find x402 evidence"
|
|
assert payload["allow_paid_execution"] is False
|
|
assert payload["max_amount_usd"] == 0.01
|
|
assert "approval_ref" not in payload
|
|
assert "token" not in str(payload).lower()
|
|
assert "secret" not in str(payload).lower()
|
|
|
|
|
|
def test_smart_research_payload_can_resume_paid_work_order_without_secret_material():
|
|
payload = build_smart_research_proxy_payload(
|
|
research_goal="what are the current discussions about MetaDAO Ranger Finance on Twitter?",
|
|
source="telegram",
|
|
agent="leo",
|
|
chat_id=123,
|
|
message_id=456,
|
|
username="tester",
|
|
work_order_id="sponsored_work_orders:f951ccc6c7762ecba6f76cf6",
|
|
original_research_goal="what are the current discussions about MetaDAO Ranger Finance on Twitter?",
|
|
)
|
|
|
|
assert payload["work_order_id"] == "sponsored_work_orders:f951ccc6c7762ecba6f76cf6"
|
|
assert (
|
|
payload["original_research_goal"]
|
|
== "what are the current discussions about MetaDAO Ranger Finance on Twitter?"
|
|
)
|
|
assert payload["allow_paid_execution"] is False
|
|
assert "approval_ref" not in payload
|
|
assert "token" not in str(payload).lower()
|
|
assert "secret" not in str(payload).lower()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("message", "expected"),
|
|
[
|
|
("/smart_research find x402 evidence", "find x402 evidence"),
|
|
("@lipleowallet0622183538bot /smart_research find x402 evidence", "find x402 evidence"),
|
|
("/paid_research@lipleowallet0622183538bot find x402 evidence", "find x402 evidence"),
|
|
("can Leo use x402 paid research now?", None),
|
|
("/smart_research", None),
|
|
],
|
|
)
|
|
def test_extract_smart_research_goal(message, expected):
|
|
assert extract_smart_research_goal(message) == expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("message", "expected"),
|
|
[
|
|
(
|
|
"work_order_id: sponsored_work_orders:f951ccc6c7762ecba6f76cf6",
|
|
"sponsored_work_orders:f951ccc6c7762ecba6f76cf6",
|
|
),
|
|
(
|
|
"paid receipt payment_receipts:f951ccc6c7762ecba6f76cf6 thanks",
|
|
"payment_receipts:f951ccc6c7762ecba6f76cf6",
|
|
),
|
|
("no paid id here", None),
|
|
],
|
|
)
|
|
def test_extract_paid_work_order_id(message, expected):
|
|
assert extract_paid_work_order_id(message) == expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("message", "expected"),
|
|
[
|
|
(
|
|
"@lipleowallet0622183538bot how much revenue does MetaDAO make today?",
|
|
"how much revenue does MetaDAO make today?",
|
|
),
|
|
(
|
|
"what is the volume and fdv of omnipair avici umbra? should i buy them yes or no",
|
|
"what is the volume and fdv of omnipair avici umbra? should i buy them yes or no",
|
|
),
|
|
("Can you find current sources on x402 usage?", "Can you find current sources on x402 usage?"),
|
|
(
|
|
"what is the latest trend of internet finance on Twitter",
|
|
"what is the latest trend of internet finance on Twitter",
|
|
),
|
|
(
|
|
"what are your thoughts on how metadao managed the ranger finance situation",
|
|
"what are your thoughts on how metadao managed the ranger finance situation",
|
|
),
|
|
(
|
|
"who was at fault for Ranger according to Twitter?",
|
|
"who was at fault for Ranger according to Twitter?",
|
|
),
|
|
(
|
|
"what is MetaDAO's position on Ranger Finance?",
|
|
"what is MetaDAO's position on Ranger Finance?",
|
|
),
|
|
(
|
|
"how did Jupiter handle the outage situation?",
|
|
"how did Jupiter handle the outage situation?",
|
|
),
|
|
(
|
|
"assess whether the protocol valuation is fair versus fintech peers",
|
|
"assess whether the protocol valuation is fair versus fintech peers",
|
|
),
|
|
(
|
|
"compare growth metrics for these products against web2 companies",
|
|
"compare growth metrics for these products against web2 companies",
|
|
),
|
|
("thanks, that makes sense", None),
|
|
("what are your thoughts on lunch", None),
|
|
("how did you sleep", None),
|
|
("/paid_research find x402 evidence", None),
|
|
],
|
|
)
|
|
def test_extract_auto_smart_research_goal(message, expected):
|
|
assert extract_auto_smart_research_goal(
|
|
message,
|
|
mention_aliases=["@lipleowallet0622183538bot", "@leo"],
|
|
) == expected
|
|
|
|
|
|
def test_extract_auto_smart_research_followup_goal_uses_previous_market_question():
|
|
previous = "what is the volume and fdv of omnipair avici umbra? should i buy them yes or no"
|
|
goal = extract_auto_smart_research_followup_goal("check it yourself", previous)
|
|
|
|
assert previous in goal
|
|
assert "Use current public sources" in goal
|
|
assert "do not provide personalized financial advice" in goal
|
|
|
|
|
|
def test_extract_auto_smart_research_followup_goal_ignores_context_without_research_intent():
|
|
assert extract_auto_smart_research_followup_goal("check it yourself", "thanks, that makes sense") is None
|
|
|
|
|
|
def test_market_data_token_extraction_maps_natural_market_question():
|
|
message = "what is the volume and fdv of omnipair avici umbra? should i buy them yes or no"
|
|
|
|
assert extract_market_data_tokens(message) == ["OMFG", "AVICI", "UMBRA"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("message", "expected"),
|
|
[
|
|
("what is the volume and fdv of omnipair avici umbra? should i buy them yes or no", True),
|
|
("show me price and liquidity for AVICI", True),
|
|
("what are the current discussions about MetaDAO Ranger Finance on Twitter?", False),
|
|
("who was at fault for Ranger according to Twitter?", False),
|
|
("how much revenue does MetaDAO make today?", False),
|
|
],
|
|
)
|
|
def test_should_attach_structured_market_context_only_for_market_data_intent(message, expected):
|
|
assert should_attach_structured_market_context(message) is expected
|
|
|
|
|
|
def test_market_data_context_formats_dexscreener_fdv_volume_liquidity():
|
|
context = format_price_context(
|
|
{
|
|
"provider": "dexscreener",
|
|
"result": (
|
|
"Live market data for OMFG | source: DexScreener | price: $0.10 | "
|
|
"FDV: $1.20M | 24h volume: $52.00K | liquidity: $101.00K"
|
|
),
|
|
},
|
|
"OMFG",
|
|
)
|
|
|
|
assert "FDV: $1.20M" in context
|
|
assert "24h volume: $52.00K" in context
|
|
|
|
|
|
def test_smart_research_command_names_are_safe_for_telegram_handlers():
|
|
assert smart_research_command_names(
|
|
[
|
|
"/smart_research",
|
|
"/paid_research",
|
|
"/paid_research@lipleowallet0622183538bot",
|
|
"not_a_command",
|
|
"/bad-name",
|
|
"/",
|
|
]
|
|
) == ["paid_research", "smart_research"]
|
|
|
|
|
|
@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"),
|
|
({"synthesis": {"decision": {"reply": "smart research reply"}}}, "smart research 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
|
|
|
|
|
|
def test_extract_chat_proxy_reply_never_displays_operator_claim_fields():
|
|
payload = {
|
|
"ok": False,
|
|
"status": "payment_authorization_required",
|
|
"strongestClaimAllowed": (
|
|
"Leo smart research can select sponsor-agent-research on Devnet, "
|
|
"but did not attempt payment because the call was not fully authorized."
|
|
),
|
|
"exactBlocker": "smart_research_paid_execution_not_allowed",
|
|
"nextExactAction": "POST again with allow_paid_execution=true.",
|
|
}
|
|
|
|
assert extract_chat_proxy_reply(payload) is None
|
|
|
|
|
|
def test_extract_chat_proxy_reply_uses_clean_route_reply_for_tool_failures():
|
|
payload = {
|
|
"ok": False,
|
|
"status": "payment_authorization_required",
|
|
"reply": "I tried to use my paid research tool, but it was not available. No funds moved.",
|
|
"strongestClaimAllowed": "Internal status text that must stay out of Telegram.",
|
|
}
|
|
|
|
assert (
|
|
extract_chat_proxy_reply(payload)
|
|
== "I tried to use my paid research tool, but it was not available. No funds moved."
|
|
)
|