Kris
b438af3ba8
本提交新增了全渠道SDK核心模块： 1. 异步锁、目标解析、动作调度等基础工具 2. 消息动作注册与统一调度系统 3. 测试套件与契约测试框架 4. 完整的目标解析流水线与工具函数 5. 资源依赖注入与生命周期管理
33 lines
874 B
Python
33 lines
874 B
Python
from dataclasses import dataclass
|
|
from typing import Any, Callable
|
|
|
|
|
|
def install_contract_suite(
|
|
channel_id: str,
|
|
plugin_factory: Callable[[], Any],
|
|
):
|
|
def _decorator(cls):
|
|
cls.make_plugin = staticmethod(plugin_factory)
|
|
return cls
|
|
return _decorator
|
|
|
|
|
|
@dataclass
|
|
class SendTestHarness:
|
|
plugin: Any
|
|
to: str = "test_target"
|
|
sent_count: int = 0
|
|
|
|
async def run_send_text(self, content: str, **kwargs):
|
|
msg_id = await self.plugin.send_text(self.to, content, **kwargs)
|
|
self.sent_count += 1
|
|
return msg_id
|
|
|
|
async def run_send_media(self, media_url: str, media_type: str = "image", **kwargs):
|
|
msg_id = await self.plugin.send_media(self.to, media_url, media_type, **kwargs)
|
|
self.sent_count += 1
|
|
return msg_id
|
|
|
|
@property
|
|
def sent_messages(self):
|
|
return self.plugin._sent |