stackchain-dashboard/tests/test_available_issue_snapshot_store.py

import threading
import os
import sqlite3

import pytest

from src.available_issue_snapshot_store import AvailableIssueSnapshotStore
from src.state_encryption import PrivateStateEncryptionError


PRIVATE_KEY = b"a" * 32


def test_published_find_work_catalog_is_encrypted_at_rest_and_survives_restart(tmp_path):
    path = tmp_path / "available.sqlite3"
    store = AvailableIssueSnapshotStore(path, clock=lambda: 100.0, encryption_key=PRIVATE_KEY)
    owner = store.try_acquire_refresh(lease_seconds=5)
    canary = "private-find-work-body-canary"

    published = store.publish(
        owner,
        items=[{"repository": "stackchain/api", "number": 7, "body": canary}],
    )

    with sqlite3.connect(path) as connection:
        payload = connection.execute(
            "SELECT items_json FROM available_issue_snapshot WHERE singleton = 1"
        ).fetchone()[0]
    assert payload.startswith("v1:")
    assert canary not in payload
    assert AvailableIssueSnapshotStore(
        path, clock=lambda: 100.0, encryption_key=PRIVATE_KEY
    ).load().items == published.items


def test_find_work_catalog_lazily_migrates_plaintext_without_changing_freshness(tmp_path):
    path = tmp_path / "available.sqlite3"
    store = AvailableIssueSnapshotStore(path, clock=lambda: 100.0, encryption_key=PRIVATE_KEY)
    with sqlite3.connect(path) as connection:
        connection.execute(
            "UPDATE available_issue_snapshot SET items_json = ?, created_at = ?, retry_at = ?",
            ('[{"repository":"stackchain/api","number":7}]', 91.0, 105.0),
        )

    state = store.load()

    with sqlite3.connect(path) as connection:
        migrated = connection.execute(
            "SELECT items_json FROM available_issue_snapshot WHERE singleton = 1"
        ).fetchone()[0]
    assert state.items == [{"repository": "stackchain/api", "number": 7}]
    assert (state.created_at, state.retry_at) == (91.0, 105.0)
    assert migrated.startswith("v1:")


def test_find_work_ciphertext_cannot_be_substituted_into_live_snapshot(tmp_path):
    available_path = tmp_path / "available.sqlite3"
    available = AvailableIssueSnapshotStore(
        available_path, clock=lambda: 100.0, encryption_key=PRIVATE_KEY
    )
    owner = available.try_acquire_refresh(lease_seconds=5)
    available.publish(owner, items=[{"body": "secret"}])
    with sqlite3.connect(available_path) as connection:
        payload = connection.execute(
            "SELECT items_json FROM available_issue_snapshot WHERE singleton = 1"
        ).fetchone()[0]

    live_path = tmp_path / "live.sqlite3"
    from src.live_snapshot_store import LiveSnapshotStore
    live = LiveSnapshotStore(live_path, encryption_key=PRIVATE_KEY)
    with sqlite3.connect(live_path) as connection:
        connection.execute("UPDATE live_snapshot SET value_json = ?", (payload,))

    with pytest.raises(PrivateStateEncryptionError, match="private state could not be decrypted"):
        live.load()


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

    def acquire(store):
        barrier.wait()
        results.append(store.try_acquire_refresh(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(owner is not None for owner in results) == 1


def test_published_catalog_and_metadata_are_visible_to_another_worker(tmp_path):
    path = tmp_path / "available.sqlite3"
    writer = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
    reader = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
    owner = writer.try_acquire_refresh(lease_seconds=5)

    writer.publish(owner, items=[{"repository": "stackchain/api", "number": 7}])

    state = reader.load()
    assert state.items == [{"repository": "stackchain/api", "number": 7}]
    assert state.created_at == 100.0
    assert state.retry_at is None
    assert state.refreshing is False


def test_failed_refresh_backoff_is_shared_and_lease_is_released(tmp_path):
    path = tmp_path / "available.sqlite3"
    writer = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
    reader = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
    owner = writer.try_acquire_refresh(lease_seconds=5)

    writer.fail_refresh(owner, retry_at=105.0)

    state = reader.load()
    assert state.retry_at == 105.0
    assert state.refreshing is False
    assert reader.try_acquire_refresh(lease_seconds=5) is not None


def test_confirmed_claim_is_removed_for_every_worker(tmp_path):
    path = tmp_path / "available.sqlite3"
    writer = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
    reader = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
    owner = writer.try_acquire_refresh(lease_seconds=5)
    writer.publish(owner, items=[
        {"repository": "stackchain/api", "number": 7},
        {"repository": "stackchain/web", "number": 8},
    ])

    writer.remove_claimed("stackchain/api", 7)

    assert reader.load().items == [{"repository": "stackchain/web", "number": 8}]


def test_invalidation_clears_catalog_and_claim_filters(tmp_path):
    store = AvailableIssueSnapshotStore(tmp_path / "available.sqlite3", clock=lambda: 100.0)
    owner = store.try_acquire_refresh(lease_seconds=5)
    store.publish(owner, items=[{"repository": "stackchain/api", "number": 7}])
    store.remove_claimed("stackchain/api", 7)

    store.invalidate()
    owner = store.try_acquire_refresh(lease_seconds=5)
    store.publish(owner, items=[{"repository": "stackchain/api", "number": 7}])

    assert store.load().items == [{"repository": "stackchain/api", "number": 7}]


def test_catalog_store_is_private(tmp_path):
    path = tmp_path / "state" / "available.sqlite3"

    AvailableIssueSnapshotStore(path)

    assert os.stat(path.parent).st_mode & 0o777 == 0o700
    assert os.stat(path).st_mode & 0o777 == 0o600


def test_cancelled_refresh_releases_its_lease_for_immediate_takeover(tmp_path):
    path = tmp_path / "available.sqlite3"
    first = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
    second = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
    owner = first.try_acquire_refresh(lease_seconds=30)

    first.release_refresh(owner)

    assert second.try_acquire_refresh(lease_seconds=30) is not None