approval-engine/tests/test_cas.py

import tempfile
import threading
import sqlite3
from pathlib import Path

import pytest

from approval_engine.errors import Conflict, Unprocessable
from approval_engine.store import Engine
from tests.conftest import FROZEN, approve, binding, validity


def test_second_supersession_loses(engine):
    obj = approve(engine)
    first = engine.supersede(obj.id)
    assert engine.get(obj.id).status == "superseded"
    assert first["successor_id"]
    try:
        engine.supersede(obj.id)
        raise AssertionError("second supersession must conflict")
    except Conflict:
        pass
    claim = engine.claim(obj.id)
    assert claim["valid_now"] is False
    assert claim["reason_code"] == "superseded"


def test_concurrent_supersessions_one_winner():
    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "a.sqlite"
        setup = Engine(path, clock=lambda: FROZEN)
        obj = approve(setup)
        setup.close()

        winners: list[str] = []
        errors: list[str] = []
        barrier = threading.Barrier(2)

        def race():
            eng = Engine(path, clock=lambda: FROZEN)
            barrier.wait()
            try:
                result = eng.supersede(obj.id)
                winners.append(result["successor_id"])
            except Conflict as exc:
                errors.append(str(exc))
            finally:
                eng.close()

        threads = [threading.Thread(target=race) for _ in range(2)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        assert len(winners) == 1
        assert len(errors) == 1
        check = Engine(path, clock=lambda: FROZEN)
        assert check.get(obj.id).status == "superseded"
        check.close()


def test_consume_same_digest_is_idempotent(engine):
    obj = approve(engine)
    digest = "sha256:" + "ab" * 32
    first = engine.consume(obj.id, digest, decision_id="decision:first")
    second = engine.consume(obj.id, digest, decision_id="decision:retry")
    assert first["idempotent"] is False
    assert second["idempotent"] is True
    assert second["decision_id"] == "decision:first"
    assert [item["class"] for item in engine.undrained()].count("use") == 1
    claim = engine.claim(obj.id)
    assert claim["consumed"] is True
    assert claim["valid_now"] is False
    assert claim["reason_code"] == "consumed"


def test_consume_different_digest_conflicts(engine):
    obj = approve(engine)
    engine.consume(obj.id, "sha256:" + "ab" * 32)
    try:
        engine.consume(obj.id, "sha256:" + "cd" * 32)
        raise AssertionError("different request digest must conflict")
    except Conflict:
        pass


def test_concurrent_same_digest_consume_is_one_use_event():
    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "consume.sqlite"
        setup = Engine(path, clock=lambda: FROZEN)
        obj = approve(setup)
        setup.close()
        digest = "sha256:" + "ef" * 32
        results: list[bool] = []
        barrier = threading.Barrier(2)

        def race():
            eng = Engine(path, clock=lambda: FROZEN)
            barrier.wait()
            try:
                results.append(eng.consume(obj.id, digest)["idempotent"])
            finally:
                eng.close()

        threads = [threading.Thread(target=race) for _ in range(2)]
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join()
        assert sorted(results) == [False, True]
        check = Engine(path, clock=lambda: FROZEN)
        assert [item["class"] for item in check.undrained()].count("use") == 1
        check.close()


def test_existing_database_migrates_consumption_columns():
    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "legacy.sqlite"
        conn = sqlite3.connect(path)
        conn.execute(
            """CREATE TABLE approvals (
                id TEXT PRIMARY KEY, status TEXT NOT NULL,
                binding_json TEXT NOT NULL, binding_digest TEXT NOT NULL,
                pdp_digest TEXT, actor TEXT NOT NULL, principal TEXT NOT NULL,
                action TEXT NOT NULL, purpose TEXT NOT NULL,
                target_json TEXT NOT NULL, not_before TEXT NOT NULL,
                expires_at TEXT NOT NULL, required_count INTEGER NOT NULL,
                superseded_by TEXT, created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL
            )"""
        )
        conn.commit()
        conn.close()
        eng = Engine(path, clock=lambda: FROZEN)
        columns = {
            row["name"]
            for row in eng._conn().execute("PRAGMA table_info(approvals)").fetchall()
        }
        assert {"consumed_digest", "consumed_decision_id", "consumed_at"} <= columns
        eng.close()


def test_pdp_path_requires_a_digest_at_issue(engine):
    """GH-DEC-2026-008: refuse at issue, not at the protected side effect."""
    with pytest.raises(Unprocessable):
        engine.create(binding(), validity(), pdp_path=True)


def test_pdp_path_approval_states_itself_on_the_claim(engine):
    obj = engine.create(
        binding(), validity(), pdp_digest="sha256:" + "ab" * 32, pdp_path=True
    )
    engine.add_entry(obj.id, "user:alice")
    claim = engine.claim(obj.id)
    assert claim["binding"]["pdp_path"] is True
    assert claim["binding"]["pdp_digest"] == "sha256:" + "ab" * 32


def test_a_recorded_digest_alone_does_not_declare_the_path(engine):
    """Intent is declared, never inferred from an incidental digest."""
    obj = engine.create(binding(), validity(), pdp_digest="sha256:" + "cd" * 32)
    claim = engine.claim(obj.id)
    assert claim["binding"]["pdp_digest"] is not None
    assert claim["binding"]["pdp_path"] is False


def test_successor_inherits_the_pdp_path_declaration(engine):
    obj = engine.create(
        binding(), validity(), pdp_digest="sha256:" + "ef" * 32, pdp_path=True
    )
    engine.add_entry(obj.id, "user:alice")
    result = engine.supersede(obj.id, None)
    successor = engine.get(result["successor_id"])
    assert successor.pdp_path is True


def test_v2_database_migrates_to_v3_preserving_approvals():
    """Schema v3 (GH-DEC-2026-008) must not disturb approvals issued under v2."""
    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / "v2.sqlite"
        eng = Engine(path, clock=lambda: FROZEN)
        obj = eng.create(binding(), validity(), pdp_digest="sha256:" + "12" * 32)
        eng.add_entry(obj.id, "user:alice")
        # simulate a store written before v3 existed
        eng._conn().execute("ALTER TABLE approvals DROP COLUMN pdp_path")
        eng._conn().execute("PRAGMA user_version=2")
        eng._conn().commit()
        eng.close()

        upgraded = Engine(path, clock=lambda: FROZEN)
        version = int(upgraded._conn().execute("PRAGMA user_version").fetchone()[0])
        assert version == 3
        survivor = upgraded.get(obj.id)
        assert survivor.status == "approved"
        assert survivor.pdp_digest == "sha256:" + "12" * 32
        # a legacy row never declared the path; intent is not back-filled from
        # a digest that happens to be present
        assert survivor.pdp_path is False
        assert upgraded.claim(obj.id)["binding"]["pdp_path"] is False
        upgraded.close()