SortedDict timeline queries for event windows under pytest: working versions (tested 2026-10-15)

sortedcontainers==2.4.0
pytest==9.1.1

from __future__ import annotations

from sortedcontainers import SortedDict

# Boundary spec for SortedDict.irange: (include_min, include_max).
Inclusive = tuple[bool, bool]


class EventTimeline:
    """In-memory event log keyed by integer timestamp (e.g. milliseconds).

    A SortedDict keeps keys ordered no matter the arrival order, so locating a
    window costs O(log n) and multiple events can share one timestamp by
    mapping it to a list.
    """

    def __init__(self) -> None:
        self._events: SortedDict[int, list[str]] = SortedDict()

    def add(self, ts: int, name: str) -> None:
        self._events.setdefault(ts, []).append(name)

    def __len__(self) -> int:
        return sum(len(names) for names in self._events.values())

    def window(self, start: int, end: int, *, inclusive: Inclusive = (True, True)) -> list[str]:
        keys = self._events.irange(start, end, inclusive=inclusive)
        return [name for ts in keys for name in self._events[ts]]

    def latest(self, n: int) -> list[tuple[int, str]]:
        pairs: list[tuple[int, str]] = []
        for ts in reversed(self._events.keys()):
            pairs.extend((ts, name) for name in reversed(self._events[ts]))
            if len(pairs) >= n:
                break
        return list(reversed(pairs))[-n:] if n > 0 else []

    def span(self) -> tuple[int, int] | None:
        if not self._events:
            return None
        return self._events.peekitem(0)[0], self._events.peekitem(-1)[0]

    def prune_before(self, ts: int) -> int:
        stale = list(self._events.irange(maximum=ts, inclusive=(True, False)))
        for key in stale:
            del self._events[key]
        return len(stale)

import pytest
from sortedcontainers import SortedDict

from timeline import EventTimeline


def test_arrival_order_does_not_matter():
    tl = EventTimeline()
    for ts, name in [(30, "c"), (10, "a"), (20, "b")]:
        tl.add(ts, name)
    assert tl.window(0, 100) == ["a", "b", "c"]
    assert tl.span() == (10, 30)


def test_same_timestamp_keeps_arrival_order():
    tl = EventTimeline()
    tl.add(100, "open")
    tl.add(100, "close")
    assert tl.window(100, 100) == ["open", "close"]
    assert len(tl) == 2


def test_window_bounds_follow_tuple_inclusive():
    tl = EventTimeline()
    for ts in (10, 20, 30):
        tl.add(ts, str(ts))
    assert tl.window(10, 30) == ["10", "20", "30"]
    assert tl.window(10, 30, inclusive=(True, False)) == ["10", "20"]
    assert tl.window(10, 30, inclusive=(False, True)) == ["20", "30"]
    assert tl.window(40, 99) == []


def test_irange_requires_tuple_not_bool():
    # A bool looks natural but irange() subscripts inclusive[0] and blows up.
    with pytest.raises(TypeError):
        list(SortedDict({1: "a"}).irange(0, 5, inclusive=False))


def test_latest_returns_chronological_tail():
    tl = EventTimeline()
    for ts, name in [(10, "a"), (10, "b"), (20, "c"), (30, "d")]:
        tl.add(ts, name)
    assert tl.latest(2) == [(20, "c"), (30, "d")]
    assert tl.latest(0) == []


def test_prune_before_is_exclusive_and_empties():
    tl = EventTimeline()
    for ts in (10, 20, 30):
        tl.add(ts, str(ts))
    assert tl.prune_before(30) == 2
    assert tl.window(0, 100) == ["30"]
    assert tl.prune_before(99) == 1
    assert tl.span() is None

[pytest]
pythonpath = .