Spaces:
Running
Running
File size: 5,436 Bytes
ce0ef24 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 | #!/usr/bin/env python3
"""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)
|