134 lines
4.0 KiB
Python
134 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from rdb_discovery.telemetry import TelemetryEvent, record_event, read_events
|
|
|
|
|
|
# -- TelemetryEvent --
|
|
|
|
def test_telemetry_event_to_dict() -> None:
|
|
event = TelemetryEvent(
|
|
timestamp="2026-06-02T12:00:00+00:00",
|
|
event_type="command",
|
|
target="rdb prompt",
|
|
details={"flag": True},
|
|
)
|
|
d = event.to_dict()
|
|
assert d["event_type"] == "command"
|
|
assert d["target"] == "rdb prompt"
|
|
assert d["details"]["flag"] is True
|
|
|
|
|
|
def test_telemetry_event_defaults_to_empty_details() -> None:
|
|
event = TelemetryEvent(
|
|
timestamp="2026-06-02T12:00:00+00:00",
|
|
event_type="command",
|
|
target="rdb prompt",
|
|
)
|
|
assert event.details == {}
|
|
|
|
|
|
def test_telemetry_event_required_fields() -> None:
|
|
event = TelemetryEvent(
|
|
timestamp="2026-06-02T12:00:00+00:00",
|
|
event_type="command",
|
|
target="rdb prompt",
|
|
)
|
|
assert event.timestamp == "2026-06-02T12:00:00+00:00"
|
|
assert event.event_type == "command"
|
|
assert event.target == "rdb prompt"
|
|
|
|
|
|
# -- record_event (file creation) --
|
|
|
|
def test_record_event_creates_session_log(tmp_path: Path) -> None:
|
|
log = tmp_path / ".rdb" / "session-log.jsonl"
|
|
assert not log.exists()
|
|
record_event(tmp_path, "command", "rdb prompt")
|
|
assert log.exists()
|
|
|
|
|
|
# -- record_event (appending) --
|
|
|
|
def test_record_event_appends_jsonl_line(tmp_path: Path) -> None:
|
|
record_event(tmp_path, "command", "rdb prompt")
|
|
record_event(tmp_path, "task", "TASK-012")
|
|
lines = log_lines(tmp_path)
|
|
assert len(lines) == 2
|
|
|
|
|
|
def test_record_event_writes_valid_json_per_line(tmp_path: Path) -> None:
|
|
record_event(tmp_path, "command", "rdb prompt", {"arg": "value"})
|
|
record_event(tmp_path, "task", "TASK-012")
|
|
for line in log_lines(tmp_path):
|
|
data = json.loads(line)
|
|
assert data["event_type"] in ("command", "task")
|
|
|
|
|
|
def test_record_event_has_required_keys(tmp_path: Path) -> None:
|
|
record_event(tmp_path, "command", "rdb prompt")
|
|
event = TelemetryEvent(**json.loads(log_lines(tmp_path)[0]))
|
|
assert hasattr(event, "timestamp")
|
|
assert hasattr(event, "event_type")
|
|
assert hasattr(event, "target")
|
|
assert hasattr(event, "details")
|
|
|
|
|
|
def test_record_event_details_default_empty_dict(tmp_path: Path) -> None:
|
|
record_event(tmp_path, "command", "rdb prompt")
|
|
event = TelemetryEvent(**json.loads(log_lines(tmp_path)[0]))
|
|
assert event.details == {}
|
|
|
|
|
|
# -- read_events (no file) --
|
|
|
|
def test_read_events_returns_empty_when_no_log(tmp_path: Path) -> None:
|
|
events = read_events(tmp_path)
|
|
assert events == []
|
|
|
|
|
|
# -- read_events (round-trip) --
|
|
|
|
def test_read_events_returns_recorded_events(tmp_path: Path) -> None:
|
|
record_event(tmp_path, "command", "rdb prompt", {"count": 42})
|
|
events = read_events(tmp_path)
|
|
assert len(events) == 1
|
|
assert events[0].event_type == "command"
|
|
assert events[0].target == "rdb prompt"
|
|
assert events[0].details["count"] == 42
|
|
|
|
|
|
def test_read_events_returns_multiple_events(tmp_path: Path) -> None:
|
|
record_event(tmp_path, "task", "TASK-001")
|
|
record_event(tmp_path, "task", "TASK-002")
|
|
record_event(tmp_path, "command", "rdb status")
|
|
events = read_events(tmp_path)
|
|
assert len(events) == 3
|
|
|
|
|
|
# -- read_events (malformed lines) --
|
|
|
|
def test_read_events_skips_malformed_lines(tmp_path: Path) -> None:
|
|
log_path = tmp_path / ".rdb" / "session-log.jsonl"
|
|
log_path.parent.mkdir(parents=True, exist_ok=True)
|
|
log_path.write_text(
|
|
'{"timestamp":"2026-01-01T00:00:00+00:00","event_type":"ok","target":"x","details":{}}\n'
|
|
'this is not json\n'
|
|
'{"timestamp":"2026-01-01T00:00:00+00:00","event_type":"also_ok","target":"y","details":{}}\n',
|
|
encoding="utf-8",
|
|
)
|
|
events = read_events(tmp_path)
|
|
assert len(events) == 2
|
|
assert events[0].event_type == "ok"
|
|
assert events[1].event_type == "also_ok"
|
|
|
|
|
|
# -- Helpers --
|
|
|
|
def log_lines(root: Path) -> list[str]:
|
|
return (root / ".rdb" / "session-log.jsonl").read_text(encoding="utf-8").strip().splitlines()
|