RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
89 lines
3.1 KiB
Python
89 lines
3.1 KiB
Python
"""Akaros Mail REST and Agent tool tests; no real email is sent."""
|
|
|
|
import importlib
|
|
from pathlib import Path
|
|
import sys
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
BASE = Path(__file__).resolve().parents[1]
|
|
sys.path[:0] = [str(BASE / "bot-utility"), str(BASE)]
|
|
|
|
from agent import ToolContext, ToolExecutor
|
|
|
|
mail = importlib.import_module("for-bot_mail")
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, status=201, data=None):
|
|
self.status_code = status
|
|
self.headers = {"Content-Type": "application/json"}
|
|
self.data = data or {"message": "accepted"}
|
|
def json(self):
|
|
return self.data
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, response):
|
|
self.response = response
|
|
self.calls = []
|
|
def __enter__(self):
|
|
return self
|
|
def __exit__(self, *args):
|
|
return False
|
|
def request(self, method, url, **kwargs):
|
|
self.calls.append((method, url, kwargs))
|
|
return self.response
|
|
|
|
|
|
class AkarosMailTests(unittest.IsolatedAsyncioTestCase):
|
|
def test_send_uses_fixed_shirobana_address(self):
|
|
session = FakeSession(FakeResponse())
|
|
with patch("requests.Session", return_value=session), \
|
|
patch.object(mail, "API_KEY", "test-key"):
|
|
result = mail.send_email(
|
|
"user@example.test", "测试主题", "正文",
|
|
cc=["copy@example.test"], bcc=[])
|
|
self.assertEqual(result["message"], "accepted")
|
|
payload = session.calls[0][2]["json"]
|
|
self.assertEqual(payload["from"], "shirobana@akaros.si")
|
|
self.assertEqual(payload["to"], ["user@example.test"])
|
|
self.assertEqual(payload["cc"], ["copy@example.test"])
|
|
self.assertEqual(payload["text"], "正文")
|
|
|
|
def test_invalid_recipient_is_rejected(self):
|
|
with self.assertRaises(mail.MailInputError):
|
|
mail.send_email("not-an-email", "subject", "text")
|
|
|
|
async def test_agent_send_and_result_unknown(self):
|
|
module = SimpleNamespace(
|
|
send_email=lambda *args: {"message": "accepted"},
|
|
list_messages=lambda *args: [],
|
|
get_message=lambda value: {},
|
|
format_messages=lambda data: "没有找到邮件",
|
|
format_message=lambda data: "正文",
|
|
)
|
|
executor = ToolExecutor(modules={"akaros_mail": module})
|
|
ctx = ToolContext(channel=SimpleNamespace(name="tg"), msg_type="private", user_id=1, target_id=1)
|
|
result = await executor.execute(
|
|
"akaros_mail",
|
|
{"action": "send", "params": {"to": "user@example.test", "subject": "s", "text": "t"}},
|
|
ctx,
|
|
)
|
|
self.assertTrue(result.ok)
|
|
def fail_send(*args):
|
|
raise RuntimeError("connection lost")
|
|
module.send_email = fail_send
|
|
result = await executor.execute(
|
|
"akaros_mail",
|
|
{"action": "send", "params": {"to": "user@example.test", "subject": "s", "text": "t"}},
|
|
ctx,
|
|
)
|
|
self.assertEqual(result.code, "send_unknown")
|
|
self.assertFalse(result.retryable)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|