"""The ulogcat parser: framing, the optional fields, and what a live log does to it.

Frames are built here from the format rather than copied from the aircraft,
because the aircraft's log is operational data. The one test that reads the
real capture is skipped wherever that capture is not present, which is
everywhere but the machine it was fetched on.
"""

import struct
from pathlib import Path

import pytest

from mcbebop.files import ulog

SAMPLE = Path(__file__).resolve().parents[1] / "captures" / "logs" / "ckcm.bin"


def pstr(text: str) -> bytes:
    raw = text.encode()
    assert len(raw) < 256
    return bytes([len(raw)]) + raw


def frame(kind: int, body: bytes) -> bytes:
    return ulog.FRAME_START + bytes([kind]) + body + ulog.FRAME_END


def header(
    *,
    priority: str = "I",
    uptime_us: int = 1_000_000,
    tid: int = 42,
    source: str | None = None,
    tag: str | None = "TEST",
    colour: bytes | None = None,
) -> bytes:
    flags = ulog.FLAG_DATE | ulog.FLAG_TID
    if colour is not None:
        flags |= ulog.FLAG_COLOUR
    if source is not None:
        flags |= ulog.FLAG_NAME
    if tag is not None:
        flags |= ulog.FLAG_TAG
    body = bytes([flags, ord(priority)])
    if colour is not None:
        body += colour
    body += struct.pack("<Qi", uptime_us, tid)
    if source is not None:
        body += pstr(source)
    if tag is not None:
        body += pstr(tag)
    return frame(ulog.KIND_HEADER, body)


def message(text: str, colour: bytes | None = None) -> bytes:
    if colour is None:
        return frame(ulog.KIND_MESSAGE, pstr(text))
    return frame(ulog.KIND_MESSAGE_COLOURED, colour + bytes([ulog.KIND_MESSAGE]) + pstr(text))


def entry(text: str = "hello", **kw) -> bytes:
    return header(**kw) + message(text, colour=kw.get("colour"))


# --- the framing -------------------------------------------------------------


def test_an_entry_is_a_header_frame_and_a_message_frame():
    (record,) = ulog.parse(entry("Machine: Milos board", tag="KERNEL", priority="W"))
    assert record.tag == "KERNEL"
    assert record.priority == "W"
    assert record.level == "warning"
    assert record.message == "Machine: Milos board"


def test_the_timestamp_is_microseconds_since_boot():
    (record,) = ulog.parse(entry(uptime_us=2_067_211_079))
    assert record.uptime_us == 2_067_211_079
    assert record.uptime == pytest.approx(2067.211079)


def test_both_name_fields_are_optional_and_independent():
    data = (
        entry("has both", source="dragon-prog/Behaviour", tag="COMMANDS")
        + entry("tag only", source=None, tag="KERNEL")
        + entry("source only", source="ephemerisd", tag=None)
    )
    records = list(ulog.parse(data))
    assert [(r.source, r.tag) for r in records] == [
        ("dragon-prog/Behaviour", "COMMANDS"),
        ("", "KERNEL"),
        ("ephemerisd", ""),
    ]


def test_the_thread_id_comes_through():
    (record,) = ulog.parse(entry(tid=1200))
    assert record.tid == 1200


def test_a_coloured_entry_keeps_the_headers_colour_and_still_reads_its_text():
    # The data frame clamps each channel to a minimum of 1, so its copy of the
    # colour disagrees with the header's for any channel the drone set to 0.
    # The header's is the true one.
    data = header(colour=b"\xff\xd0\x00") + message("Switch AWB to AUTO", colour=b"\xff\xd0\x01")
    (record,) = ulog.parse(data)
    assert record.colour == b"\xff\xd0\x00"
    assert record.message == "Switch AWB to AUTO"


def test_an_uncoloured_entry_has_no_colour():
    (record,) = ulog.parse(entry("plain"))
    assert record.colour is None


def test_a_message_of_the_maximum_length_round_trips():
    text = "x" * 255
    (record,) = ulog.parse(entry(text))
    assert record.message == text


def test_the_trailing_newline_the_drone_printed_is_dropped_but_inner_ones_stay():
    (record,) = ulog.parse(entry("first\nsecond\n"))
    assert record.message == "first\nsecond"


def test_every_entry_reports_where_in_the_file_it_came_from():
    first = entry("a")
    records = list(ulog.parse(first + entry("b")))
    assert [r.offset for r in records] == [0, len(first)]


# --- a log that is still being written ---------------------------------------


def test_a_fetch_that_lands_mid_record_keeps_everything_before_it():
    data = entry("complete") + entry("cut in half")
    for cut in range(len(entry("complete")) + 1, len(data)):
        stats = ulog.ParseStats()
        records = list(ulog.parse(data[:cut], stats))
        assert [r.message for r in records] == ["complete"], f"truncated at {cut}"
        assert stats.truncated or stats.unpaired, f"truncation at {cut} went unreported"


def test_a_truncation_in_the_very_first_record_yields_nothing_and_does_not_raise():
    stats = ulog.ParseStats()
    assert list(ulog.parse(entry("x")[:6], stats)) == []
    assert stats.truncated


def test_a_fetch_that_starts_mid_record_resyncs_on_the_next_marker():
    data = entry("first") + entry("second")
    stats = ulog.ParseStats()
    records = list(ulog.parse(data[5:], stats))
    assert [r.message for r in records] == ["second"]


