94 lines
3.6 KiB
Python
94 lines
3.6 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from kota_tech.core.logline import LineFramer, LineKind, parse_line, split_glued
|
|
|
|
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():
|
|
# Real capture: stale FIFO tail released together with the next live line.
|
|
f = LineFramer(drop_first_partial=True)
|
|
got = f.feed(b"I (96) boot: 6 ota_1 OTA app 00 11 00340000"
|
|
b"W (19497) BME688: RTC is uninitialized\r\nI (19820) SEN66: x\r\n")
|
|
assert got == ["W (19497) BME688: RTC is uninitialized\r", "I (19820) SEN66: x\r"]
|
|
|
|
|
|
def test_framer_keeps_first_line_from_idle_board():
|
|
# Board silent at open; the boot banner arriving later must survive.
|
|
f = LineFramer(drop_first_partial=True)
|
|
assert f.feed(b"ESP-ROM:esp32s3-20210327\r\n") == ["ESP-ROM:esp32s3-20210327\r"]
|
|
|
|
|
|
def test_framer_keeps_complete_first_line():
|
|
f = LineFramer(drop_first_partial=True)
|
|
assert f.feed(b"W (1) A: complete\r\n") == ["W (1) A: complete\r"]
|
|
|
|
|
|
def test_framer_splits_interleaved_lines():
|
|
f = LineFramer()
|
|
assert f.feed(b"I (2302) chip[DL]: I (2302) chip[DL]: Posting event\n") == [
|
|
"I (2302) chip[DL]: ", "I (2302) chip[DL]: Posting event"]
|
|
assert f.feed(b"wifi:\x1b[0;32mI (5) X: y\n") == ["wifi:", "\x1b[0;32mI (5) X: y"]
|
|
|
|
|
|
def test_framer_tolerates_invalid_utf8():
|
|
f = LineFramer()
|
|
# Garbage in front of a log line is split off rather than corrupting it.
|
|
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_parts = [p for r in path.read_text("utf-8", "replace").splitlines() for p in split_glued(r)]
|
|
raw_errors = sum(1 for p in raw_parts if p.startswith("E ("))
|
|
assert sum(1 for ln in lines if ln.level == "E") == raw_errors
|
|
assert raw_errors > 0
|