feat: use telemetry in guardrails
This commit is contained in:
@@ -7,6 +7,8 @@ import pytest
|
||||
from rdb_discovery.guardrails import (
|
||||
check_repeated_reads,
|
||||
check_repeated_commands,
|
||||
check_repeated_commands_telemetry,
|
||||
check_repeated_reads_telemetry,
|
||||
check_no_recent_file_changes,
|
||||
check_no_test_run_recorded,
|
||||
check_run_log_updated,
|
||||
@@ -79,6 +81,76 @@ def test_check_repeated_commands_warning(tmp_path: Path) -> None:
|
||||
assert result["status"] == "warning"
|
||||
|
||||
|
||||
# -- check_repeated_commands_telemetry --
|
||||
|
||||
def test_check_repeated_commands_telemetry_no_events(tmp_path: Path) -> None:
|
||||
"""No telemetry events → status ok with fallback message."""
|
||||
result = check_repeated_commands_telemetry(tmp_path)
|
||||
assert result["status"] == "ok"
|
||||
assert "No telemetry data available" in result["details"]
|
||||
|
||||
|
||||
def test_check_repeated_commands_telemetry_warning(tmp_path: Path) -> None:
|
||||
"""Same command repeated >3 times in telemetry → warning."""
|
||||
from rdb_discovery.telemetry import record_event
|
||||
|
||||
for _ in range(5):
|
||||
record_event(tmp_path, "command", "rdb prompt")
|
||||
result = check_repeated_commands_telemetry(tmp_path)
|
||||
assert result["status"] == "warning"
|
||||
assert "rdb prompt" in result["details"]
|
||||
|
||||
|
||||
def test_check_repeated_commands_telemetry_no_over_threshold(tmp_path: Path) -> None:
|
||||
"""Same command repeated <=3 times → ok."""
|
||||
from rdb_discovery.telemetry import record_event
|
||||
|
||||
for _ in range(3):
|
||||
record_event(tmp_path, "command", "rdb prompt")
|
||||
result = check_repeated_commands_telemetry(tmp_path)
|
||||
assert result["status"] == "ok"
|
||||
|
||||
|
||||
# -- check_repeated_reads_telemetry --
|
||||
|
||||
def test_check_repeated_reads_telemetry_no_events(tmp_path: Path) -> None:
|
||||
"""No telemetry events → status ok with fallback message."""
|
||||
result = check_repeated_reads_telemetry(tmp_path)
|
||||
assert result["status"] == "ok"
|
||||
assert "No telemetry data available" in result["details"]
|
||||
|
||||
|
||||
def test_check_repeated_reads_telemetry_no_read_events(tmp_path: Path) -> None:
|
||||
"""Only command events, no read events → ok with message."""
|
||||
from rdb_discovery.telemetry import record_event
|
||||
|
||||
record_event(tmp_path, "command", "rdb prompt")
|
||||
result = check_repeated_reads_telemetry(tmp_path)
|
||||
assert result["status"] == "ok"
|
||||
assert "No file read events found" in result["details"]
|
||||
|
||||
|
||||
def test_check_repeated_reads_telemetry_warning(tmp_path: Path) -> None:
|
||||
"""Same file read >3 times in telemetry → warning."""
|
||||
from rdb_discovery.telemetry import record_event
|
||||
|
||||
for _ in range(5):
|
||||
record_event(tmp_path, "read", "README.md")
|
||||
result = check_repeated_reads_telemetry(tmp_path)
|
||||
assert result["status"] == "warning"
|
||||
assert "README.md" in result["details"]
|
||||
|
||||
|
||||
def test_check_repeated_reads_telemetry_no_over_threshold(tmp_path: Path) -> None:
|
||||
"""Same file read <=3 times → ok."""
|
||||
from rdb_discovery.telemetry import record_event
|
||||
|
||||
for _ in range(3):
|
||||
record_event(tmp_path, "read", "TASKS.md")
|
||||
result = check_repeated_reads_telemetry(tmp_path)
|
||||
assert result["status"] == "ok"
|
||||
|
||||
|
||||
# -- check_no_recent_file_changes --
|
||||
|
||||
def test_check_no_recent_file_changes_ok(tmp_path: Path) -> None:
|
||||
|
||||
Reference in New Issue
Block a user