multi-proto-agent 0.1.5

用统一的 “Player” 抽象管理 TCP / WebSocket（WS/WSS）连接、请求发送、响应接收与断言流程。

Multi-Proto-Agent

多协议通信工具库，面向自动化测试与高并发连接场景。
用统一的 Player 抽象管理 TCP / WebSocket（WS/WSS）连接、请求发送、响应接收与断言流程。

---

1. 适合谁用

• 需要在 Python 中统一管理 TCP/WS 长连接
• 需要可插拔的 protobuf 报文打包/解包策略
• 需要在测试框架里做响应断言与参数提取

---

2. 安装

pip install multi-proto-agent

或源码安装：

git clone https://gitee.com/flinttina/multi-proto-agent.git
cd multi-proto-agent
pip install -e .

---

3. 快速开始

from multi_proto_agent.utils.player import Player
from multi_proto_agent.utils.protocol_type import ProtocolType

player = Player(
    account_id="u001",
    protocol_type=ProtocolType.WS,  # 也支持 "ws"
)

player.set_ap_address("ws://127.0.0.1:20000")
player.set_secret_key("your_secret")
player.connect()

# req_obj 为 protobuf 消息对象
# ok = player.send_request("Demo.ReqPing", req_obj)

rsp_data = player.receive_response()  # 默认返回 bytes
print(rsp_data)

player.tear_down()

---

4. 核心概念

4.1 协议类型

from multi_proto_agent.utils.protocol_type import ProtocolType

ProtocolType.TCP
ProtocolType.WS
ProtocolType.WSS

Player 支持传字符串或枚举：

• Player(protocol_type="ws")
• Player(protocol_type=ProtocolType.WS)

4.2 编解码策略（关键）

框架默认协议中立，不绑定业务包头：

• 未传 msg_packer：发送裸 protobuf bytes
• 未传 msg_unpacker：接收原始 bytes

有业务外层协议时（如项目自定义 BaseMsg），在初始化注入：

player = Player(
    account_id="u001",
    protocol_type=ProtocolType.WS,
    msg_packer=my_packer,
    msg_unpacker=my_unpacker,
)

策略对象约定：

• pack(proto_msg_name, obj_to_send, trace_id="") -> bytes
• unpack(rsp_data: bytes) -> Any

---

5. 对外 API（稳定优先）

5.1 player.py

• Player(account_id, role_name=None, role_uid=None, protocol_type='ws', heartbeat_interval=30, heartbeat_class_name="", heartbeat_obj=None, msg_packer=None, msg_unpacker=None)

  • 作用：创建连接与收发控制对象，统一管理 TCP/WS/WSS 通信、心跳和编解码策略。
  • 入参：
    • account_id：账号标识（必填）。
    • role_name / role_uid：角色信息（可选）。
    • protocol_type：协议类型，支持 tcp / ws / wss 或 ProtocolType 枚举。
    • heartbeat_interval：心跳发送间隔（秒）。
    • heartbeat_class_name / heartbeat_obj：心跳消息名与对象（配置后自动心跳）。
    • msg_packer / msg_unpacker：自定义打包/解包策略（可选）。
  • 示例：

    from multi_proto_agent.utils.player import Player
    from multi_proto_agent.utils.protocol_type import ProtocolType
    
    player = Player(account_id="u001", protocol_type=ProtocolType.WS)
    

• connect(max_retry_times=3)

  • 作用：根据 protocol_type 建立连接，并初始化响应队列；连接成功返回 True。
  • 入参：
    • max_retry_times：连接失败后的最大重试次数，默认 3。
  • 示例：

    player.set_ap_address("ws://127.0.0.1:20000")
    player.set_secret_key("your_secret")
    player.connect(max_retry_times=5)
    

