hermes-agent/tests/hermes_cli/test_logs.py

"""Tests for hermes_cli.logs — log viewing and filtering."""

from datetime import datetime, timedelta

from hermes_cli.logs import (
    LOG_FILES,
    _extract_level,
    _extract_logger_name,
    _line_matches_component,
    _matches_filters,
    _parse_line_timestamp,
    _parse_since,
    _read_last_n_lines,
    _read_tail,
)

# ---------------------------------------------------------------------------
# Timestamp parsing
# ---------------------------------------------------------------------------

class TestParseSince:
    def test_hours(self):
        cutoff = _parse_since("2h")
        assert cutoff is not None
        assert abs((datetime.now() - cutoff).total_seconds() - 7200) < 2

    def test_invalid_returns_none(self):
        assert _parse_since("abc") is None
        assert _parse_since("") is None
        assert _parse_since("10x") is None

    def test_whitespace_tolerance(self):
        cutoff = _parse_since("  5m  ")
        assert cutoff is not None

class TestParseLineTimestamp:
    def test_standard_format(self):
        ts = _parse_line_timestamp("2026-04-11 10:23:45 INFO gateway.run: msg")
        assert ts == datetime(2026, 4, 11, 10, 23, 45)

class TestExtractLevel:
    def test_info(self):
        assert _extract_level("2026-01-01 00:00:00 INFO gateway.run: msg") == "INFO"

# ---------------------------------------------------------------------------
# Logger name extraction (new for component filtering)
# ---------------------------------------------------------------------------

class TestExtractLoggerName:
    def test_standard_line(self):
        line = "2026-04-11 10:23:45 INFO gateway.run: Starting gateway"
        assert _extract_logger_name(line) == "gateway.run"

    def test_no_match(self):
        assert _extract_logger_name("random text") is None

class TestLineMatchesComponent:

    def test_gateway_nested(self):
        # Migrated platform adapters log under plugins.platforms.* (#41112) and
        # must still resolve to the gateway component. Use the real expanded
        # gateway prefixes (COMPONENT_PREFIXES["gateway"]) the CLI passes, not a
        # bare ("gateway",), since the logger name no longer literally starts
        # with "gateway".
        from hermes_logging import COMPONENT_PREFIXES
        line = "2026-04-11 10:23:45 INFO plugins.platforms.telegram.adapter: msg"
        assert _line_matches_component(line, COMPONENT_PREFIXES["gateway"])

    def test_unparseable_line(self):
        assert not _line_matches_component("random text", ("gateway",))

# ---------------------------------------------------------------------------
# Combined filter
# ---------------------------------------------------------------------------

class TestMatchesFilters:

    def test_level_filter(self):
        assert _matches_filters(
            "2026-01-01 00:00:00 WARNING x: msg", min_level="WARNING")
        assert not _matches_filters(
            "2026-01-01 00:00:00 INFO x: msg", min_level="WARNING")

    def test_combined_filters(self):
        """All filters must pass for a line to match."""
        line = "2026-04-11 10:00:00 WARNING [sess_1] gateway.run: connection lost"
        assert _matches_filters(
            line,
            min_level="WARNING",
            session_filter="sess_1",
            component_prefixes=("gateway",),
        )
        # Fails component filter
        assert not _matches_filters(
            line,
            min_level="WARNING",
            session_filter="sess_1",
            component_prefixes=("tools",),
        )

    def test_since_filter(self):
        # Line with a very old timestamp should be filtered out
        assert not _matches_filters(
            "2020-01-01 00:00:00 INFO x: old msg",
            since=datetime.now() - timedelta(hours=1))
        # Line with a recent timestamp should pass
        recent = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        assert _matches_filters(
            f"{recent} INFO x: recent msg",
            since=datetime.now() - timedelta(hours=1))

# ---------------------------------------------------------------------------
# File reading
# ---------------------------------------------------------------------------

