"""单元测试:§24 v3 rotation-only compact 检测 测试 _get_recent_gateway_logs 和 _check_compact_in_progress_gateway。 用 tmp_path 构造 mock gateway 日志文件。 """ import json import os from datetime import datetime, timedelta, timezone from pathlib import Path import pytest from src.daemon.spawner import AgentSpawner # ── helpers ── _SESSION_KEY = "agent:pangtong-fujunshi:main" _TODAY_STR = datetime.now().strftime("%Y-%m-%d") def _make_rotation_event(session_key: str, ts: datetime) -> dict: """构造一条 rotation 日志事件""" return { "time": ts.isoformat(), "message": f"[compaction] rotated active transcript after compaction (sessionKey={session_key})", } def _make_other_event(session_key: str, ts: datetime, msg: str = "something else") -> dict: """构造一条普通日志事件""" return { "time": ts.isoformat(), "message": f"{msg} (sessionKey={session_key})", } def _write_log(tmp_path: Path, date_str: str, lines: list[dict]): """写 mock 日志文件""" log_file = tmp_path / f"openclaw-{date_str}.log" with open(log_file, "w") as f: for obj in lines: f.write(json.dumps(obj, ensure_ascii=False) + "\n") @pytest.fixture(autouse=True) def _set_log_dir(tmp_path, monkeypatch): """每个测试自动设置 OPENCLAW_LOG_DIR 到 tmp_path""" monkeypatch.setenv("OPENCLAW_LOG_DIR", str(tmp_path)) # ── 测试用例 ── class TestCheckCompactInProgress: """§24 v3: _check_compact_in_progress_gateway 单元测试""" def test_rotation_within_window_returns_true(self, tmp_path): """TC1: rotation 事件在窗口内 → True""" now = datetime.now(timezone.utc) recent = now - timedelta(seconds=30) _write_log(tmp_path, _TODAY_STR, [_make_rotation_event(_SESSION_KEY, recent)]) assert AgentSpawner._check_compact_in_progress_gateway(_SESSION_KEY) is True def test_rotation_outside_window_returns_false(self, tmp_path): """TC2: rotation 事件超出窗口 → False""" now = datetime.now(timezone.utc) old = now - timedelta(seconds=200) _write_log(tmp_path, _TODAY_STR, [_make_rotation_event(_SESSION_KEY, old)]) assert AgentSpawner._check_compact_in_progress_gateway(_SESSION_KEY) is False def test_no_rotation_event_returns_false(self, tmp_path): """TC3: 无 rotation 事件 → False""" now = datetime.now(timezone.utc) _write_log(tmp_path, _TODAY_STR, [ _make_other_event(_SESSION_KEY, now, "model.completed"), ]) assert AgentSpawner._check_compact_in_progress_gateway(_SESSION_KEY) is False def test_log_file_not_exists_returns_false(self, tmp_path): """TC4: 日志文件不存在 → False""" # tmp_path 为空目录,无日志文件 assert AgentSpawner._check_compact_in_progress_gateway(_SESSION_KEY) is False def test_session_key_mismatch_returns_false(self, tmp_path): """TC5: sessionKey 不匹配 → False""" now = datetime.now(timezone.utc) recent = now - timedelta(seconds=10) other_key = "agent:simayi-challenger:main" _write_log(tmp_path, _TODAY_STR, [_make_rotation_event(other_key, recent)]) assert AgentSpawner._check_compact_in_progress_gateway(_SESSION_KEY) is False