57 lines
2.2 KiB
Plaintext
57 lines
2.2 KiB
Plaintext
# -*- coding: utf-8 -*-
|
||
"""日志契约样例：复制为 tests/test_logging_contract.py 后按业务改入口。
|
||
|
||
验证主任务 mock/离线路径写入统一日志事件词（见 development/LOGGING.md §3）。
|
||
默认套件不收集本文件（.sample）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
import logging
|
||
import unittest
|
||
from contextlib import redirect_stderr, redirect_stdout
|
||
from unittest.mock import patch
|
||
|
||
# 复制到 tests/ 根目录后取消下一行注释，并按技能改 import
|
||
# from _support import IsolatedDataRoot
|
||
|
||
|
||
class TestLoggingContractSample(unittest.TestCase):
|
||
"""示范：捕获 skill logger，断言关键事件词出现。"""
|
||
|
||
def test_main_path_emits_task_start_and_terminal_event(self) -> None:
|
||
self.skipTest(
|
||
"复制本文件为 tests/test_logging_contract.py 后："
|
||
"用 IsolatedDataRoot + patch 域 RPA，调用 cmd_run/cmd_add_one，"
|
||
"对 logging.Handler 断言消息含 task_start 与 task_success|task_failed"
|
||
)
|
||
# 示例骨架（复制后实现）：
|
||
# with IsolatedDataRoot():
|
||
# log = logging.getLogger("openclaw.skill.track_fund_progress_default")
|
||
# records: list[str] = []
|
||
# class _H(logging.Handler):
|
||
# def emit(self, record):
|
||
# records.append(record.getMessage())
|
||
# h = _H()
|
||
# log.addHandler(h)
|
||
# try:
|
||
# with patch(...), redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
|
||
# rc = cmd_add_one(...) # 或 cmd_run
|
||
# finally:
|
||
# log.removeHandler(h)
|
||
# blob = "\n".join(records)
|
||
# self.assertIn("task_start", blob)
|
||
# self.assertTrue("task_failed" in blob or "task_success" in blob)
|
||
# if "task_failed" in blob:
|
||
# self.assertIn("error_code=", blob)
|
||
|
||
def test_failure_path_logs_error_code_before_print(self) -> None:
|
||
self.skipTest(
|
||
"复制后：注入业务错误（如 ELEMENT_NOT_FOUND），断言 logger 含 "
|
||
"task_failed error_code=...，且 stdout 含 ERROR:CODE"
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|