Kairo-Jinkendo/backend/tests/test_rights_registry.py

"""Rights registry and DB sync tests."""

from __future__ import annotations

import rights_registrations  # noqa: F401
from db import get_connection
from rights_registry import get_registered_capabilities, sync_rights_registry_to_db


def test_registry_contains_initial_capabilities():
    keys = {reg.key for reg in get_registered_capabilities()}
    assert keys == {
        "kairo.admin.access",
        "kairo.tenant.manage",
        "kairo.actor.manage",
        "kairo.actor.read",
        "kairo.context.read",
        "kairo.entitlements.read",
        "kairo.feature.registry.read",
        "kairo.feature.registry.manage",
        "kairo.prompt.registry.read",
        "kairo.prompt.registry.manage",
        "kairo.config.registry.read",
        "kairo.config.registry.manage",
        "kairo.prompt.render",
        "kairo.initiative.read",
        "kairo.initiative.manage",
        "kairo.action.read",
        "kairo.action.manage",
        "kairo.workspace.read",
        "kairo.attention.read",
        "kairo.blocker.read",
        "kairo.blocker.manage",
        "kairo.backlog.read",
        "kairo.backlog.manage",
        "kairo.milestone.read",
        "kairo.milestone.manage",
        "kairo.evidence.read",
        "kairo.evidence.manage",
        "kairo.decision.read",
        "kairo.decision.manage",
        "kairo.review.read",
        "kairo.review.manage",
        "kairo.recurring.read",
        "kairo.recurring.manage",
        "kairo.roadmap.reopen",
        "kairo.roadmap.reopen.emergency",
        "kairo.project.read",
        "kairo.project.manage",
    }


def test_sync_is_idempotent():
    assert sync_rights_registry_to_db() == 0
    assert sync_rights_registry_to_db() == 0

    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT COUNT(*) FROM capabilities")
            assert cur.fetchone()[0] == 37
            cur.execute("SELECT COUNT(*) FROM role_capability_grants")
            assert cur.fetchone()[0] >= 5
    finally:
        conn.close()