"""Tests for the fast ``MessageDisplay`` deltas-appender hook.""" from __future__ import annotations import io import json import os import sys from pathlib import Path import pytest from omnigent import claude_native_message_display_hook as hook from omnigent.claude_native_bridge import read_message_deltas_from_offset def _run_hook( payload: dict[str, object], bridge_dir: Path, monkeypatch: pytest.MonkeyPatch ) -> int: """ Drive the hook ``main`` with one payload on stdin. :param payload: Hook JSON object to feed on stdin, e.g. ``{"hook_event_name": "MessageDisplay", "message_id": "m1", "index": 0, "final": False, "delta": "hi"}``. :param bridge_dir: Bridge directory the hook should append to. :param monkeypatch: Pytest monkeypatch fixture used to set stdin. :returns: The hook's process exit code. """ monkeypatch.setattr(sys, "stdin", io.StringIO(json.dumps(payload))) return hook.main(["--bridge-dir", str(bridge_dir)]) def test_message_display_hook_appends_well_formed_record( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """ A valid ``MessageDisplay`` payload is appended verbatim, in order. Fails if the hook drops fields, reorders, or writes a shape the bridge reader can't parse — which would break live streaming end to end (the forwarder would forward nothing). """ bridge_dir = tmp_path / "bridge" bridge_dir.mkdir() assert ( _run_hook( { "hook_event_name": "MessageDisplay", "message_id": "m1", "index": 0, "final": False, "delta": "Hello ", }, bridge_dir, monkeypatch, ) == 0 ) assert ( _run_hook( { "hook_event_name": "MessageDisplay", "message_id": "m1", "index": 1, "final": True, "delta": "world", }, bridge_dir, monkeypatch, ) == 0 ) result = read_message_deltas_from_offset(bridge_dir, 0) # Both chunks land, in order, with every field preserved — proving # the on-disk shape round-trips through the reader the forwarder uses. assert [(d.message_id, d.index, d.final, d.delta) for d in result.deltas] == [ ("m1", 0, False, "Hello "), ("m1", 1, True, "world"), ] def test_message_display_hook_writes_owner_only_file( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """ The deltas file is created with owner-only (0600) permissions. Fails if the streamed assistant text (same content as the message) becomes world-readable on a shared host. """ bridge_dir = tmp_path / "bridge" bridge_dir.mkdir() _run_hook( {"hook_event_name": "MessageDisplay", "message_id": "m1", "index": 0, "delta": "x"}, bridge_dir, monkeypatch, ) mode = (bridge_dir / hook.MESSAGE_DELTAS_FILE).stat().st_mode assert oct(mode & 0o777) == "0o600" @pytest.mark.parametrize( "payload", [ {"hook_event_name": "MessageDisplay", "delta": "no id"}, {"hook_event_name": "MessageDisplay", "message_id": "", "delta": "empty id"}, {"hook_event_name": "MessageDisplay", "message_id": "m1"}, {"hook_event_name": "MessageDisplay", "message_id": "m1", "delta": 123}, {"hook_event_name": "MessageDisplay", "message_id": "m1", "delta": None}, ], ids=["missing-id", "empty-id", "missing-delta", "non-string-delta", "null-delta"], ) def test_message_display_hook_skips_unforwardable_payloads( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, payload: dict[str, object] ) -> None: """ Payloads lacking a usable ``message_id``/``delta`` write nothing. Fails if the hook appends a record the forwarder couldn't turn into a valid delta event (e.g. an empty message id or a non-string delta), which would surface as a malformed SSE event downstream. """ bridge_dir = tmp_path / "bridge" bridge_dir.mkdir() assert _run_hook(payload, bridge_dir, monkeypatch) == 0 # No file at all — there was nothing forwardable to record. assert not (bridge_dir / hook.MESSAGE_DELTAS_FILE).exists() def test_message_display_hook_defaults_missing_index_to_zero( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """ A single-chunk message with no ``index`` still forwards (index 0). Fails if a missing index is dropped or coerced to something the reader rejects, which would lose the one-and-only chunk of a short assistant message. """ bridge_dir = tmp_path / "bridge" bridge_dir.mkdir() _run_hook( {"hook_event_name": "MessageDisplay", "message_id": "m1", "final": True, "delta": "hi"}, bridge_dir, monkeypatch, ) result = read_message_deltas_from_offset(bridge_dir, 0) assert [(d.message_id, d.index, d.final, d.delta) for d in result.deltas] == [ ("m1", 0, True, "hi") ] def test_message_display_hook_swallows_malformed_json( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] ) -> None: """ Malformed stdin exits 0 (never blocks Claude) and writes nothing. Claude blocks on command hooks, so a parse error must be a silent no-op for the TUI; fails if the hook raises or appends garbage. """ bridge_dir = tmp_path / "bridge" bridge_dir.mkdir() monkeypatch.setattr(sys, "stdin", io.StringIO("{not json")) assert hook.main(["--bridge-dir", str(bridge_dir)]) == 0 assert not (bridge_dir / hook.MESSAGE_DELTAS_FILE).exists() # Diagnostic goes to stderr so it never lands in Claude's stdout # (which Claude would interpret as hook output). assert "malformed JSON" in capsys.readouterr().err def test_message_display_hook_many_appends_stay_line_clean( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """ Many successive appends to the shared file stay newline-framed. Each ``MessageDisplay`` chunk is a separate hook invocation appending one line; this asserts every line is independently parseable and the reader's reported offset reaches EOF after consuming them all. Fails if the hook ever wrote a record without a trailing newline (which would make the next record's bytes glom onto it and break the reader's per-line decode). NOTE: this exercises sequential appends, not true concurrent subprocesses — the O_APPEND atomicity that makes real per-chunk parallelism safe is a POSIX guarantee, not asserted here. """ bridge_dir = tmp_path / "bridge" bridge_dir.mkdir() for i in range(50): _run_hook( { "hook_event_name": "MessageDisplay", "message_id": "m1", "index": i, "delta": f"c{i} ", }, bridge_dir, monkeypatch, ) result = read_message_deltas_from_offset(bridge_dir, 0) # All 50 chunks parse and arrive in index order — no line was torn. assert [d.index for d in result.deltas] == list(range(50)) assert result.byte_offset == os.path.getsize(bridge_dir / hook.MESSAGE_DELTAS_FILE)