• send_request(req_msg_name, req_obj, is_heartbeat=False)

  • 作用：发送请求消息；非心跳请求会自动生成并记录 trace_id。
  • 入参：
    • req_msg_name：protobuf 消息名（如 Demo.ReqPing）。
    • req_obj：protobuf 请求对象实例。
    • is_heartbeat：是否为心跳消息，默认 False。
  • 示例：

    ok = player.send_request("Demo.ReqPing", req_obj)
    

• receive_response()

  • 作用：从内部响应队列非阻塞读取一条消息；队列为空返回 None。
  • 入参：无。
  • 示例：

    rsp_data = player.receive_response()
    if rsp_data is not None:
        print("got response")
    

• tear_down()

  • 作用：关闭连接并清理资源（含心跳任务）。
  • 入参：无。
  • 示例：

    player.tear_down()
    

• set_ap_address(ap_address)

  • 作用：设置服务端地址（连接前必设）。
  • 入参：
    • ap_address：地址字符串；TCP 格式通常为 ip:port，WS/WSS 为 URL。
  • 示例：

    player.set_ap_address("127.0.0.1:9000")  # tcp
    

• set_secret_key(secret_key)

  • 作用：设置连接鉴权所需密钥（连接前必设）。
  • 入参：
    • secret_key：密钥字符串。
  • 示例：

    player.set_secret_key("my_secret_key")
    

5.2 translator.py

• handle_send_data(proto_msg_name, obj_to_send, trace_id="", packer=None)

  • 作用：将请求对象转为可发送的 bytes；若传入 packer 则走自定义打包。
  • 入参：
    • proto_msg_name：消息名（如 Demo.ReqPing）。
    • obj_to_send：protobuf 对象。
    • trace_id：链路标识（可选）。
    • packer：实现 pack(...) 的策略对象（可选）。
  • 示例：

    req_bytes = handle_send_data("Demo.ReqPing", req_obj)
    

• handle_rsp_data(rsp_data, unpacker=None)

  • 作用：处理响应数据；未传 unpacker 时原样返回 bytes。
  • 入参：
    • rsp_data：原始响应 bytes。
    • unpacker：实现 unpack(...) 的策略对象（可选）。
  • 示例：

    rsp_obj = handle_rsp_data(rsp_bytes, unpacker=my_unpacker)
    

• deserialize_proto_object(proto_msg_name, bytes_data)

  • 作用：按消息名动态反序列化 bytes 为 protobuf 对象。
  • 入参：
    • proto_msg_name：消息名（模块名需在环境变量中可映射到 protobuf 包）。
    • bytes_data：待反序列化的二进制数据。
  • 示例：

    proto_obj = deserialize_proto_object("Demo.RspPing", rsp_bytes)
    

• get_string_of_req_obj(req_obj)

  • 作用：将 protobuf 对象转为 JSON 字符串，便于日志打印。
  • 入参：
    • req_obj：protobuf 对象。
  • 示例：

    print(get_string_of_req_obj(req_obj))
    

5.3 assert_util.py

• is_player_passed(player, assertion_dict, trace_id="", trace_desc="", max_timeout=5, is_isNotGot_assertion=False)

  • 作用：消费 player 的响应队列并执行断言规则，返回 AssertResult（是否通过、提取值、断言信息）。
  • 入参：
    • player：Player 实例。
    • assertion_dict：断言配置（按消息名组织 assertion_rule_list / value_to_get_list）。
    • trace_id / trace_desc：断言失败时上报追踪信息（可选）。
    • max_timeout：最大等待时长（秒）。
    • is_isNotGot_assertion：是否启用 isNotGot 语义控制（通常默认即可）。
  • 示例：

    result = is_player_passed(player, assertion_dict, max_timeout=10)
    print(result.get_is_player_pass(), result.get_value_dict())
    

• get_supported_assertions()

  • 作用：返回当前支持的断言类型列表（如 ==、in、isNotNone、isGot 等）。
  • 入参：无。
  • 示例：

    print(get_supported_assertions())
    

5.4 config_util.py

