85 lines
3.1 KiB
Python
85 lines
3.1 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from kota_tech.core.logline import LineFramer, LineKind, parse_line
|
|
|
|
FIXTURES = sorted((Path(__file__).parent / "fixtures").glob("*.txt"))
|
|
|
|
|
|
def test_parse_esp_log_line():
|
|
ln = parse_line("E (21863) INFLUX: Failed to perform HTTP request: 0, ESP_ERR_HTTP_CONNECT")
|
|
assert ln.kind == LineKind.LOG
|
|
assert (ln.level, ln.uptime_ms, ln.tag) == ("E", 21863, "INFLUX")
|
|
assert ln.message.startswith("Failed to perform HTTP request")
|
|
|
|
|
|
def test_ansi_colour_stripped():
|
|
ln = parse_line("\x1b[0;33mW (14131) INFLUX: Initializing Influx client\x1b[0m\r")
|
|
assert ln.text == "W (14131) INFLUX: Initializing Influx client"
|
|
assert ln.level == "W"
|
|
|
|
|
|
def test_boot_and_reset_markers():
|
|
assert parse_line("ESP-ROM:esp32s3-20210327").kind == LineKind.BOOT
|
|
r = parse_line("rst:0xc (RTC_SW_CPU_RST),boot:0x8 (SPI_FAST_FLASH_BOOT)")
|
|
assert r.kind == LineKind.RESET and r.reset_reason == "RTC_SW_CPU_RST"
|
|
assert parse_line("abort() was called at PC 0x42008e1b on core 0").kind == LineKind.PANIC
|
|
|
|
|
|
def test_non_log_lines_are_raw():
|
|
assert parse_line("gather_task \tB\t1\t1992\t20").kind == LineKind.RAW
|
|
|
|
|
|
def test_framer_splits_and_idle_flushes():
|
|
f = LineFramer(idle_flush_s=0.3)
|
|
assert f.feed(b"I (1) A: one\r\nI (2) B: tw", now=0.0) == ["I (1) A: one\r"]
|
|
assert f.poll_idle(now=0.1) == []
|
|
assert f.poll_idle(now=0.5) == ["I (2) B: tw"]
|
|
|
|
|
|
def test_framer_drops_stale_fragment_after_open():
|
|
f = LineFramer(drop_first_partial=True, now=0.0)
|
|
assert f.feed(b"current time as t", now=0.01) == []
|
|
assert f.feed(b"imestamp\r\nW (15899) DeviceStatus: RTC\r\n", now=0.02) == ["W (15899) DeviceStatus: RTC\r"]
|
|
|
|
|
|
def test_framer_keeps_first_line_from_idle_board():
|
|
# Board was silent at open; the boot banner arriving later must survive.
|
|
f = LineFramer(drop_first_partial=True, now=0.0)
|
|
assert f.feed(b"", now=0.1) == []
|
|
assert f.feed(b"ESP-ROM:esp32s3-20210327\r\n", now=5.0) == ["ESP-ROM:esp32s3-20210327\r"]
|
|
|
|
|
|
def test_framer_cancel_drop():
|
|
f = LineFramer(drop_first_partial=True, now=0.0)
|
|
f.cancel_drop()
|
|
assert f.feed(b"ESP-ROM:esp32s3\n", now=0.01) == ["ESP-ROM:esp32s3"]
|
|
|
|
|
|
def test_framer_tolerates_invalid_utf8():
|
|
f = LineFramer()
|
|
assert f.feed(b"\xff\xfeI (1) X: y\n") == ["��I (1) X: y"]
|
|
|
|
|
|
@pytest.mark.parametrize("path", FIXTURES, ids=lambda p: p.name)
|
|
def test_replay_sample_log_chunked_equals_whole(path):
|
|
data = path.read_bytes()
|
|
whole = LineFramer().feed(data)
|
|
f = LineFramer()
|
|
chunked = []
|
|
for i in range(0, len(data), 37): # odd chunk size splits lines and UTF-8
|
|
chunked += f.feed(data[i : i + 37])
|
|
chunked += f.flush()
|
|
tail = LineFramer()
|
|
tail.feed(data)
|
|
assert chunked == whole + tail.flush()
|
|
|
|
|
|
@pytest.mark.parametrize("path", FIXTURES, ids=lambda p: p.name)
|
|
def test_replay_sample_log_parses(path):
|
|
lines = [parse_line(r) for r in LineFramer().feed(path.read_bytes())]
|
|
raw_errors = sum(1 for r in path.read_text("utf-8", "replace").splitlines() if r.startswith("E ("))
|
|
assert sum(1 for ln in lines if ln.level == "E") == raw_errors
|
|
assert raw_errors > 0
|