| |
| """Telemetry safety contract: append-only, fail-open, and control-path neutral. |
| |
| No live services and no FFmpeg. The important production guarantee is not |
| that storage can never fail, but that telemetry can fail/drop its own records |
| without raising into, blocking, or steering a stream. |
| """ |
| import json |
| import os |
| import pathlib |
| import sys |
| import tempfile |
| import time |
|
|
| ROOT = pathlib.Path(__file__).resolve().parent |
| sys.path.insert(0, str(ROOT)) |
|
|
| from telemetry import Telemetry, http_status_from_exception |
|
|
|
|
| passed = failed = 0 |
|
|
|
|
| def check(name, condition, detail=""): |
| global passed, failed |
| if condition: |
| passed += 1 |
| print("PASS %s" % name) |
| else: |
| failed += 1 |
| print("FAIL %s %s" % (name, detail)) |
|
|
|
|
| def wait_records(store, count, timeout=3.0): |
| deadline = time.time() + timeout |
| rows = [] |
| while time.time() < deadline: |
| rows = store.read_records(100) |
| if len(rows) >= count: |
| return rows |
| time.sleep(0.02) |
| return rows |
|
|
|
|
| with tempfile.TemporaryDirectory() as tmp: |
| path = pathlib.Path(tmp) / "telemetry" / "telemetry.jsonl" |
| store = Telemetry(path) |
| key = store.start_session("alice", { |
| "mode": "copy", |
| "url": "https://signed.example/private?token=should-not-persist", |
| "nested": {"stream_key": "should-not-persist", "room_id": "123"}, |
| }) |
| check("session starts", bool(key)) |
| check("observe accepts ffmpeg progress", store.observe(key, { |
| "ffmpeg_speed": 0.80, "ffmpeg_fps": 30.0, "source_bps": 900_000, |
| "output_seconds": 12.0, |
| })) |
| check("observe accepts recovered speed", store.observe(key, { |
| "ffmpeg_speed": 1.01, "ffmpeg_fps": 30.0, "source_bps": 2_400_000, |
| "output_seconds": 27.0, |
| })) |
|
|
| class Response: |
| status_code = 403 |
|
|
| class HttpFailure(Exception): |
| response = Response() |
|
|
| check("http status extracted without message", http_status_from_exception(HttpFailure()) == 403) |
| check("source-open event accepted", store.event(key, "alice", "SOURCE_OPEN_ERROR", { |
| "stage": "initial_open", "error_type": "HTTPError", "http_status": 403, |
| "url": "https://signed.example/do-not-store", |
| })) |
| check("resolution event accepted", store.event(key, "alice", "RESOLUTION_CHANGE", { |
| "old_width": 720, "old_height": 1280, "new_width": 428, "new_height": 856, |
| })) |
| check("checkpoint accepted", store.checkpoint(key, "unit-test")) |
| check("finish accepted", store.finish(key, "unit finished", {"output_seconds": 27.0})) |
|
|
| rows = wait_records(store, 5) |
| kinds = [row.get("type") for row in rows] |
| check("writer persisted all record kinds", kinds == [ |
| "session_start", "event", "event", "checkpoint", "session_finish" |
| ], str(kinds)) |
| raw = path.read_text(encoding="utf-8") |
| check("signed URL is not persisted", "signed.example" not in raw and "should-not-persist" not in raw) |
| finish = [row for row in rows if row.get("type") == "session_finish"][-1] |
| summary = finish["summary"] |
| check("summary identifies broadcaster", summary["broadcaster"] == "alice") |
| check("summary counts source-open status", summary["counters"].get("source_open_http_403") == 1, |
| str(summary["counters"])) |
| check("summary counts slow ffmpeg samples", summary["counters"].get("ffmpeg_speed_below_0_95_samples") == 1, |
| str(summary["counters"])) |
| check("summary averages ffmpeg speed", abs(summary["stats"]["ffmpeg_speed"]["avg"] - 0.905) < 0.0001, |
| str(summary["stats"].get("ffmpeg_speed"))) |
| check("summary keeps minute buckets", len(summary["minutes"]) == 1, str(summary["minutes"])) |
|
|
| before = path.read_bytes() |
| reopened = Telemetry(path) |
| check("reopening telemetry does not truncate", path.read_bytes() == before) |
| check("reopened store reads old records", len(reopened.read_records(10)) == len(rows)) |
| check("no delete/rotate/truncate API", not any( |
| hasattr(Telemetry, name) for name in ("delete", "rotate", "truncate", "clear", "purge") |
| )) |
|
|
| os.environ["TELEMETRY_ENABLED"] = "0" |
| try: |
| disabled_path = pathlib.Path(tmp) / "disabled" / "telemetry.jsonl" |
| disabled = Telemetry(disabled_path) |
| check("disabled telemetry is a no-op", disabled.start_session("bob") is None and not disabled_path.exists()) |
| finally: |
| os.environ.pop("TELEMETRY_ENABLED", None) |
|
|
| blocker = pathlib.Path(tmp) / "blocker" |
| blocker.write_text("not a directory", encoding="utf-8") |
| broken = Telemetry(blocker / "telemetry" / "telemetry.jsonl") |
| broken_key = broken.start_session("carol") |
| broken.observe(broken_key, {"ffmpeg_speed": 1.0}) |
| broken.finish(broken_key, "storage unavailable") |
| deadline = time.time() + 3.0 |
| while time.time() < deadline and broken.write_errors < 2: |
| time.sleep(0.02) |
| check("storage failure stays inside telemetry", broken.write_errors >= 2, |
| "write_errors=%s" % broken.write_errors) |
|
|
|
|
| relay_source = (ROOT / "relay.py").read_text(encoding="utf-8") |
| check("telemetry result is never a streaming condition", "if telemetry_" not in relay_source and |
| "while telemetry_" not in relay_source) |
| check("relay records HTTP status at initial open", "'stage':'initial_open'" in relay_source and |
| "http_status_from_exception" in relay_source) |
|
|
| print("\n%d passed, %d failed" % (passed, failed)) |
| sys.exit(1 if failed else 0) |
|
|
