nisse/nisse-dev/tests/test_state.py

import sqlite3
from pathlib import Path

import pytest

from nisse_dev.state 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_vm_block_persists_across_reopen(db: Path) -> None:
    with StateStore(db) as store:
        store.block_vm("root@vm1", cause="broken", reason="pip crashes", work_item="x")
        store.block_vm("root@vm1", cause="broken", reason="later", work_item="y")
    with StateStore(db) as store:
        [block] = store.vm_blocks().values()
        assert (block.host, block.reason, block.work_item) == (
            "root@vm1",
            "pip crashes",  # the first block is kept
            "x",
        )
        assert store.unblock_vm("root@vm1", cause="test") == block
        assert store.unblock_vm("root@vm1", cause="test") is None
        assert store.vm_blocks() == {}


def test_a_version_4_store_gains_vm_blocks(db: Path) -> None:
    with StateStore(db, schema_version=4) as store:
        store._conn.execute("DROP TABLE vm_blocks")  # as a real version 4 store
    with StateStore(db) as store:
        assert store.schema_version == 5
        store.block_vm("root@vm1", cause="broken", reason="r", work_item=None)
        assert list(store.vm_blocks()) == ["root@vm1"]