nisse/nisse-core/tests/test_state.py

import sqlite3
from pathlib import Path

import pytest

from nisse_core.state.store import StateSchemaError, StateStore


@pytest.fixture
def db(tmp_path: Path) -> Path:
    return tmp_path / "state.db"


def test_wal_mode_and_full_sync(db: Path) -> None:
    with StateStore(db) as store:
        journal = store._conn.execute("PRAGMA journal_mode").fetchone()[0]
        sync = store._conn.execute("PRAGMA synchronous").fetchone()[0]
    assert journal == "wal"
    assert sync == 2  # FULL


def test_fresh_store_stamps_schema_version(db: Path) -> None:
    with StateStore(db, schema_version=1) as store:
        assert store.schema_version == 1


def test_a_work_item_persists_across_reopen(db: Path) -> None:
    with StateStore(db) as store:
        store.open_work_item(
            "demo-1", title="t", scope="demo", position="WAITING", cause="test"
        )
        store.set_active("demo-1", cause="test")
        store.update_work_item(
            "demo-1", position="RUNNING", data={"milestone": 2}, cause="test"
        )
    with StateStore(db) as store:
        item = store.active_item()
    assert item is not None
    assert (item.id, item.scope, item.position) == ("demo-1", "demo", "RUNNING")
    assert item.data == {"milestone": 2}


def test_a_fresh_store_has_no_active_item(db: Path) -> None:
    with StateStore(db) as store:
        assert store.active_item() is None


def test_intent_lifecycle(db: Path) -> None:
    with StateStore(db) as store:
        intent_id = store.begin_intent("push_branch", {"branch": "m2"}, cause="test")
        [pending] = store.pending_intents()
        assert pending.id == intent_id
        assert pending.kind == "push_branch"
        assert pending.payload == {"branch": "m2"}
        store.commit_intent(intent_id, cause="test")
        assert store.pending_intents() == []


def test_pending_intent_survives_reopen(db: Path) -> None:
    with StateStore(db) as store:
        store.begin_intent("open_pr", {"title": "hello"}, cause="test")
    with StateStore(db) as store:
        [pending] = store.pending_intents()
    assert pending.kind == "open_pr"


def test_aborted_intent_is_not_pending(db: Path) -> None:
    with StateStore(db) as store:
        intent_id = store.begin_intent("push_branch", {}, cause="test")
        store.abort_intent(intent_id, cause="test")
        assert store.pending_intents() == []


def test_resolving_nonpending_intent_raises(db: Path) -> None:
    with StateStore(db) as store:
        intent_id = store.begin_intent("push_branch", {}, cause="test")
        store.commit_intent(intent_id, cause="test")
        with pytest.raises(KeyError):
            store.commit_intent(intent_id, cause="test")
        with pytest.raises(KeyError):
            store.abort_intent(9999, cause="test")


def test_migration_runs_and_bumps_version(db: Path) -> None:
    with StateStore(db, schema_version=1):
        pass

    def add_notes_table(conn: sqlite3.Connection) -> None:
        conn.execute("CREATE TABLE notes (id INTEGER PRIMARY KEY, body TEXT)")

    with StateStore(db, schema_version=2, migrations={1: add_notes_table}) as store:
        assert store.schema_version == 2
        store._conn.execute("SELECT * FROM notes")


def test_missing_migration_raises(db: Path) -> None:
    with StateStore(db, schema_version=1):
        pass
    with pytest.raises(StateSchemaError, match="no migration"):
        StateStore(db, schema_version=2, migrations={})


def test_failed_migration_leaves_version_unchanged(db: Path) -> None:
    with StateStore(db, schema_version=1):
        pass

    def broken(conn: sqlite3.Connection) -> None:
        conn.execute("CREATE TABLE notes (id INTEGER PRIMARY KEY)")
        raise RuntimeError("boom")

    with pytest.raises(RuntimeError, match="boom"):
        StateStore(db, schema_version=2, migrations={1: broken})
    with StateStore(db, schema_version=1) as store:
        assert store.schema_version == 1
        with pytest.raises(sqlite3.OperationalError):
            store._conn.execute("SELECT * FROM notes")


def test_newer_schema_is_refused(db: Path) -> None:
    with StateStore(db, schema_version=3):
        pass
    with pytest.raises(StateSchemaError, match="newer"):
        StateStore(db, schema_version=1)


def test_a_transcript_since_an_event_views_only_what_follows(tmp_path: Path) -> None:
    from nisse_core.log.canonical import Fact

    with StateStore(tmp_path / "s.db") as store:
        store.open_work_item("w", title="t", scope="r", position="p", cause="t")
        whole = store.transcript("w", "orchestrator")
        first = whole.append(Fact(text="the failed attempt"))
        whole.append(Fact(text="the retry's brief"))
        fresh = whole.since(first)
        assert [e.event.text for e in fresh.view()] == ["the retry's brief"]
        assert len(fresh.read()) == 2  # the log keeps everything
        assert len(whole.view()) == 2