first commit

tests/utils/test_call_timeline.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+import json
+import logging
+import os
+import tempfile
+import unittest
+
+from app.utils.call_timeline import CallTimeline
+
+
+class CallTimelineTests(unittest.TestCase):
+    def test_emit_writes_jsonl_with_relative_time(self) -> None:
+        logger = logging.getLogger("test.call_timeline")
+        logger.handlers = []
+        logger.addHandler(logging.NullHandler())
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            old_dir = os.environ.get("CALL_TIMELINE_DIR")
+            old_console = os.environ.get("CALL_TIMELINE_CONSOLE")
+            old_enabled = os.environ.get("CALL_TIMELINE_ENABLED")
+
+            os.environ["CALL_TIMELINE_DIR"] = tmpdir
+            os.environ["CALL_TIMELINE_CONSOLE"] = "0"
+            os.environ["CALL_TIMELINE_ENABLED"] = "1"
+
+            try:
+                timeline = CallTimeline(
+                    logger=logger,
+                    component="bridge",
+                    timeline_id="room-dev-123",
+                    protocol="PRT-1",
+                    room="room-dev-123",
+                    session_id="PRT-1",
+                    phone_number="31999999999",
+                    origin_unix_ms=1000,
+                )
+                timeline.emit("call_start", ok=True, nested={"a": 1})
+                timeline.emit("ready_sent")
+                self.assertTrue(timeline.flush(timeout=5.0))
+
+                with timeline.path.open("r", encoding="utf-8") as handle:
+                    lines = [json.loads(line) for line in handle if line.strip()]
+            finally:
+                if old_dir is None:
+                    os.environ.pop("CALL_TIMELINE_DIR", None)
+                else:
+                    os.environ["CALL_TIMELINE_DIR"] = old_dir
+
+                if old_console is None:
+                    os.environ.pop("CALL_TIMELINE_CONSOLE", None)
+                else:
+                    os.environ["CALL_TIMELINE_CONSOLE"] = old_console
+
+                if old_enabled is None:
+                    os.environ.pop("CALL_TIMELINE_ENABLED", None)
+                else:
+                    os.environ["CALL_TIMELINE_ENABLED"] = old_enabled
+
+        self.assertEqual(2, len(lines))
+        self.assertEqual("bridge", lines[0]["component"])
+        self.assertEqual("call_start", lines[0]["event"])
+        self.assertEqual("room-dev-123", lines[0]["timeline_id"])
+        self.assertEqual("PRT-1", lines[0]["protocol"])
+        self.assertEqual("31999999999", lines[0]["phone_number"])
+        self.assertTrue(lines[0]["ok"])
+        self.assertEqual({"a": 1}, lines[0]["nested"])
+        self.assertGreaterEqual(lines[0]["t_rel_ms"], 0)
+        self.assertEqual("ready_sent", lines[1]["event"])
+
+    def test_background_writer_preserves_order_for_many_events(self) -> None:
+        logger = logging.getLogger("test.call_timeline.order")
+        logger.handlers = [logging.NullHandler()]
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            old_dir = os.environ.get("CALL_TIMELINE_DIR")
+            old_enabled = os.environ.get("CALL_TIMELINE_ENABLED")
+            os.environ["CALL_TIMELINE_DIR"] = tmpdir
+            os.environ["CALL_TIMELINE_ENABLED"] = "1"
+            try:
+                timeline = CallTimeline(
+                    logger=logger,
+                    component="bridge",
+                    timeline_id="room-order",
+                    origin_unix_ms=1000,
+                )
+                for index in range(200):
+                    timeline.emit("tick", index=index)
+                self.assertTrue(timeline.flush(timeout=5.0))
+                with timeline.path.open("r", encoding="utf-8") as handle:
+                    lines = [json.loads(line) for line in handle if line.strip()]
+            finally:
+                if old_dir is None:
+                    os.environ.pop("CALL_TIMELINE_DIR", None)
+                else:
+                    os.environ["CALL_TIMELINE_DIR"] = old_dir
+                if old_enabled is None:
+                    os.environ.pop("CALL_TIMELINE_ENABLED", None)
+                else:
+                    os.environ["CALL_TIMELINE_ENABLED"] = old_enabled
+
+        self.assertEqual(list(range(200)), [line["index"] for line in lines])
+
+
+if __name__ == "__main__":
+    unittest.main()