"""Executable acceptance tests for ARCH-01 Slice A (#822).

Each acceptance criterion (#822 §12) and named test (#822 §13) is exercised
against a real SQLite database. The migration runs on a fresh DB in ``setUp``;
the test-run output is the durable evidence the issue requires (§14).

Enforcement being proven:

* ``[TRUSTED-SERVICE]`` — the ``cp_*`` actor functions exist only on the
  trusted kernel connection; a raw connection cannot satisfy the triggers.
* ``[SCHEMA]`` — fail-closed aborts, exact dominance set, NOT-NULL class,
  immutability, and the last-active-grant floor are enforced by
  CHECK/FK/trigger, verified here including raw-write bypass and concurrency.
"""

from __future__ import annotations

import os
import sqlite3
import tempfile
import threading
import unittest
from concurrent.futures import ThreadPoolExecutor

import arch01_platform as ap
from arch01_platform import (
    ALREADY_INSTALLED,
    AUTHORIZATION_DENIED,
    CONCURRENT_INSTALLATION_LOST,
    DISTINGUISHED_ISSUER_ID,
    DOMINANCE_SET_MISMATCH,
    DOMINANCE_TUPLES,
    INSTALLED,
    INVALID_ACTOR_CONTEXT,
    INVALID_BOOTSTRAP_STATE,
    PlatformKernel,
)

INSTALLER = "platform.installer"

_BOOTSTRAP_TABLES = (
    "principal_equivalence_classes",
    "principals",
    "authoritative_issuers",
    "authority_dominance",
    "platform_bootstrap_seed",
    "platform_bootstrap_grants",
    "platform_active_invariant",
    "install_state",
)


def _count(kernel: PlatformKernel, table: str) -> int:
    return kernel._conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]


def _count_where(kernel: PlatformKernel, table: str, where: str) -> int:
    return kernel._conn.execute(f"SELECT COUNT(*) FROM {table} WHERE {where}").fetchone()[0]


def _all_bootstrap_empty(kernel: PlatformKernel) -> bool:
    return all(_count(kernel, t) == 0 for t in _BOOTSTRAP_TABLES)