def test_garbage_between_entries_costs_one_resync_and_nothing_else():
    # Nothing is pending here, so an unreadable frame is damage rather than a
    # binary payload, and reading it as damage is what stops it swallowing the
    # header that follows.
    data = entry("before") + ulog.FRAME_START + b"\x99junk" + ulog.FRAME_END + entry("after")
    stats = ulog.ParseStats()
    records = list(ulog.parse(data, stats))
    assert [r.message for r in records] == ["before", "after"]
    assert stats.resyncs == 1


def test_a_header_with_no_message_after_it_is_counted_not_guessed():
    stats = ulog.ParseStats()
    records = list(ulog.parse(header() + entry("real"), stats))
    assert [r.message for r in records] == ["real"]
    assert stats.unpaired == 1


def test_a_message_with_no_header_before_it_is_dropped():
    stats = ulog.ParseStats()
    records = list(ulog.parse(message("orphan") + entry("real"), stats))
    assert [r.message for r in records] == ["real"]
    assert stats.unpaired == 1


@pytest.mark.parametrize("bit", [ulog.FLAG_PC, ulog.FLAG_THREAD_PRIORITY, 0x80])
def test_a_header_whose_opt_bits_describe_a_shape_no_source_documents_is_refused(bit):
    # The renderer never writes PC or thread-priority, so nothing public says
    # how wide they are. An unread field makes every field after it wrong, so
    # skipping the record beats reporting a plausible lie.
    flags = ulog.FLAG_DATE | ulog.FLAG_TID | ulog.FLAG_TAG | bit
    bad = ulog.FRAME_START + bytes([ulog.KIND_HEADER, flags, ord("I")])
    bad += struct.pack("<Qi", 1, 1) + pstr("TAG") + ulog.FRAME_END
    stats = ulog.ParseStats()
    records = list(ulog.parse(bad + message("orphaned by the refusal") + entry("real"), stats))
    assert [r.message for r in records] == ["real"]
    assert stats.resyncs == 1


def test_a_binary_payload_is_kept_rather_than_losing_the_entry():
    # A ulog_bin entry's data frame has no kind byte and no length, so there
    # is nothing to decode and nothing to check it against. Keeping the bytes
    # and saying so beats dropping the entry and its header with it.
    payload = b"\x00\x01\x02\x03\x04"
    data = header(tag="BIN") + ulog.FRAME_START + payload + ulog.FRAME_END
    stats = ulog.ParseStats()
    (record,) = ulog.parse(data, stats)
    assert record.binary == payload
    assert record.tag == "BIN"
    assert "binary" in record.message
    assert stats.resyncs == 0


def test_a_buffer_with_no_marker_at_all_is_empty_rather_than_an_error():
    stats = ulog.ParseStats()
    assert list(ulog.parse(b"not a ulogcat file at all", stats)) == []
    assert not stats.truncated


def test_undecodable_text_does_not_stop_the_parse():
    body = bytes([ulog.FLAG_TAG, ord("I")]) + struct.pack("<QI", 1, 1) + pstr("T")
    data = frame(ulog.KIND_HEADER, body) + frame(ulog.KIND_MESSAGE, b"\x03\xff\xfe\xfd")
    (record,) = ulog.parse(data)
    assert record.message  # replacement characters, not an exception


# --- priorities --------------------------------------------------------------


def test_priority_letters_spell_out_to_syslog_levels():
    assert ulog.PRIORITY_NAMES["E"] == "error"
    assert ulog.rank("E") < ulog.rank("W") < ulog.rank("I") < ulog.rank("D")


def test_there_is_no_notice_level_because_the_renderer_writes_notice_as_info():
    # cprio[] in libulogcat_ckcm.c maps both ULOG_NOTICE and ULOG_INFO to 'I',
    # with a FIXME saying the CKCM tooling does not know NOTICE. Offering a
    # notice level here would be offering a filter that can never match.
    assert "N" not in ulog.PRIORITY_NAMES


def test_an_unknown_priority_letter_survives_and_never_filters_out_silently():
    (record,) = ulog.parse(entry(priority="Z"))
    assert record.priority == "Z"
    assert record.level == "Z"
    assert ulog.rank("Z") == ulog.UNKNOWN_RANK


# --- statistics --------------------------------------------------------------


def test_the_tag_histogram_covers_everything_parsed():
    data = entry("a", tag="KERNEL") + entry("b", tag="KERNEL") + entry("c", tag="COMMANDS")
    stats = ulog.ParseStats()
    list(ulog.parse(data, stats))
    assert stats.entries == 3
    assert stats.tags == {"KERNEL": 2, "COMMANDS": 1}


# --- the real thing ----------------------------------------------------------


@pytest.mark.skipif(not SAMPLE.exists(), reason="no captured log on this machine")
def test_the_captured_log_parses_to_the_byte():
    # The framing was derived from this file, so the bar is total: every
    # header consumed exactly, nothing skipped, no residue. A resync here
    # would mean a field shape that was missed.
    stats = ulog.ParseStats()
    records = list(ulog.parse(SAMPLE.read_bytes(), stats))
    assert stats.resyncs == 0
    assert stats.unpaired == 0
    assert len(records) == stats.entries > 1000
    assert all(r.priority in ulog.PRIORITY_NAMES for r in records)
    assert "KERNEL" in stats.tags
    # Monotonic apart from the kernel stream, which ulogcat merges in from
    # /proc/kmsg and which can land a few milliseconds out of order.
    userspace = [r.uptime_us for r in records if r.tag != "KERNEL"]
    assert userspace == sorted(userspace)