| """Tests for hermes_cli.logs — log viewing and filtering.""" |
|
|
| import os |
| from datetime import datetime, timedelta |
| from pathlib import Path |
|
|
| import pytest |
|
|
| 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, |
| ) |
|
|
|
|
| |
| |
| |
|
|
| 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_minutes(self): |
| cutoff = _parse_since("30m") |
| assert cutoff is not None |
| assert abs((datetime.now() - cutoff).total_seconds() - 1800) < 2 |
|
|
| def test_days(self): |
| cutoff = _parse_since("1d") |
| assert cutoff is not None |
| assert abs((datetime.now() - cutoff).total_seconds() - 86400) < 2 |
|
|
| def test_seconds(self): |
| cutoff = _parse_since("120s") |
| assert cutoff is not None |
| assert abs((datetime.now() - cutoff).total_seconds() - 120) < 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) |
|
|
| def test_no_timestamp(self): |
| assert _parse_line_timestamp("no timestamp here") is None |
|
|
|
|
| class TestExtractLevel: |
| def test_info(self): |
| assert _extract_level("2026-01-01 00:00:00 INFO gateway.run: msg") == "INFO" |
|
|
| def test_warning(self): |
| assert _extract_level("2026-01-01 00:00:00 WARNING tools.file: msg") == "WARNING" |
|
|
| def test_error(self): |
| assert _extract_level("2026-01-01 00:00:00 ERROR run_agent: msg") == "ERROR" |
|
|
| def test_debug(self): |
| assert _extract_level("2026-01-01 00:00:00 DEBUG agent.aux: msg") == "DEBUG" |
|
|
| def test_no_level(self): |
| assert _extract_level("random text") is None |
|
|
|
|
| |
| |
| |
|
|
| 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_nested_logger(self): |
| line = "2026-04-11 10:23:45 INFO gateway.platforms.telegram: connected" |
| assert _extract_logger_name(line) == "gateway.platforms.telegram" |
|
|
| def test_warning_level(self): |
| line = "2026-04-11 10:23:45 WARNING tools.terminal_tool: timeout" |
| assert _extract_logger_name(line) == "tools.terminal_tool" |
|
|
| def test_with_session_tag(self): |
| line = "2026-04-11 10:23:45 INFO [abc123] tools.file_tools: reading file" |
| assert _extract_logger_name(line) == "tools.file_tools" |
|
|
| def test_with_session_tag_and_error(self): |
| line = "2026-04-11 10:23:45 ERROR [sess_xyz] agent.context_compressor: failed" |
| assert _extract_logger_name(line) == "agent.context_compressor" |
|
|
| def test_top_level_module(self): |
| line = "2026-04-11 10:23:45 INFO run_agent: starting conversation" |
| assert _extract_logger_name(line) == "run_agent" |
|
|
| def test_no_match(self): |
| assert _extract_logger_name("random text") is None |
|
|
|
|
| class TestLineMatchesComponent: |
| def test_gateway_component(self): |
| line = "2026-04-11 10:23:45 INFO gateway.run: msg" |
| assert _line_matches_component(line, ("gateway",)) |
|
|
| def test_gateway_nested(self): |
| line = "2026-04-11 10:23:45 INFO gateway.platforms.telegram: msg" |
| assert _line_matches_component(line, ("gateway",)) |
|
|
| def test_tools_component(self): |
| line = "2026-04-11 10:23:45 INFO tools.terminal_tool: msg" |
| assert _line_matches_component(line, ("tools",)) |
|
|
| def test_agent_with_multiple_prefixes(self): |
| prefixes = ("agent", "run_agent", "model_tools") |
| assert _line_matches_component( |
| "2026-04-11 10:23:45 INFO agent.context_compressor: msg", prefixes) |
| assert _line_matches_component( |
| "2026-04-11 10:23:45 INFO run_agent: msg", prefixes) |
| assert _line_matches_component( |
| "2026-04-11 10:23:45 INFO model_tools: msg", prefixes) |
|
|
| def test_no_match(self): |
| line = "2026-04-11 10:23:45 INFO tools.browser: msg" |
| assert not _line_matches_component(line, ("gateway",)) |
|
|
| def test_with_session_tag(self): |
| line = "2026-04-11 10:23:45 INFO [abc] gateway.run: msg" |
| assert _line_matches_component(line, ("gateway",)) |
|
|
| def test_unparseable_line(self): |
| assert not _line_matches_component("random text", ("gateway",)) |
|
|
|
|
| |
| |
| |
|
|
| class TestMatchesFilters: |
| def test_no_filters_passes_everything(self): |
| assert _matches_filters("any line") |
|
|
| 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_session_filter(self): |
| assert _matches_filters( |
| "2026-01-01 00:00:00 INFO [abc123] x: msg", session_filter="abc123") |
| assert not _matches_filters( |
| "2026-01-01 00:00:00 INFO [xyz789] x: msg", session_filter="abc123") |
|
|
| def test_component_filter(self): |
| assert _matches_filters( |
| "2026-01-01 00:00:00 INFO gateway.run: msg", |
| component_prefixes=("gateway",)) |
| assert not _matches_filters( |
| "2026-01-01 00:00:00 INFO tools.file: msg", |
| component_prefixes=("gateway",)) |
|
|
| 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",), |
| ) |
| |
| assert not _matches_filters( |
| line, |
| min_level="WARNING", |
| session_filter="sess_1", |
| component_prefixes=("tools",), |
| ) |
|
|
| def test_since_filter(self): |
| |
| assert not _matches_filters( |
| "2020-01-01 00:00:00 INFO x: old msg", |
| since=datetime.now() - timedelta(hours=1)) |
| |
| 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)) |
|
|
|
|
| |
| |
| |
|
|
| 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_read_with_component_filter(self, tmp_path): |
| log_file = tmp_path / "test.log" |
| lines = [ |
| "2026-01-01 00:00:00 INFO gateway.run: gw msg\n", |
| "2026-01-01 00:00:01 INFO tools.file: tool msg\n", |
| "2026-01-01 00:00:02 INFO gateway.session: session msg\n", |
| "2026-01-01 00:00:03 INFO agent.compressor: agent msg\n", |
| ] |
| log_file.write_text("".join(lines)) |
|
|
| result = _read_tail( |
| log_file, 50, |
| has_filters=True, |
| component_prefixes=("gateway",), |
| ) |
| assert len(result) == 2 |
| assert "gw msg" in result[0] |
| assert "session msg" in result[1] |
|
|
| def test_empty_file(self, tmp_path): |
| log_file = tmp_path / "empty.log" |
| log_file.write_text("") |
| result = _read_last_n_lines(log_file, 10) |
| assert result == [] |
|
|
|
|
| |
| |
| |
|
|
| class TestLogFiles: |
| def test_known_log_files(self): |
| assert "agent" in LOG_FILES |
| assert "errors" in LOG_FILES |
| assert "gateway" in LOG_FILES |
|
|