class TestReadTail:
    def test_read_small_file(self, tmp_path):
        log_file = tmp_path / "test.log"
        lines = [f"2026-01-01 00:00:0{i} INFO x: line {i}\n" for i in range(10)]
        log_file.write_text("".join(lines))

        result = _read_last_n_lines(log_file, 5)
        assert len(result) == 5
        assert "line 9" in result[-1]

    def test_unstamped_lines_share_the_verdict_of_the_record_above(self, tmp_path):
        old = (datetime.now() - timedelta(hours=3)).strftime("%Y-%m-%d %H:%M:%S,000")
        new = datetime.now().strftime("%Y-%m-%d %H:%M:%S,000")
        frames = ["Traceback (most recent call last):\n", '  File "x.py", line 1, in f\n']
        log_file = tmp_path / "errors.log"
        log_file.write_text("".join([
            "orphan tail of a record that started before the window\n",
            f"{old} ERROR gateway.run: old failure\n", *frames, "ValueError: old\n",
            f"{new} INFO tools.x: multi-line info\n", "  info continuation\n",
            f"{new} ERROR gateway.run: new failure\n", *frames, "ValueError: new\n",
        ]))
        since = datetime.now() - timedelta(hours=1)
        from hermes_logging import COMPONENT_PREFIXES

        def read(**filters):
            return "".join(_read_tail(log_file, 50, has_filters=True, **filters))

        assert read(since=since) == "".join([
            f"{new} INFO tools.x: multi-line info\n", "  info continuation\n",
            f"{new} ERROR gateway.run: new failure\n", *frames, "ValueError: new\n",
        ])
        new_failure = "".join([f"{new} ERROR gateway.run: new failure\n", *frames, "ValueError: new\n"])
        assert read(since=since, min_level="WARNING") == new_failure
        assert read(since=since, component_prefixes=COMPONENT_PREFIXES["gateway"]) == new_failure
        # No time/level filter: the orphan lines before the first stamp stay visible.
        assert read(session_filter="orphan") == "orphan tail of a record that started before the window\n"

# ---------------------------------------------------------------------------
# LOG_FILES registry
# ---------------------------------------------------------------------------

def _python_log_line(logger_name: str) -> str:
    import logging

    from agent.redact import RedactingFormatter
    from hermes_logging import _LOG_FORMAT

    record = logging.LogRecord(logger_name, logging.WARNING, __file__, 1, "sample", None, None)
    record.session_tag = ""
    return RedactingFormatter(_LOG_FORMAT).format(record)


def _mcp_output_line() -> str:
    import io

    from tools.mcp_tool_config import _StderrTee

    log = io.StringIO()
    tee = _StderrTee(log)
    tee.sink.write(b"server says hello\n")
    tee.close()
    return log.getvalue()


def _log_file_samples() -> dict:
    """One line per LOG_FILES entry, produced by that file's real writer where Python can run it."""
    return {
        "agent": _python_log_line("run_agent"),
        "errors": _python_log_line("run_agent"),
        "gateway": _python_log_line("gateway.run"),
        "gui": _python_log_line("hermes_cli.web_server"),
        # Written by TypeScript; apps/desktop/electron/desktop-log-line.test.ts pins the same shape.
        "desktop": "2026-09-28 13:18:46,062 [hermes] [boot] ready",
        "mcp": _mcp_output_line(),
    }


def test_every_log_file_writes_a_stamp_hermes_logs_since_can_read():
    samples = _log_file_samples()
    assert set(samples) == set(LOG_FILES), "add a real sample line for each new LOG_FILES entry"
    for name, line in samples.items():
        assert _parse_line_timestamp(line) is not None, (name, line)
    # gateway.error.log (launchd stderr, not in LOG_FILES) uses the shared stamper.
    from hermes_cli.stderr_timestamp import stamp_line
    assert _parse_line_timestamp(stamp_line("raw gateway stderr")) is not None