class Arch01MemoryTest(unittest.TestCase):
    """Single-connection behavior on an in-memory database."""

    def setUp(self) -> None:
        self.kernel = PlatformKernel(":memory:")

    def tearDown(self) -> None:
        self.kernel.close()

    # -- AC1 -------------------------------------------------------------- #
    def test_install_clean(self) -> None:  # t_install_clean(+)
        res = self.kernel.install_platform(INSTALLER)
        self.assertEqual(res.code, INSTALLED)
        self.assertTrue(self.kernel.is_installed())
        self.assertEqual(_count(self.kernel, "install_state"), 1)
        self.assertEqual(self.kernel.active_grant_count(), 1)
        self.assertIn(ap.EVT_PLATFORM_INSTALLED, self.kernel.audit_events())
        rows = set(
            self.kernel._conn.execute(
                "SELECT dominant, subordinate FROM authority_dominance"
            ).fetchall()
        )
        self.assertEqual(rows, set(DOMINANCE_TUPLES))
        issuer_ref = self.kernel._conn.execute(
            "SELECT i.issuer_ref FROM principals p JOIN authoritative_issuers i "
            "ON p.issuer_id = i.issuer_id WHERE p.principal_id = ?",
            (INSTALLER,),
        ).fetchone()
        self.assertEqual(issuer_ref[0], DISTINGUISHED_ISSUER_ID)

    # -- AC2 -------------------------------------------------------------- #
    def test_install_twice(self) -> None:  # t_install_twice(-)
        self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
        res2 = self.kernel.install_platform(INSTALLER)
        self.assertEqual(res2.code, ALREADY_INSTALLED)
        self.assertEqual(_count(self.kernel, "principals"), 1)
        self.assertEqual(_count(self.kernel, "platform_bootstrap_grants"), 1)
        self.assertEqual(_count(self.kernel, "install_state"), 1)

    # -- AC3 / AC5 -------------------------------------------------------- #
    def test_install_stage_rollback(self) -> None:  # t_install_stage_rollback
        for stop in range(1, 9):
            with self.subTest(stages=stop):
                k = PlatformKernel(":memory:")
                try:
                    self._partial_bootstrap_then_rollback(k, stop)
                    self.assertTrue(
                        _all_bootstrap_empty(k),
                        f"partial rows survived rollback at stage {stop}",
                    )
                    self.assertFalse(k.is_installed())
                finally:
                    k.close()

    def test_no_partial_after_rollback(self) -> None:  # t_no_partial_after_rollback
        k = PlatformKernel(":memory:")
        try:
            code = self._seed_bootstrap_and_mark(k, dominance=DOMINANCE_TUPLES[:-1])
            self.assertEqual(code, DOMINANCE_SET_MISMATCH)
            self.assertTrue(_all_bootstrap_empty(k))
            self.assertFalse(k.is_installed())
        finally:
            k.close()

    # -- AC4 -------------------------------------------------------------- #
    def test_dominance_missing(self) -> None:  # t_dominance_missing(-)
        k = PlatformKernel(":memory:")
        try:
            self.assertEqual(
                self._seed_bootstrap_and_mark(k, dominance=DOMINANCE_TUPLES[:-1]),
                DOMINANCE_SET_MISMATCH,
            )
            self.assertFalse(k.is_installed())
        finally:
            k.close()

    def test_dominance_extra(self) -> None:  # t_dominance_extra(-)
        k = PlatformKernel(":memory:")
        try:
            extra = DOMINANCE_TUPLES + (("platform.bootstrap", "rogue.extra"),)
            self.assertEqual(
                self._seed_bootstrap_and_mark(k, dominance=extra),
                DOMINANCE_SET_MISMATCH,
            )
            self.assertFalse(k.is_installed())
        finally:
            k.close()

    def test_dominance_malformed(self) -> None:  # t_dominance_malformed(-)
        k = PlatformKernel(":memory:")
        try:
            malformed = DOMINANCE_TUPLES[:-1] + (("supervisor.root", "WRONG.subordinate"),)
            self.assertEqual(
                self._seed_bootstrap_and_mark(k, dominance=malformed),
                DOMINANCE_SET_MISMATCH,
            )
            self.assertFalse(k.is_installed())
        finally:
            k.close()

    # -- AC6 -------------------------------------------------------------- #
    def test_principal_no_class(self) -> None:  # t_principal_no_class(-)
        with self.kernel.actor_context("op", "operator", "install"):
            with self.assertRaises(sqlite3.IntegrityError):
                self.kernel._conn.execute(
                    "INSERT INTO principals"
                    "(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
                    "VALUES ('x', 'operator', NULL, NULL, NULL, '2026-01-01T00:00:00Z')"
                )

    # -- AC7 -------------------------------------------------------------- #
    def test_noninstaller_null_issuer(self) -> None:  # t_nonobstaller_null_issuer(-)
        self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
        with self.kernel.actor_context("op", "operator", "normal"):
            cur = self.kernel._conn.execute(
                "INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')"
            )
            class_id = cur.lastrowid
            with self.assertRaises(sqlite3.IntegrityError) as ctx:
                self.kernel._conn.execute(
                    "INSERT INTO principals"
                    "(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
                    "VALUES ('rogue', 'operator', ?, NULL, NULL, '2026-01-01T00:00:00Z')",
                    (class_id,),
                )
            self.assertIn("INVALID_BOOTSTRAP_STATE", str(ctx.exception))

    def test_installer_null_issuer_only_during_install(self) -> None:
        self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
        with self.kernel.actor_context("i2", "installer", "install"):
            cur = self.kernel._conn.execute(
                "INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')"
            )
            class_id = cur.lastrowid
            with self.assertRaises(sqlite3.IntegrityError):
                self.kernel._conn.execute(
                    "INSERT INTO principals"
                    "(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
                    "VALUES ('i2', 'installer', ?, NULL, NULL, '2026-01-01T00:00:00Z')",
                    (class_id,),
                )

    # -- AC8 -------------------------------------------------------------- #
    def test_context_missing(self) -> None:  # t_context_missing(-)
        self.assertIsNone(self.kernel._ctx)
        with self.assertRaises(sqlite3.IntegrityError) as ctx:
            self.kernel._conn.execute(
                "INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')"
            )
        self.assertIn("INVALID_ACTOR_CONTEXT", str(ctx.exception))

    def test_context_stale(self) -> None:  # t_context_stale(-)
        with self.kernel.actor_context("op", "operator", "normal"):
            self.kernel._ctx.expired = True
            with self.assertRaises(sqlite3.IntegrityError) as ctx:
                self.kernel._conn.execute(
                    "INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')"
                )
            self.assertIn("INVALID_ACTOR_CONTEXT", str(ctx.exception))

    def test_context_epoch_shift(self) -> None:  # t_context_epoch_shift(-)
        with self.kernel.actor_context("op", "operator", "normal"):
            self.kernel._ctx.live_epoch = self.kernel._ctx.bound_epoch + 99
            with self.assertRaises(sqlite3.IntegrityError) as ctx:
                self.kernel._conn.execute(
                    "INSERT INTO principal_equivalence_classes(created_at) VALUES ('2026-01-01T00:00:00Z')"
                )
            self.assertIn("INVALID_ACTOR_CONTEXT", str(ctx.exception))

    def test_bad_actor_kind_or_mode_rejected(self) -> None:
        for kind, mode in (("intruder", "normal"), ("operator", "sabotage")):
            with self.subTest(kind=kind, mode=mode):
                with self.kernel.actor_context("op", kind, mode):
                    with self.assertRaises(sqlite3.IntegrityError):
                        self.kernel._conn.execute(
                            "INSERT INTO principal_equivalence_classes(created_at) "
                            "VALUES ('2026-01-01T00:00:00Z')"
                        )

    # -- AC9 -------------------------------------------------------------- #
    def test_bootstrap_immutable_update(self) -> None:  # t_bootstrap_immutable_{update}
        self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
        cases = [
            ("UPDATE install_state SET installed_at = 'x' WHERE id = 1", "IMMUTABLE_INSTALL_STATE"),
            ("UPDATE platform_bootstrap_seed SET created_at = 'x' WHERE seed_id = 1", "IMMUTABLE_SEED"),
            ("UPDATE authority_dominance SET subordinate = 'x' WHERE dominant = 'supervisor.root'", "IMMUTABLE_DOMINANCE"),
            (f"UPDATE authoritative_issuers SET issuer_ref = 'x' WHERE issuer_ref = '{DISTINGUISHED_ISSUER_ID}'", "IMMUTABLE_ISSUER"),
            (f"UPDATE principals SET actor_kind = 'operator' WHERE principal_id = '{INSTALLER}'", "IMMUTABLE_PRINCIPAL"),
        ]
        for sql, tag in cases:
            with self.subTest(sql=sql):
                with self.kernel.actor_context("op", "operator", "normal"):
                    with self.assertRaises(sqlite3.IntegrityError) as ctx:
                        self.kernel._conn.execute(sql)
                    self.assertIn(tag, str(ctx.exception))

    def test_bootstrap_immutable_delete(self) -> None:  # t_bootstrap_immutable_{delete}
        self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
        cases = [
            ("DELETE FROM install_state WHERE id = 1", "IMMUTABLE_INSTALL_STATE"),
            ("DELETE FROM platform_bootstrap_seed WHERE seed_id = 1", "IMMUTABLE_SEED"),
            ("DELETE FROM authority_dominance", "IMMUTABLE_DOMINANCE"),
            ("DELETE FROM authoritative_issuers", "IMMUTABLE_ISSUER"),
            (f"DELETE FROM principals WHERE principal_id = '{INSTALLER}'", "IMMUTABLE_PRINCIPAL"),
            ("DELETE FROM platform_bootstrap_grants", "IMMUTABLE_GRANT"),
        ]
        for sql, tag in cases:
            with self.subTest(sql=sql):
                with self.kernel.actor_context("op", "operator", "normal"):
                    with self.assertRaises(sqlite3.IntegrityError) as ctx:
                        self.kernel._conn.execute(sql)
                    self.assertIn(tag, str(ctx.exception))

    def test_grant_reactivation_rejected(self) -> None:
        self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
        self.kernel.register_principal(
            "op1", "operator", DISTINGUISHED_ISSUER_ID, actor_principal=INSTALLER
        )
        self.assertEqual(
            self.kernel.grant_platform_bootstrap("op1", INSTALLER).code, INSTALLED
        )
        gid = self.kernel._conn.execute(
            "SELECT grant_id FROM platform_bootstrap_grants WHERE grantee_principal_id = 'op1'"
        ).fetchone()[0]
        self.assertEqual(
            self.kernel.revoke_platform_bootstrap(gid, actor_principal=INSTALLER).code,
            INSTALLED,
        )
        with self.kernel.actor_context("op", "operator", "normal"):
            with self.assertRaises(sqlite3.IntegrityError) as ctx:
                self.kernel._conn.execute(
                    "UPDATE platform_bootstrap_grants SET active = 1 WHERE grant_id = ?",
                    (gid,),
                )
            self.assertIn("IMMUTABLE_GRANT", str(ctx.exception))

    # -- AC12 ------------------------------------------------------------- #
    def test_raw_write_bypass(self) -> None:  # t_raw_write_bypass(raw-bypass)
        with tempfile.TemporaryDirectory() as tmp:
            path = os.path.join(tmp, "p.sqlite3")
            k = PlatformKernel(path)
            self.assertEqual(k.install_platform(INSTALLER).code, INSTALLED)
            k.close()
            raw = sqlite3.connect(path)
            raw.execute("PRAGMA foreign_keys = ON")
            try:
                with self.assertRaises(sqlite3.Error):
                    raw.execute(
                        "INSERT INTO audit_records(event, created_at) "
                        "VALUES ('forged', '2026-01-01T00:00:00Z')"
                    )
                    raw.commit()
                with self.assertRaises(sqlite3.Error):
                    raw.execute("UPDATE install_state SET installed_at = 'x' WHERE id = 1")
                    raw.commit()
                with self.assertRaises(sqlite3.Error):
                    raw.execute("DELETE FROM platform_bootstrap_grants")
                    raw.commit()
            finally:
                raw.close()

    # -- AC13 ------------------------------------------------------------- #
    def test_audit_created(self) -> None:  # t_audit_created(+)
        self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
        self.kernel.register_principal(
            "op1", "operator", DISTINGUISHED_ISSUER_ID, actor_principal=INSTALLER
        )
        self.assertEqual(
            self.kernel.grant_platform_bootstrap("op1", INSTALLER).code, INSTALLED
        )
        gid = self.kernel._conn.execute(
            "SELECT grant_id FROM platform_bootstrap_grants WHERE grantee_principal_id = 'op1'"
        ).fetchone()[0]
        self.assertEqual(
            self.kernel.revoke_platform_bootstrap(gid, actor_principal=INSTALLER).code,
            INSTALLED,
        )
        events = self.kernel.audit_events()
        for evt in (
            ap.EVT_PLATFORM_INSTALLED,
            ap.EVT_GRANT_CREATED,
            ap.EVT_GRANT_REVOKED,
            ap.EVT_PRINCIPAL_REGISTERED,
        ):
            self.assertIn(evt, events)

    # -- AC14 ------------------------------------------------------------- #
    def test_audit_immutable(self) -> None:  # t_audit_immutable(raw-bypass)
        self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
        with self.kernel.actor_context("op", "operator", "normal"):
            with self.assertRaises(sqlite3.IntegrityError) as up:
                self.kernel._conn.execute("UPDATE audit_records SET event = 'x' WHERE audit_id = 1")
            self.assertIn("IMMUTABLE_AUDIT", str(up.exception))
            with self.assertRaises(sqlite3.IntegrityError) as dl:
                self.kernel._conn.execute("DELETE FROM audit_records WHERE audit_id = 1")
            self.assertIn("IMMUTABLE_AUDIT", str(dl.exception))

    # -- meta ------------------------------------------------------------- #
    def test_schema_meta(self) -> None:
        rows = dict(self.kernel._conn.execute("SELECT key, value FROM arch01_meta").fetchall())
        self.assertEqual(rows["schema_version"], str(ap.SCHEMA_VERSION))
        self.assertIn("disabled by default", rows["architecture"])

    def test_register_principal_creates_class_first(self) -> None:
        self.assertEqual(self.kernel.install_platform(INSTALLER).code, INSTALLED)
        res = self.kernel.register_principal(
            "svc1", "service", DISTINGUISHED_ISSUER_ID, actor_principal=INSTALLER
        )
        self.assertEqual(res.code, INSTALLED)
        row = self.kernel._conn.execute(
            "SELECT current_class_id FROM principals WHERE principal_id = 'svc1'"
        ).fetchone()
        self.assertIsNotNone(row[0])

    # -- helpers ---------------------------------------------------------- #
    def _partial_bootstrap_then_rollback(self, k: PlatformKernel, stop: int) -> None:
        """Execute the first ``stop`` bootstrap statements, then ROLLBACK."""
        now = "2026-01-01T00:00:00Z"
        k._conn.execute("BEGIN IMMEDIATE")
        class_id = None
        issuer_id = None
        try:
            with k.actor_context(INSTALLER, "installer", "install"):
                c = k._conn
                if stop >= 1:
                    class_id = c.execute(
                        "INSERT INTO principal_equivalence_classes(created_at) VALUES (?)", (now,)
                    ).lastrowid
                if stop >= 2:
                    c.execute(
                        "INSERT INTO principals(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
                        "VALUES (?, 'installer', ?, NULL, ?, ?)",
                        (INSTALLER, class_id, INSTALLER, now),
                    )
                if stop >= 3:
                    issuer_id = c.execute(
                        "INSERT INTO authoritative_issuers(issuer_kind, issuer_ref, created_at) VALUES ('operator-key', ?, ?)",
                        (DISTINGUISHED_ISSUER_ID, now),
                    ).lastrowid
                if stop >= 4:
                    c.execute(
                        "UPDATE principals SET issuer_id = ? WHERE principal_id = ?",
                        (issuer_id, INSTALLER),
                    )
                if stop >= 5:
                    c.executemany(
                        "INSERT INTO authority_dominance(dominant, subordinate) VALUES (?, ?)",
                        DOMINANCE_TUPLES,
                    )
                if stop >= 6:
                    c.execute(
                        "INSERT INTO platform_bootstrap_seed(seed_id, installer_principal_id, created_at) VALUES (1, ?, ?)",
                        (INSTALLER, now),
                    )
                if stop >= 7:
                    c.execute(
                        "INSERT INTO platform_bootstrap_grants(grantee_principal_id, granted_by, active, created_at) VALUES (?, NULL, 1, ?)",
                        (INSTALLER, now),
                    )
                if stop >= 8:
                    c.execute("INSERT INTO platform_active_invariant(id, active_count) VALUES (1, 1)")
        finally:
            k._conn.execute("ROLLBACK")

    def _seed_bootstrap_and_mark(self, k: PlatformKernel, dominance) -> str:
        """Seed a full bootstrap with a caller-supplied dominance set, then
        attempt the marker insert. Returns the classified failure code (or
        INSTALLED). Rolls back on failure so no partial rows remain."""
        now = "2026-01-01T00:00:00Z"
        k._conn.execute("BEGIN IMMEDIATE")
        try:
            with k.actor_context(INSTALLER, "installer", "install"):
                c = k._conn
                class_id = c.execute(
                    "INSERT INTO principal_equivalence_classes(created_at) VALUES (?)", (now,)
                ).lastrowid
                c.execute(
                    "INSERT INTO principals(principal_id, actor_kind, current_class_id, issuer_id, registered_by, created_at) "
                    "VALUES (?, 'installer', ?, NULL, ?, ?)",
                    (INSTALLER, class_id, INSTALLER, now),
                )
                issuer_id = c.execute(
                    "INSERT INTO authoritative_issuers(issuer_kind, issuer_ref, created_at) VALUES ('operator-key', ?, ?)",
                    (DISTINGUISHED_ISSUER_ID, now),
                ).lastrowid
                c.execute(
                    "UPDATE principals SET issuer_id = ? WHERE principal_id = ?",
                    (issuer_id, INSTALLER),
                )
                c.executemany(
                    "INSERT INTO authority_dominance(dominant, subordinate) VALUES (?, ?)",
                    dominance,
                )
                c.execute(
                    "INSERT INTO platform_bootstrap_seed(seed_id, installer_principal_id, created_at) VALUES (1, ?, ?)",
                    (INSTALLER, now),
                )
                c.execute(
                    "INSERT INTO platform_bootstrap_grants(grantee_principal_id, granted_by, active, created_at) VALUES (?, NULL, 1, ?)",
                    (INSTALLER, now),
                )
                c.execute("INSERT INTO platform_active_invariant(id, active_count) VALUES (1, 1)")
                c.execute(
                    "INSERT INTO install_state(id, marker, installed_at) VALUES (1, 'installed', ?)",
                    (now,),
                )
            k._conn.execute("COMMIT")
            return INSTALLED
        except sqlite3.Error as exc:
            k._safe_rollback()
            return PlatformKernel._classify(exc)


