65 lines
1.7 KiB
Python
65 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from sshctl.operations.events import Event, flush, log_event, recent
|
|
|
|
|
|
def test_event_to_dict() -> None:
|
|
ev = Event(
|
|
command="host list",
|
|
status="ok",
|
|
args={"--verbose": "true"},
|
|
result="",
|
|
duration_ms=12.5,
|
|
)
|
|
d = ev.to_dict()
|
|
assert d["cmd"] == "host list"
|
|
assert d["status"] == "ok"
|
|
assert d["args"] == {"--verbose": "true"}
|
|
assert d["duration_ms"] == 12.5
|
|
assert isinstance(d["ts"], float)
|
|
|
|
|
|
def test_log_event_and_flush(ssh_dir: Path) -> None:
|
|
log_event(Event(command="check", status="ok", duration_ms=1.5))
|
|
flush(ssh_dir)
|
|
rows = recent(ssh_dir)
|
|
assert len(rows) == 1
|
|
assert rows[0]["cmd"] == "check"
|
|
assert rows[0]["status"] == "ok"
|
|
|
|
|
|
def test_flush_empty_no_file(ssh_dir: Path) -> None:
|
|
flush(ssh_dir)
|
|
assert recent(ssh_dir) == []
|
|
|
|
|
|
def test_recent_empty_when_no_file(ssh_dir: Path) -> None:
|
|
assert recent(ssh_dir) == []
|
|
|
|
|
|
def test_recent_limits(ssh_dir: Path) -> None:
|
|
for i in range(10):
|
|
log_event(Event(command=f"cmd{i}", status="ok"))
|
|
flush(ssh_dir)
|
|
rows = recent(ssh_dir, limit=3)
|
|
assert len(rows) == 3
|
|
assert rows[0]["cmd"] == "cmd7"
|
|
|
|
|
|
def test_recent_skips_bad_json(ssh_dir: Path) -> None:
|
|
report = ssh_dir / "report"
|
|
report.mkdir(exist_ok=True)
|
|
(report / "events.ndjson").write_text("not-json\n")
|
|
assert recent(ssh_dir) == []
|
|
|
|
|
|
def test_flush_writes_valid_ndjson(ssh_dir: Path) -> None:
|
|
log_event(Event(command="tree", status="ok"))
|
|
log_event(Event(command="diff", status="error"))
|
|
flush(ssh_dir)
|
|
path = ssh_dir / "report" / "events.ndjson"
|
|
lines = path.read_text().strip().splitlines()
|
|
assert len(lines) == 2
|