"""Tests for the dedicated events journal (write path and read path).""" import asyncio import json from hbd.server import journal def _make_journal(tmp_path, **overrides): cfg = {"journal_dir": str(tmp_path), "journal_file": "events.journal"} cfg.update(overrides) j = journal.MessageJournal(cfg) assert asyncio.run(j.initialize()) return j def _read_lines(tmp_path, name="events.journal"): return (tmp_path / name).read_text(encoding="utf-8").splitlines() EV1 = {"ts": 1000.0, "host": "h1", "level": "INFO", "service": None, "message": "host up"} EV2 = {"ts": 2000.0, "host": "h2", "level": "CRITICAL", "service": "cpu", "message": "cpu high"} def test_log_event_writes_one_json_line(tmp_path): j = _make_journal(tmp_path) asyncio.run(j.log_event(EV1)) asyncio.run(j.close()) lines = _read_lines(tmp_path) assert len(lines) == 1 assert json.loads(lines[0]) == EV1 def test_log_event_appends_in_order(tmp_path): j = _make_journal(tmp_path) asyncio.run(j.log_event(EV1)) asyncio.run(j.log_event(EV2)) asyncio.run(j.close()) lines = _read_lines(tmp_path) assert [json.loads(ln)["ts"] for ln in lines] == [1000.0, 2000.0] def test_log_event_rotates_at_max_size(tmp_path): # max_size fits one serialized event line (~77 bytes) but not two, so the # second write triggers exactly one rotation j = _make_journal(tmp_path, journal_max_size=120) asyncio.run(j.log_event(EV1)) asyncio.run(j.log_event(EV2)) asyncio.run(j.close()) backups = list(tmp_path.glob("events.journal.*")) assert len(backups) == 1 assert json.loads(backups[0].read_text().splitlines()[0]) == EV1 assert json.loads(_read_lines(tmp_path)[0]) == EV2 def test_log_event_noop_when_disabled(tmp_path): j = journal.MessageJournal( {"journal_dir": str(tmp_path), "journal_file": "events.journal", "journal_enabled": False} ) asyncio.run(j.initialize()) asyncio.run(j.log_event(EV1)) assert not (tmp_path / "events.journal").exists() def test_backfill_seeds_empty_journal(tmp_path): j = _make_journal(tmp_path) asyncio.run(j.backfill([EV1, EV2])) asyncio.run(j.close()) lines = _read_lines(tmp_path) assert len(lines) == 2 assert json.loads(lines[0]) == EV1 def test_backfill_skipped_when_journal_nonempty(tmp_path): (tmp_path / "events.journal").write_text(json.dumps(EV1) + "\n") j = _make_journal(tmp_path) # initialize() picks up the existing size asyncio.run(j.backfill([EV2])) asyncio.run(j.close()) assert len(_read_lines(tmp_path)) == 1 def test_get_events_journal_uses_events_config_keys(tmp_path): journal._events_journal_instance = None try: ej = journal.get_events_journal( { "journal_dir": str(tmp_path), "events_journal_file": "ev.jsonl", "events_journal_max_size": 12345, "events_journal_max_backups": 3, } ) assert ej.journal_file == "ev.jsonl" assert ej.max_size == 12345 assert ej.max_backups == 3 assert ej.journal_dir == tmp_path # singleton: second call returns the same instance assert journal.get_events_journal() is ej finally: journal._events_journal_instance = None def test_get_events_journal_defaults(tmp_path): journal._events_journal_instance = None try: ej = journal.get_events_journal({"journal_dir": str(tmp_path)}) assert ej.journal_file == "events.journal" assert ej.max_size == 10 * 1024 * 1024 assert ej.max_backups == 10 finally: journal._events_journal_instance = None # ---- read path ------------------------------------------------------------- def _write_journal(path, events): path.write_text("".join(json.dumps(e) + "\n" for e in events), encoding="utf-8") def _evts(*ts_list): return [ {"ts": float(t), "host": f"host{i}", "level": "INFO", "service": None, "message": f"msg {t}"} for i, t in enumerate(ts_list) ] def test_read_events_newest_first(tmp_path): _write_journal(tmp_path / "events.journal", _evts(1, 2, 3)) events, more = journal.read_events(tmp_path) assert [e["ts"] for e in events] == [3.0, 2.0, 1.0] assert more is False def test_read_events_limit_and_more(tmp_path): _write_journal(tmp_path / "events.journal", _evts(1, 2, 3)) events, more = journal.read_events(tmp_path, limit=2) assert [e["ts"] for e in events] == [3.0, 2.0] assert more is True def test_read_events_before_cursor(tmp_path): _write_journal(tmp_path / "events.journal", _evts(1, 2, 3)) events, _ = journal.read_events(tmp_path, before=3.0) assert [e["ts"] for e in events] == [2.0, 1.0] def test_read_events_spans_rotated_files(tmp_path): # rotated backup holds the oldest events; current file the newest _write_journal(tmp_path / "events.journal.20260101-000000", _evts(1, 2)) _write_journal(tmp_path / "events.journal.20260201-000000", _evts(3, 4)) _write_journal(tmp_path / "events.journal", _evts(5, 6)) events, more = journal.read_events(tmp_path, limit=10) assert [e["ts"] for e in events] == [6.0, 5.0, 4.0, 3.0, 2.0, 1.0] assert more is False def test_read_events_pagination_across_files(tmp_path): _write_journal(tmp_path / "events.journal.20260101-000000", _evts(1, 2)) _write_journal(tmp_path / "events.journal", _evts(3, 4)) page1, more1 = journal.read_events(tmp_path, limit=3) assert [e["ts"] for e in page1] == [4.0, 3.0, 2.0] assert more1 is True page2, more2 = journal.read_events(tmp_path, limit=3, before=page1[-1]["ts"]) assert [e["ts"] for e in page2] == [1.0] assert more2 is False def test_read_events_host_filter_substring_case_insensitive(tmp_path): evs = [ {"ts": 1.0, "host": "Wentworth", "level": "INFO", "service": None, "message": "a"}, {"ts": 2.0, "host": "winter", "level": "INFO", "service": None, "message": "b"}, ] _write_journal(tmp_path / "events.journal", evs) events, _ = journal.read_events(tmp_path, host="went") assert [e["host"] for e in events] == ["Wentworth"] def test_read_events_level_filter_exact_case_insensitive(tmp_path): evs = [ {"ts": 1.0, "host": "h", "level": "CRITICAL", "service": None, "message": "a"}, {"ts": 2.0, "host": "h", "level": "INFO", "service": None, "message": "b"}, ] _write_journal(tmp_path / "events.journal", evs) events, _ = journal.read_events(tmp_path, level="critical") assert [e["level"] for e in events] == ["CRITICAL"] def test_read_events_message_filter(tmp_path): evs = [ {"ts": 1.0, "host": "h", "level": "INFO", "service": None, "message": "disk almost full"}, {"ts": 2.0, "host": "h", "level": "INFO", "service": None, "message": "all quiet"}, ] _write_journal(tmp_path / "events.journal", evs) events, _ = journal.read_events(tmp_path, q="Disk") assert [e["ts"] for e in events] == [1.0] def test_read_events_skips_malformed_lines(tmp_path): p = tmp_path / "events.journal" p.write_text('{"ts": 1.0, "host": "h", "level": "INFO", "message": "ok"}\nnot json\n[1,2]\n') events, _ = journal.read_events(tmp_path) assert [e["ts"] for e in events] == [1.0] def test_read_events_predicate(tmp_path): _write_journal(tmp_path / "events.journal", _evts(1, 2)) events, _ = journal.read_events(tmp_path, predicate=lambda e: e["host"] == "host0") assert [e["host"] for e in events] == ["host0"] def test_read_events_missing_dir(tmp_path): events, more = journal.read_events(tmp_path / "nope") assert events == [] and more is False def test_filter_events_over_in_memory_ring(): ring = _evts(1, 2, 3) # oldest-first, like data.msgs events, more = journal.filter_events(reversed(ring), limit=2) assert [e["ts"] for e in events] == [3.0, 2.0] assert more is True # ---- eventlog wiring -------------------------------------------------------- def test_eventlog_writes_to_events_journal(tmp_path): from hbd.server import data, notify journal._events_journal_instance = None saved_msgs = data.msgs data.msgs = [] try: ej = journal.get_events_journal({"journal_dir": str(tmp_path)}) async def scenario(): await ej.initialize() notify.setup({}, loop=asyncio.get_running_loop()) notify.eventlog("h1", "INFO", "hello world") await asyncio.sleep(0.05) # let the scheduled journal write run await ej.close() asyncio.run(scenario()) lines = _read_lines(tmp_path) assert len(lines) == 1 ev = json.loads(lines[0]) assert ev["host"] == "h1" assert ev["level"] == "INFO" assert ev["message"] == "hello world" assert isinstance(ev["ts"], float) finally: journal._events_journal_instance = None data.msgs = saved_msgs notify._loop = None