"""log_test.py —— 任务日志模块单元测试。 覆盖：open 建目录/写表头、事件 JSONL、遥测 CSV、telemetry_every_n 节流、 disabled 时零副作用、close 收尾、简洁任务简报（task_brief.log）。 """ import csv import json import pytest from task_runner.log import TaskLogConfig, TaskLogger, TelemetryRecord def _cfg(tmp_path, **kw): c = TaskLogConfig(dir=str(tmp_path), **kw) return c def test_open_creates_dir_and_csv_header(tmp_path): logger = TaskLogger(_cfg(tmp_path)) logger.open() logger.close() assert (tmp_path / "task_events.jsonl").exists() assert (tmp_path / "task_telemetry.csv").exists() with open(tmp_path / "task_telemetry.csv", encoding="utf-8") as f: header = next(csv.reader(f)) assert header[0] == "t_s" assert "state" in header assert "distance_to_target_m" in header def test_log_event_writes_jsonl(tmp_path): logger = TaskLogger(_cfg(tmp_path)) logger.open() logger.log_event(1000, "state", from_state="wait", to_state="arm_takeoff", reason="收到地面坐标") logger.log_event(2000, "emergency", reason="RC 接管") logger.close() with open(tmp_path / "task_events.jsonl", encoding="utf-8") as f: lines = [json.loads(x) for x in f if x.strip()] assert len(lines) == 2 assert lines[0]["type"] == "state" assert lines[0]["from_state"] == "wait" assert lines[0]["to_state"] == "arm_takeoff" assert lines[0]["ts_ms"] == 1000 assert lines[0]["t_s"] == 0.0 # 首个事件 t_s = 0 assert lines[1]["t_s"] == 1.0 # 第二个事件 t_s = 1.0s def test_log_telemetry_writes_csv_rows(tmp_path): logger = TaskLogger(_cfg(tmp_path)) logger.open() rec = TelemetryRecord(state="goto", pos_valid=True, latitude_deg=47.39, distance_to_target_m=33.5, mode="VELOCITY_HEADING", yaw_rate_dps=-30.0, vx_body_mps=1.7) logger.log_telemetry(1000, rec) logger.log_telemetry(1100, rec) logger.close() with open(tmp_path / "task_telemetry.csv", encoding="utf-8") as f: rows = list(csv.DictReader(f)) assert len(rows) == 2 assert rows[0]["state"] == "goto" assert rows[0]["pos_valid"] == "True" assert rows[0]["distance_to_target_m"] == "33.5" assert rows[0]["t_s"] == "0.0" assert rows[1]["t_s"] == "0.1" def test_telemetry_throttled_by_every_n(tmp_path): logger = TaskLogger(_cfg(tmp_path, telemetry_every_n=5)) logger.open() rec = TelemetryRecord(state="goto") for i in range(12): logger.log_telemetry(1000 + i * 100, rec) logger.close() with open(tmp_path / "task_telemetry.csv", encoding="utf-8") as f: rows = list(csv.DictReader(f)) assert len(rows) == 2 # 第 5、10 次各记一次 def test_disabled_logs_nothing(tmp_path): logger = TaskLogger(_cfg(tmp_path, enabled=False)) logger.open() logger.log_event(1000, "state", from_state="wait", to_state="arm_takeoff") logger.log_telemetry(1000, TelemetryRecord(state="goto")) logger.close() # 未 open（disabled）→ 不建目录、不写文件 assert not (tmp_path / "task_events.jsonl").exists() assert not (tmp_path / "task_telemetry.csv").exists() def test_log_before_open_is_noop(tmp_path): logger = TaskLogger(_cfg(tmp_path)) logger.log_event(1000, "state", from_state="wait", to_state="arm_takeoff") logger.log_telemetry(1000, TelemetryRecord(state="goto")) logger.close() assert not (tmp_path / "task_events.jsonl").exists() def test_append_keeps_existing_rows(tmp_path): """重复 open（追加模式）不清空既有内容，表头不重复写。""" logger = TaskLogger(_cfg(tmp_path)) logger.open() logger.log_telemetry(1000, TelemetryRecord(state="wait")) logger.close() logger2 = TaskLogger(_cfg(tmp_path)) logger2.open() logger2.log_telemetry(2000, TelemetryRecord(state="goto")) logger2.close() with open(tmp_path / "task_telemetry.csv", encoding="utf-8") as f: rows = list(csv.DictReader(f)) assert len(rows) == 2 assert rows[0]["state"] == "wait" assert rows[1]["state"] == "goto" # --- 简洁任务简报（task_brief.log）------------------------------------------- def test_brief_log_lines_and_summary(tmp_path): logger = TaskLogger(_cfg(tmp_path)) logger.open() logger.log_event(1000, "state", from_state="wait", to_state="wait", reason="任务启动") logger.log_event(2000, "state", from_state="wait", to_state="arm_takeoff", reason="收到地面坐标") logger.log_event(3000, "command", name="rtl") logger.log_event(4000, "fc_status", severity="WARNING", text="Low battery\t") logger.log_event(5000, "takeoff", altitude_m=5.0) logger.log_event(6000, "rtl") logger.log_telemetry(1000, TelemetryRecord( state="wait", relative_altitude_m=0.0, distance_to_target_m=33.5)) logger.log_telemetry(1100, TelemetryRecord( state="arm_takeoff", relative_altitude_m=5.2, distance_to_target_m=20.0)) logger.log_telemetry(1200, TelemetryRecord( state="arm_takeoff", relative_altitude_m=5.0, distance_to_target_m=float("inf"))) logger.close() text = (tmp_path / "task_brief.log").read_text(encoding="utf-8") assert "任务开始" in text assert "[00:00.0] 任务启动（当前: 待命）" in text assert "[00:01.0] 阶段: 待命 → 起飞（收到地面坐标）" in text assert "[00:02.0] 地面站命令: rtl" in text assert "[00:03.0] 飞控 WARNING: Low battery" in text # 尾随 tab 已 strip assert "[00:04.0] 起飞至 5.0 m" in text assert "[00:05.0] 开始返航" in text assert "本次运行汇总" in text assert "时长 0.2s" in text # 最后一行遥测的 t_s assert "最高 5.2m" in text assert "最近距目标 20.0m" in text # inf 不计入 assert "事件 6 条" in text assert "状态链 待命→起飞" in text def test_brief_required_no_file_when_disabled_or_closed(tmp_path): logger = TaskLogger(_cfg(tmp_path, enabled=False)) logger.open() logger.log_event(1000, "rtl") logger.close() assert not (tmp_path / "task_brief.log").exists() # --- 遥测落盘节流（2026-10-15：主循环 20Hz 不得每行同步 flush）---------------- def test_telemetry_buffered_until_flush_interval(tmp_path): """telemetry_flush_ms 内只写不 flush；close 兜底落盘，一行不丢。""" logger = TaskLogger(_cfg(tmp_path, telemetry_flush_ms=1000)) logger.open() rec = TelemetryRecord(state="track") logger.log_telemetry(1000, rec) # 首行：落盘一次并打戳 for i in range(1, 19): # 其后 19 行都在 1s 内 → 不再 flush logger.log_telemetry(1000 + i * 50, rec) logger.close() with open(tmp_path / "task_telemetry.csv", encoding="utf-8") as f: rows = list(csv.DictReader(f)) assert len(rows) == 19 def test_telemetry_flush_zero_flushes_every_row(tmp_path): """telemetry_flush_ms=0 保留旧的「每行立即 flush」行为（兼容/调试）。""" logger = TaskLogger(_cfg(tmp_path, telemetry_flush_ms=0)) logger.open() logger.log_telemetry(1000, TelemetryRecord(state="goto")) assert logger._last_tel_flush_ms == 1000 logger.close() def test_events_still_flushed_immediately(tmp_path): """事件（阶段切换）仍即时落盘：地面站页面实时性不受遥测节流影响。""" logger = TaskLogger(_cfg(tmp_path, telemetry_flush_ms=10_000)) logger.open() logger.log_event(1000, "state", from_state="goto", to_state="search") with open(tmp_path / "task_events.jsonl", encoding="utf-8") as f: lines = [json.loads(x) for x in f if x.strip()] assert len(lines) == 1 logger.close()