perf: 量化商品页打开耗时并去重检查 (#109)

client/test/test_performance_timing.py

@@ -0,0 +1,107 @@
+"""任务性能计时测试，不连接设备。"""
+
+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()