class Arch01ConcurrencyTest(unittest.TestCase):
    """Concurrency invariants require file-backed DBs and independent connections."""

    def setUp(self) -> None:
        self._tmp = tempfile.TemporaryDirectory()
        self.path = os.path.join(self._tmp.name, "p.sqlite3")

    def tearDown(self) -> None:
        self._tmp.cleanup()

    # -- AC10 ------------------------------------------------------------- #
    def test_concurrent_install(self) -> None:  # t_concurrent_install(concurrency)
        k1 = PlatformKernel(self.path, busy_timeout_ms=0)
        k2 = PlatformKernel(self.path, busy_timeout_ms=0)
        barrier = threading.Barrier(2)
        results = {}

        def _install(name, kernel):
            barrier.wait()
            results[name] = kernel.install_platform(INSTALLER).code

        try:
            with ThreadPoolExecutor(max_workers=2) as ex:
                f1 = ex.submit(_install, "a", k1)
                f2 = ex.submit(_install, "b", k2)
                f1.result()
                f2.result()
            codes = sorted(results.values())
            self.assertEqual(codes.count(INSTALLED), 1, f"exactly one install expected: {results}")
            other = [c for c in results.values() if c != INSTALLED][0]
            self.assertIn(other, (ALREADY_INSTALLED, CONCURRENT_INSTALLATION_LOST))
            self.assertTrue(k1.is_installed())
            self.assertEqual(_count(k1, "install_state"), 1)
            self.assertEqual(_count(k1, "principals"), 1)
        finally:
            k1.close()
            k2.close()

    # -- AC11 ------------------------------------------------------------- #
    def test_concurrent_last_grant_revoke(self) -> None:  # t_concurrent_last_grant_revoke
        setup = PlatformKernel(self.path)
        self.assertEqual(setup.install_platform(INSTALLER).code, INSTALLED)
        setup.register_principal("op1", "operator", DISTINGUISHED_ISSUER_ID, actor_principal=INSTALLER)
        self.assertEqual(setup.grant_platform_bootstrap("op1", INSTALLER).code, INSTALLED)
        self.assertEqual(setup.active_grant_count(), 2)
        gids = [
            r[0]
            for r in setup._conn.execute(
                "SELECT grant_id FROM platform_bootstrap_grants WHERE active = 1 ORDER BY grant_id"
            ).fetchall()
        ]
        setup.close()
        self.assertEqual(len(gids), 2)

        k1 = PlatformKernel(self.path, busy_timeout_ms=3000)
        k2 = PlatformKernel(self.path, busy_timeout_ms=3000)
        barrier = threading.Barrier(2)
        results = {}

        def _revoke(name, kernel, gid):
            barrier.wait()
            results[name] = kernel.revoke_platform_bootstrap(gid, actor_principal=INSTALLER).code

        try:
            with ThreadPoolExecutor(max_workers=2) as ex:
                f1 = ex.submit(_revoke, "a", k1, gids[0])
                f2 = ex.submit(_revoke, "b", k2, gids[1])
                f1.result()
                f2.result()
            codes = list(results.values())
            self.assertEqual(codes.count(INSTALLED), 1, f"exactly one revoke should win: {results}")
            self.assertEqual(codes.count(AUTHORIZATION_DENIED), 1, f"one revoke must be denied: {results}")
            self.assertEqual(k1.active_grant_count(), 1)
            self.assertEqual(_count_where(k1, "platform_bootstrap_grants", "active = 1"), 1)
        finally:
            k1.close()
            k2.close()

    def test_revoke_final_grant_denied(self) -> None:
        k = PlatformKernel(self.path)
        try:
            self.assertEqual(k.install_platform(INSTALLER).code, INSTALLED)
            gid = k._conn.execute(
                "SELECT grant_id FROM platform_bootstrap_grants WHERE active = 1"
            ).fetchone()[0]
            res = k.revoke_platform_bootstrap(gid, actor_principal=INSTALLER)
            self.assertEqual(res.code, AUTHORIZATION_DENIED)
            self.assertEqual(k.active_grant_count(), 1)
            self.assertEqual(_count_where(k, "platform_bootstrap_grants", "active = 1"), 1)
        finally:
            k.close()


if __name__ == "__main__":
    unittest.main()