stackchain-dashboard/tests/test_live_snapshot_store.py

import threading
import os

import pytest

from src import live_snapshot_store
from src.live_snapshot_store import LiveSnapshotStore, RefreshLeaseLost


def test_independent_stores_racing_for_refresh_have_one_lease_winner(tmp_path):
    path = tmp_path / "live.sqlite3"
    first = LiveSnapshotStore(path, clock=lambda: 100.0)
    second = LiveSnapshotStore(path, clock=lambda: 100.0)
    barrier = threading.Barrier(2)
    results = []

    def acquire(store):
        barrier.wait()
        results.append(store.try_acquire_refresh({"context", "events"}, lease_seconds=5))

    threads = [threading.Thread(target=acquire, args=(store,)) for store in (first, second)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()

    assert sum(result is not None for result in results) == 1


def test_snapshot_and_shared_revisions_publish_atomically(tmp_path):
    path = tmp_path / "live.sqlite3"
    writer = LiveSnapshotStore(path, clock=lambda: 100.0)
    reader = LiveSnapshotStore(path, clock=lambda: 100.0)
    lease = writer.try_acquire_refresh(set(("context", "events", "notifications")), lease_seconds=5)
    assert lease is not None

    state = writer.publish_refresh(
        lease,
        value={
            "context": {"generation": 1},
            "events": [],
            "notifications": [],
            "sections": {section: "fresh" for section in ("context", "events", "notifications")},
        },
        created_at={section: 100.0 for section in ("context", "events", "notifications")},
        failure_count={section: 0 for section in ("context", "events", "notifications")},
        retry_at={section: None for section in ("context", "events", "notifications")},
        changed_sections=set(("context", "events", "notifications")),
    )

    observed = reader.load()
    assert observed.value == state.value
    assert observed.revisions == {"context": 1, "events": 1, "notifications": 1}
    assert observed.refreshing_sections == set()
    with pytest.raises(RefreshLeaseLost):
        writer.publish_refresh(
            lease,
            value={"context": {"generation": 2}},
            created_at=observed.created_at,
            failure_count=observed.failure_count,
            retry_at=observed.retry_at,
            changed_sections={"context"},
        )


def test_expired_refresh_lease_can_be_recovered(tmp_path):
    now = 100.0
    path = tmp_path / "live.sqlite3"
    abandoned = LiveSnapshotStore(path, clock=lambda: now)
    recovery = LiveSnapshotStore(path, clock=lambda: now)

    first = abandoned.try_acquire_refresh({"context"}, lease_seconds=5)
    assert first is not None
    assert recovery.try_acquire_refresh({"context"}, lease_seconds=5) is None
    now = 106.0
    second = recovery.try_acquire_refresh({"context"}, lease_seconds=5)

    assert second is not None
    assert second != first


def test_default_store_clock_uses_reboot_stable_wall_time(tmp_path, monkeypatch):
    epoch = 1_700_000_000.0
    monkeypatch.setattr(live_snapshot_store.time, "time", lambda: epoch)
    store = LiveSnapshotStore(tmp_path / "live.sqlite3")

    lease = store.try_acquire_refresh({"context"}, lease_seconds=5)

    assert lease is not None
    assert store.load().lease_expires_at == epoch + 5


def test_store_is_private_and_does_not_persist_upstream_token(tmp_path, monkeypatch):
    token = "gitea-super-secret-token-material"
    monkeypatch.setenv("GITEA_TOKEN", token)
    path = tmp_path / "private-state" / "live.sqlite3"

    LiveSnapshotStore(path)

    assert os.stat(path).st_mode & 0o777 == 0o600
    assert os.stat(path.parent).st_mode & 0o777 == 0o700
    assert token.encode() not in path.read_bytes()


def test_remove_notifications_filters_batch_with_one_revision_advance(tmp_path):
    store = LiveSnapshotStore(tmp_path / "live.sqlite3", clock=lambda: 100.0)
    lease = store.try_acquire_refresh({"notifications"}, lease_seconds=5)
    assert lease is not None
    metadata = {section: None for section in ("context", "events", "notifications")}
    published = store.publish_refresh(
        lease,
        value={
            "notifications": [{"id": 7}, {"id": 8}, {"id": 9}],
            "sections": {"notifications": "fresh"},
        },
        created_at=metadata,
        failure_count={section: 0 for section in metadata},
        retry_at=metadata,
        changed_sections={"notifications"},
    )

    removed = store.remove_notifications([7, 9, 7])

    assert removed.value["notifications"] == [{"id": 8}]
    assert removed.revisions["notifications"] == published.revisions["notifications"] + 1


def test_read_notification_filter_is_shared_with_future_publications(tmp_path):
    path = tmp_path / "live.sqlite3"
    first = LiveSnapshotStore(path, clock=lambda: 100.0)
    second = LiveSnapshotStore(path, clock=lambda: 100.0)
    lease = first.try_acquire_refresh({"notifications"}, lease_seconds=5)
    assert lease is not None
    metadata = {section: None for section in ("context", "events", "notifications")}
    first.publish_refresh(
        lease,
        value={"notifications": [{"id": 7}, {"id": 8}], "sections": {"notifications": "fresh"}},
        created_at=metadata,
        failure_count={section: 0 for section in metadata},
        retry_at=metadata,
        changed_sections={"notifications"},
    )

    removed = second.remove_notification(7)
    lease = first.try_acquire_refresh({"notifications"}, lease_seconds=5)
    assert lease is not None
    republished = first.publish_refresh(
        lease,
        value={"notifications": [{"id": 7}, {"id": 8}], "sections": {"notifications": "fresh"}},
        created_at=metadata,
        failure_count={section: 0 for section in metadata},
        retry_at=metadata,
        changed_sections={"notifications"},
    )

    assert removed.value["notifications"] == [{"id": 8}]
    assert republished.value["notifications"] == [{"id": 8}]