truetype
b20f3c0f31
- .gitea/workflows/ci.yml: ruff + matrix pytest on 3.10/3.11/3.12 on push and PR to main; concurrency cancels superseded runs - .gitea/workflows/release.yml: gate (ruff + pytest) + build sdist/wheel on tag 'v*', uploads via actions/upload-artifact@v4 - README.md: add ci and version badges right under H1 - src/viandante/cli.py: drop unused prompt_toolkit.patch_stdout import - tests/test_deadsouls.py: add missing RoomBlock import (F821) - tests/test_logger.py: drop unused local 'raw' in unicode test - tests/test_mapper.py: rename unused locals to _a/_b/_c (F841) to preserve intent (B has 's' back-edge; C is orphan)
276 lines
8.5 KiB
Python
276 lines
8.5 KiB
Python
"""Tests for :mod:`viandante.logger`."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from viandante.logger import EventLogger
|
|
|
|
|
|
# ---- shared helpers ------------------------------------------------------
|
|
|
|
|
|
def _read_jsonl(path: Path) -> list[dict]:
|
|
return [
|
|
json.loads(line)
|
|
for line in path.read_text(encoding="utf-8").splitlines()
|
|
if line.strip()
|
|
]
|
|
|
|
|
|
# ---- json sink -----------------------------------------------------------
|
|
|
|
|
|
def test_json_sink_roundtrip(tmp_path: Path) -> None:
|
|
out = tmp_path / "session.jsonl"
|
|
log = EventLogger(json_path=str(out))
|
|
log.session_start("localhost", 6666, "foo")
|
|
log.room("vnum42", "Town Square", ["n", "e"])
|
|
log.vitals(hp=80, max_hp=100)
|
|
log.line_sent("north")
|
|
log.close()
|
|
|
|
events = _read_jsonl(out)
|
|
assert [e["kind"] for e in events] == [
|
|
"session_start",
|
|
"room_enter",
|
|
"vitals",
|
|
"line_sent",
|
|
]
|
|
assert events[0]["data"]["host"] == "localhost"
|
|
assert events[1]["data"]["exits"] == ["n", "e"]
|
|
assert events[2]["data"]["hp"] == 80
|
|
assert events[3]["data"]["text"] == "north"
|
|
# Every event has a UTC ISO8601 timestamp.
|
|
for e in events:
|
|
assert e["ts"].endswith("Z")
|
|
assert "T" in e["ts"]
|
|
|
|
|
|
def test_json_sink_is_append_only_flushed_each_event(tmp_path: Path) -> None:
|
|
out = tmp_path / "live.jsonl"
|
|
log = EventLogger(json_path=str(out))
|
|
log.event("ping", {"n": 1})
|
|
# File must be readable mid-session.
|
|
inline = out.read_text(encoding="utf-8")
|
|
assert "ping" in inline
|
|
log.event("pong", {"n": 2})
|
|
log.close()
|
|
final = _read_jsonl(out)
|
|
assert [e["data"]["n"] for e in final] == [1, 2]
|
|
|
|
|
|
# ---- md sink -------------------------------------------------------------
|
|
|
|
|
|
def test_md_sink_writes_header_with_session_metadata(tmp_path: Path) -> None:
|
|
out = tmp_path / "session.md"
|
|
log = EventLogger(md_path=str(out), host="h.example", port=7000, user="alice")
|
|
log.close()
|
|
body = out.read_text(encoding="utf-8")
|
|
assert body.startswith("# MUD log — ")
|
|
assert "**host**: `h.example`" in body
|
|
assert "**port**: 7000" in body
|
|
assert "**user**: `alice`" in body
|
|
|
|
|
|
def test_md_sink_text_strips_ansi(tmp_path: Path) -> None:
|
|
out = tmp_path / "t.md"
|
|
log = EventLogger(md_path=str(out))
|
|
log.text("\x1b[32mhello\x1b[0m world\nline two\n")
|
|
log.close()
|
|
body = out.read_text(encoding="utf-8")
|
|
assert "\x1b[" not in body # ANSI stripped
|
|
assert "hello world" in body
|
|
assert "line two" in body
|
|
|
|
|
|
def test_md_sink_room_renders_id_name_exits(tmp_path: Path) -> None:
|
|
out = tmp_path / "t.md"
|
|
log = EventLogger(md_path=str(out))
|
|
log.room("vnum7", "Dungeon Lobby", ["d"])
|
|
log.close()
|
|
body = out.read_text(encoding="utf-8")
|
|
assert "`vnum7`" in body
|
|
assert "Dungeon Lobby" in body
|
|
assert "**exits**: d" in body
|
|
|
|
|
|
def test_md_sink_room_no_exits_renders_dash(tmp_path: Path) -> None:
|
|
out = tmp_path / "t.md"
|
|
log = EventLogger(md_path=str(out))
|
|
log.room("v1", "Cell", [])
|
|
log.close()
|
|
assert "**exits**: —" in out.read_text(encoding="utf-8")
|
|
|
|
|
|
def test_md_sink_gmcp_serializes_payload_as_json(tmp_path: Path) -> None:
|
|
out = tmp_path / "t.md"
|
|
log = EventLogger(md_path=str(out))
|
|
log.gmcp("Char.Vitals", {"hp": 90, "max_hp": 100})
|
|
log.close()
|
|
body = out.read_text(encoding="utf-8")
|
|
assert "`Char.Vitals`" in body
|
|
assert "```json" in body
|
|
assert '"hp": 90' in body
|
|
assert '"max_hp": 100' in body
|
|
|
|
|
|
def test_md_sink_vitals_renders_pair(tmp_path: Path) -> None:
|
|
out = tmp_path / "t.md"
|
|
log = EventLogger(md_path=str(out))
|
|
log.vitals(hp=1, max_hp=10, sp=5, max_sp=20, mv=3, max_mv=30)
|
|
log.close()
|
|
body = out.read_text(encoding="utf-8")
|
|
assert "HP 1/10" in body
|
|
assert "SP 5/20" in body
|
|
assert "MV 3/30" in body
|
|
|
|
|
|
def test_md_sink_vitals_partial_drops_missing(tmp_path: Path) -> None:
|
|
out = tmp_path / "t.md"
|
|
log = EventLogger(md_path=str(out))
|
|
log.vitals(hp=42) # sp/mv missing -> not in markdown
|
|
log.close()
|
|
body = out.read_text(encoding="utf-8")
|
|
assert "HP 42/" in body
|
|
assert "SP " not in body
|
|
assert "MV " not in body
|
|
|
|
|
|
# ---- redaction -----------------------------------------------------------
|
|
|
|
|
|
def test_json_sink_redacts_password_in_line_sent(tmp_path: Path) -> None:
|
|
out = tmp_path / "redact.jsonl"
|
|
log = EventLogger(json_path=str(out))
|
|
log.line_sent("hunter2", redacted=True)
|
|
log.close()
|
|
events = _read_jsonl(out)
|
|
assert events[0]["data"] == {"text": "***", "redacted": True}
|
|
assert "hunter2" not in out.read_text(encoding="utf-8")
|
|
|
|
|
|
def test_json_sink_redacts_first_arg_in_local_cmd(tmp_path: Path) -> None:
|
|
out = tmp_path / "redact.jsonl"
|
|
log = EventLogger(json_path=str(out))
|
|
log.local_cmd("password", ["hunter2", "extra"], redacted=True)
|
|
log.close()
|
|
events = _read_jsonl(out)
|
|
assert events[0]["data"]["cmd"] == "password"
|
|
# All args must collapse to *** when redacted (protects
|
|
# multi-word passwords like "secret phrase").
|
|
assert events[0]["data"]["args"] == ["***", "***"]
|
|
assert "hunter2" not in out.read_text(encoding="utf-8")
|
|
assert "extra" not in out.read_text(encoding="utf-8")
|
|
|
|
|
|
def test_md_sink_redacted_line_sent_marker(tmp_path: Path) -> None:
|
|
out = tmp_path / "t.md"
|
|
log = EventLogger(md_path=str(out))
|
|
log.line_sent("hunter2", redacted=True)
|
|
log.close()
|
|
body = out.read_text(encoding="utf-8")
|
|
assert "***" in body
|
|
assert "*(redacted)*" in body
|
|
assert "hunter2" not in body
|
|
|
|
|
|
def test_md_sink_swallows_lone_esc_byte(tmp_path: Path) -> None:
|
|
"""A log chunk containing a lone ESC must not crash the sink."""
|
|
out = tmp_path / "t.md"
|
|
log = EventLogger(md_path=str(out))
|
|
log.text("hello\x1b world\n") # \x1b without '[' falls outside CSI
|
|
log.close()
|
|
body = out.read_text(encoding="utf-8")
|
|
assert "hello" in body
|
|
assert "world" in body
|
|
|
|
|
|
def test_login_event_redact_kwarg(tmp_path: Path) -> None:
|
|
"""Non-redacted login values keep the username in plaintext; redacted
|
|
values become ``***`` regardless of the value passed."""
|
|
out = tmp_path / "login.jsonl"
|
|
log = EventLogger(json_path=str(out))
|
|
log.login("user_sent", value="alice", redact=False)
|
|
log.login("password_sent", value="hunter2")
|
|
log.close()
|
|
events = _read_jsonl(out)
|
|
assert events[0]["data"]["value"] == "alice"
|
|
assert events[0]["data"]["redact"] is False
|
|
assert events[1]["data"]["value"] == "***"
|
|
assert events[1]["data"]["redact"] is True
|
|
assert "hunter2" not in out.read_text(encoding="utf-8")
|
|
|
|
|
|
# ---- lifecycle -----------------------------------------------------------
|
|
|
|
|
|
def test_no_paths_is_a_noop() -> None:
|
|
log = EventLogger()
|
|
log.session_start("h", 1, "u")
|
|
log.text("\x1b[31mhi\x1b[0m")
|
|
log.room("v1", "Plaza", ["e"])
|
|
log.vitals(hp=1)
|
|
log.line_sent("north")
|
|
log.local_cmd("map")
|
|
log.login("user_sent", "alice")
|
|
log.session_end()
|
|
log.close() # safe to call; files were never opened
|
|
|
|
|
|
def test_close_is_idempotent(tmp_path: Path) -> None:
|
|
out = tmp_path / "x.jsonl"
|
|
log = EventLogger(json_path=str(out))
|
|
log.event("a", {})
|
|
log.close()
|
|
log.close()
|
|
log.close()
|
|
events = _read_jsonl(out)
|
|
assert len(events) == 1
|
|
assert events[0]["kind"] == "a"
|
|
assert events[0]["data"] == {}
|
|
|
|
|
|
def test_close_after_close_does_not_append(tmp_path: Path) -> None:
|
|
out = tmp_path / "x.jsonl"
|
|
log = EventLogger(json_path=str(out))
|
|
log.event("first", {})
|
|
log.close()
|
|
log.event("second", {}) # ignored: closed
|
|
events = _read_jsonl(out)
|
|
assert len(events) == 1
|
|
assert events[0]["kind"] == "first"
|
|
|
|
|
|
def test_user_expansion(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("HOME", str(tmp_path))
|
|
log = EventLogger(json_path="~/session.jsonl")
|
|
log.event("hello", {})
|
|
log.close()
|
|
expected = tmp_path / "session.jsonl"
|
|
assert expected.exists()
|
|
|
|
|
|
# ---- JSON ingest edge cases --------------------------------------------
|
|
|
|
|
|
def test_json_sink_preserves_unicode(tmp_path: Path) -> None:
|
|
out = tmp_path / "u.jsonl"
|
|
log = EventLogger(json_path=str(out))
|
|
log.text("ilà ★")
|
|
log.close()
|
|
events = _read_jsonl(out)
|
|
assert events[0]["data"]["raw"] == "ilà ★"
|
|
|
|
|
|
def test_json_sink_payload_may_be_non_dict(tmp_path: Path) -> None:
|
|
out = tmp_path / "p.jsonl"
|
|
log = EventLogger(json_path=str(out))
|
|
log.gmcp("Room.Info", [1, 2, 3])
|
|
log.close()
|
|
events = _read_jsonl(out)
|
|
assert events[0]["data"]["payload"] == [1, 2, 3]
|