RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
71 lines
2.7 KiB
Python
71 lines
2.7 KiB
Python
"""Agent schedule-tool adapter tests with an isolated fake scheduler."""
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
|
|
BASE = Path(__file__).resolve().parents[1]
|
|
sys.path[:0] = [str(BASE / "bot-utility"), str(BASE)]
|
|
|
|
from agent import ToolContext, ToolExecutor
|
|
|
|
|
|
class FakeScheduler:
|
|
def __init__(self):
|
|
self.created = None
|
|
self.cancelled = None
|
|
def create_from_params(self, ctx, params):
|
|
self.created = (ctx, params)
|
|
return {"id": "abc12345", "message": params["message"], "kind": params["kind"], "next_run_text": "今晚八点"}
|
|
def list_for_context(self, ctx):
|
|
return [{"id": "abc12345", "message": "吃饭", "kind": "once", "next_run_text": "今晚八点"}]
|
|
def cancel_for_context(self, ctx, task_id):
|
|
self.cancelled = (ctx, task_id)
|
|
return {"id": "abc12345", "message": "吃饭"}
|
|
|
|
|
|
class ScheduleToolTests(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self):
|
|
self.fake = FakeScheduler()
|
|
module = SimpleNamespace(scheduler=self.fake, format_task=lambda task: f"{task['id']} {task['message']} {task.get('next_run_text', '')}")
|
|
self.executor = ToolExecutor(modules={"schedule": module})
|
|
self.ctx = ToolContext(
|
|
channel=SimpleNamespace(name="tg"), msg_type="private",
|
|
user_id=1, target_id=10001, is_admin=False,
|
|
)
|
|
|
|
async def test_schema_exposes_schedule_to_normal_users(self):
|
|
names = [item["name"] for item in self.executor.prompt_schema(self.ctx)]
|
|
self.assertIn("schedule", names)
|
|
|
|
async def test_create_uses_current_conversation(self):
|
|
result = await self.executor.execute(
|
|
"schedule",
|
|
{"action": "create", "params": {"message": "吃饭", "kind": "once", "run_at": "2026-10-30T20:00:00+08:00"}},
|
|
self.ctx,
|
|
)
|
|
self.assertTrue(result.ok)
|
|
self.assertIs(self.fake.created[0], self.ctx)
|
|
self.assertEqual(self.fake.created[1]["message"], "吃饭")
|
|
|
|
async def test_list_and_cancel(self):
|
|
listed = await self.executor.execute("schedule", {"action": "list", "params": {}}, self.ctx)
|
|
self.assertIn("abc12345", listed.safe_message)
|
|
cancelled = await self.executor.execute(
|
|
"schedule", {"action": "cancel", "params": {"id": "abc"}}, self.ctx
|
|
)
|
|
self.assertTrue(cancelled.ok)
|
|
self.assertEqual(self.fake.cancelled, (self.ctx, "abc"))
|
|
|
|
async def test_cancel_rejects_extra_params(self):
|
|
result = await self.executor.execute(
|
|
"schedule", {"action": "cancel", "params": {"id": "abc", "uid": 2}}, self.ctx
|
|
)
|
|
self.assertFalse(result.ok)
|
|
self.assertEqual(result.code, "invalid_args")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|