tests/test_durable_jsonl_log.py

import json

from app.services.durable_jsonl_log import advance_cursor, append_line, read_pending_range


def test_append_and_read_pending_range_returns_new_lines(tmp_path):
    jsonl_path = tmp_path / "log.jsonl"
    cursor_path = tmp_path / "log.cursor.json"

    append_line(jsonl_path, {"id": 1})
    append_line(jsonl_path, {"id": 2})

    lines, from_offset, through_offset = read_pending_range(jsonl_path, cursor_path, "reviewed_through_offset")

    assert [json.loads(line)["id"] for line in lines] == [1, 2]
    assert from_offset == 0
    assert through_offset == jsonl_path.stat().st_size


def test_read_pending_range_excludes_entries_before_cursor(tmp_path):
    jsonl_path = tmp_path / "log.jsonl"
    cursor_path = tmp_path / "log.cursor.json"

    append_line(jsonl_path, {"id": 1})
    _, _, through_offset = read_pending_range(jsonl_path, cursor_path, "reviewed_through_offset")
    advance_cursor(cursor_path, "reviewed_through_offset", through_offset)

    append_line(jsonl_path, {"id": 2})
    lines, from_offset, _ = read_pending_range(jsonl_path, cursor_path, "reviewed_through_offset")

    assert [json.loads(line)["id"] for line in lines] == [2]
    assert from_offset == through_offset


def test_read_pending_range_ignores_entries_appended_after_eof_snapshot(tmp_path):
    jsonl_path = tmp_path / "log.jsonl"
    cursor_path = tmp_path / "log.cursor.json"

    append_line(jsonl_path, {"id": 1})
    lines, from_offset, through_offset = read_pending_range(jsonl_path, cursor_path, "reviewed_through_offset")

    # Simulates a message arriving while a Commit's read is already in flight.
    append_line(jsonl_path, {"id": 2})

    assert [json.loads(line)["id"] for line in lines] == [1]
    advance_cursor(cursor_path, "reviewed_through_offset", through_offset)

    remaining, _, _ = read_pending_range(jsonl_path, cursor_path, "reviewed_through_offset")
    assert [json.loads(line)["id"] for line in remaining] == [2]


def test_advance_cursor_is_atomic_replace(tmp_path):
    cursor_path = tmp_path / "log.cursor.json"
    advance_cursor(cursor_path, "reviewed_through_offset", 100)
    advance_cursor(cursor_path, "reviewed_through_offset", 250)

    assert json.loads(cursor_path.read_text())["reviewed_through_offset"] == 250


def test_read_exact_range_ignores_entries_appended_after_the_range(tmp_path):
    from app.services.durable_jsonl_log import read_exact_range

    jsonl_path = tmp_path / "log.jsonl"
    append_line(jsonl_path, {"id": 1})
    _, from_offset, through_offset = read_pending_range(jsonl_path, tmp_path / "log.cursor.json", "reviewed_through_offset")

    append_line(jsonl_path, {"id": 2})  # appended after the range was captured

    lines = read_exact_range(jsonl_path, from_offset, through_offset)
    assert [json.loads(line)["id"] for line in lines] == [1]