| """Tests for Slack/Telegram bot common logic.""" |
|
|
| from __future__ import annotations |
|
|
| import asyncio |
|
|
| import pytest |
|
|
| from symbolic_recursion.bots.common import BotAgent, HELP_TEXT, format_reply |
| from symbolic_recursion.inference.chat_engine import ChatResponse |
|
|
|
|
| def test_help_command() -> None: |
| agent = BotAgent() |
| reply = asyncio.run(agent.handle("test", "user1", "/help")) |
| assert "Symbolic Recursive Engine" in reply |
| assert "/sensors" in reply |
|
|
|
|
| def test_session_id_unique_per_platform() -> None: |
| agent = BotAgent() |
| assert agent.session_id("slack", "U1", "C1") == "slack:C1:U1" |
| assert agent.session_id("telegram", "123", "456") == "telegram:456:123" |
|
|
|
|
| def test_format_reply_truncates() -> None: |
| result = ChatResponse( |
| reply="x" * 5000, |
| tools_used=["health"], |
| kernel_state={}, |
| model="test", |
| session_id="s1", |
| ) |
| out = format_reply(result) |
| assert len(out) <= 3900 |
| assert "truncated" in out |
|
|
|
|
| def test_clear_command() -> None: |
| agent = BotAgent() |
| reply = asyncio.run(agent.handle("test", "u", "/clear")) |
| assert "cleared" in reply.lower() |
|
|
|
|
| def test_sensors_command() -> None: |
| agent = BotAgent() |
| reply = asyncio.run(agent.handle("test", "u", "/sensors")) |
| assert "Q(t)" in reply or "Live sensors" in reply |