• set_config(env_type, config_path=None)

  • 作用：从环境配置文件读取指定环境并写入环境变量。
  • 入参：
    • env_type：环境标识（如 test、prod）。
    • config_path：环境配置 YAML 路径。
  • 示例：

    set_config(env_type="test", config_path="./config/env_config.yaml")
    

• set_protos_config(config_path=None)

  • 作用：读取 protobuf 相关配置并写入环境变量（用于动态反序列化映射）。
  • 入参：
    • config_path：protos_config.yaml 路径。
  • 示例：

    set_protos_config("./config/protos_config.yaml")
    

• add_python_protos_to_path(python_protos_path=None)

  • 作用：将 protobuf 生成目录及其子目录加入 sys.path，便于动态导入。
  • 入参：
    • python_protos_path：protobuf Python 文件根目录。
  • 示例：

    add_python_protos_to_path("./python_protos")
    

• find_root_directory(start_file=None, root_marker_file="i_am_the_root.md")

  • 作用：向上查找项目根目录（通过根标记文件定位）。
  • 入参：
    • start_file：起始路径或文件（默认当前工作目录）。
    • root_marker_file：根标记文件名，默认 i_am_the_root.md。
  • 示例：

    root_dir = find_root_directory(__file__)
    

5.5 其他常用工具

• trace_id_util.py

  • generate_trace_id()
    • 作用：生成追踪 ID（十六进制字符串）。
    • 入参：无。
    • 示例：trace_id = generate_trace_id()
  • send_trace_id(url, trace_id, trace_desc)
    • 作用：将 trace 信息上报到指定 URL（GET 请求）。
    • 入参：url 上报地址，trace_id 追踪 ID，trace_desc 描述信息。
    • 示例：resp = send_trace_id(report_url, trace_id, "断言失败")

• random_util.py

  • 作用：提供随机文本/随机数值生成能力（用于构造测试数据）。
  • 示例：

    # 具体函数名以 random_util.py 中导出为准
    from multi_proto_agent.utils import random_util
    

• logger_config.py

  • setup_logging(level=None, log_dir="./logs")
    • 作用：初始化日志系统（控制台 + 文件）。
    • 入参：level 日志级别，log_dir 日志目录。
    • 示例：setup_logging(level="INFO", log_dir="./logs")
  • get_logger(name=None)
    • 作用：获取 logger 实例；未配置时会自动初始化。
    • 入参：name logger 名称（可选）。
    • 示例：logger = get_logger(__name__)

• protocol_type.py

  • ProtocolType
    • 作用：协议类型枚举，包含 TCP / WS / WSS。
    • 示例：ProtocolType.WS
  • ProtocolType.from_value(value)
    • 作用：将字符串或枚举值规范化为 ProtocolType。
    • 入参：value，如 "ws" 或 ProtocolType.WS。
    • 示例：pt = ProtocolType.from_value("wss")

---

6. 常见使用模式

6.1 只做通信（最简）

• 使用 Player + ProtocolType
• 不注入 msg_packer/msg_unpacker
• 自己处理收到的裸 bytes

6.2 业务协议封装（推荐）

• 项目侧实现 packer/unpacker
• 在 Player 初始化注入
• 断言和日志统一处理解包后的消息

6.3 自动化断言

• 用 is_player_passed(...) 消费 player.rsp_queue
• 通过 assertion_dict 声明断言规则与提取字段

---

7. FAQ

Q1: 为什么断言拿不到字段？

通常因为未注入 msg_unpacker，响应仍是裸 bytes。
请先在 Player(...) 注入项目侧解包策略。

Q2: 协议类型大小写敏感吗？

不敏感。ProtocolType.from_value() 会统一归一化并校验。

Q3: 何时直接用 TcpClient/WebSocketClient？

当你需要绕过 Player，做更底层的连接控制时。常规场景优先用 Player。

---

8. 文档索引

• 使用者文档：当前文件 README.md
• 开发者文档：README.dev.md（内部结构、分层、关键实现）

---

9. 许可证

MIT License