108 lines
3.5 KiB
Python
108 lines
3.5 KiB
Python
"""任务性能计时测试，不连接设备。"""
|
|
|
|
import json
|
|
import logging
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from src.performance_timing import (
|
|
TaskPerformanceTrace,
|
|
configure_performance_logging,
|
|
current_performance_trace,
|
|
)
|
|
|
|
|
|
class MutableClock:
|
|
def __init__(self) -> None:
|
|
self.value = 10.0
|
|
|
|
def __call__(self) -> float:
|
|
return self.value
|
|
|
|
|
|
class TaskPerformanceTraceTest(unittest.TestCase):
|
|
def test_stages_are_monotonic_scoped_and_use_only_safe_fields(self):
|
|
clock = MutableClock()
|
|
records = []
|
|
trace = TaskPerformanceTrace(monotonic=clock, sink=records.append)
|
|
|
|
with trace.activate():
|
|
self.assertIs(current_performance_trace(), trace)
|
|
with trace.stage("admin_claim_request"):
|
|
clock.value = 10.125
|
|
trace.bind_task("COL-001")
|
|
clock.value = 10.250
|
|
trace.checkpoint("end_to_end_total")
|
|
|
|
self.assertIsNone(current_performance_trace())
|
|
self.assertEqual(
|
|
records,
|
|
[
|
|
{
|
|
"task_id": "COL-001",
|
|
"operation": "admin_claim_request",
|
|
"duration_ms": 125,
|
|
"result": "succeeded",
|
|
},
|
|
{
|
|
"task_id": "COL-001",
|
|
"operation": "end_to_end_total",
|
|
"duration_ms": 250,
|
|
"result": "succeeded",
|
|
},
|
|
],
|
|
)
|
|
self.assertEqual(
|
|
set(records[0]),
|
|
{"task_id", "operation", "duration_ms", "result"},
|
|
)
|
|
|
|
def test_negative_duration_is_clamped_and_cross_task_reuse_is_rejected(self):
|
|
records = []
|
|
trace = TaskPerformanceTrace(sink=records.append)
|
|
trace.bind_task("COL-001")
|
|
trace.record("sqlite_local_save", -1, "succeeded")
|
|
|
|
self.assertEqual(records[0]["duration_ms"], 0)
|
|
with self.assertRaisesRegex(RuntimeError, "不能跨任务"):
|
|
trace.bind_task("COL-002")
|
|
|
|
def test_failed_stage_does_not_log_exception_details(self):
|
|
records = []
|
|
trace = TaskPerformanceTrace(sink=records.append)
|
|
trace.bind_task("COL-SECRET")
|
|
|
|
with self.assertRaisesRegex(RuntimeError, "token-secret"):
|
|
with trace.stage("open_url"):
|
|
raise RuntimeError("token-secret")
|
|
|
|
self.assertEqual(records[0]["result"], "failed")
|
|
self.assertNotIn("token-secret", str(records[0]))
|
|
|
|
def test_configured_jsonl_log_can_be_viewed_and_has_safe_schema(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
path = configure_performance_logging(Path(temporary))
|
|
trace = TaskPerformanceTrace()
|
|
trace.bind_task("COL-001")
|
|
trace.record("first_app_current", 0.125, "succeeded")
|
|
|
|
logger = logging.getLogger("cmautobuy.performance")
|
|
for handler in logger.handlers:
|
|
handler.flush()
|
|
record = json.loads(path.read_text(encoding="utf-8").strip())
|
|
self.assertEqual(record["duration_ms"], 125)
|
|
self.assertEqual(
|
|
set(record),
|
|
{"task_id", "operation", "duration_ms", "result"},
|
|
)
|
|
|
|
for handler in list(logger.handlers):
|
|
if getattr(handler, "baseFilename", None) == str(path.resolve()):
|
|
logger.removeHandler(handler)
|
|
handler.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|