fastapi-gsap/tests/test_compliance_gate.py

# Copyright 2026 Guildhouse Dev
# SPDX-License-Identifier: Apache-2.0

"""Tests for compliance-gated AC issuance."""

import pytest
from unittest.mock import AsyncMock, patch, MagicMock
from httpx import AsyncClient, ASGITransport
from sqlmodel import SQLModel
from sqlalchemy.ext.asyncio import create_async_engine
from gsap_broker.app import app
from gsap_broker import db as db_module
from gsap_broker.drivers.base import AuthResult


@pytest.fixture(autouse=True)
async def test_db():
    engine = create_async_engine("sqlite+aiosqlite:///./test_compliance.db")
    db_module.engine = engine
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)
    yield
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.drop_all)


@pytest.fixture
async def client():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c


def _mock_auth_result(device_id=None):
    """Return a mock auth result with optional device_id."""
    return AuthResult(
        status=AuthResult.STATUS_AUTHORIZED,
        principal_did="did:web:test/p/alice",
        display_name="Alice",
        token_jti="jti-test",
        mfa_satisfied=True,
        device_id=device_id,
    )


def _authorize_body(accord_template="test-ops"):
    return {
        "playbook": "test",
        "corpus_entry_cid": "sha256:" + "a" * 64,
        "parameters_cid": "sha256:" + "b" * 64,
        "accord_template": accord_template,
        "driver_id": "keycloak",
    }


# ── TEST 13: Compliance disabled by default ───────────────────

@pytest.mark.asyncio
async def test_compliance_disabled_by_default(client, mocker):
    """With default settings (intune_enabled=False), no compliance check runs."""
    mocker.patch(
        "gsap_broker.drivers.keycloak.KeycloakDriver.authenticate",
        return_value=_mock_auth_result(),
    )
    resp = await client.post("/governance/authorize/", json=_authorize_body())
    assert resp.status_code == 200
    ac = resp.json()["authorization_context"]
    assert ac["device_id"] is None
    assert ac["device_compliant"] is None


# ── TEST 9: Compliant device → AC issued ──────────────────────

@pytest.mark.asyncio
async def test_compliance_required_compliant_device(client, mocker):
    """Compliance required + compliant device → AC issued with device metadata."""
    mocker.patch(
        "gsap_broker.drivers.keycloak.KeycloakDriver.authenticate",
        return_value=_mock_auth_result(device_id="dev-123"),
    )
    mocker.patch("gsap_broker.routers.authorize.settings.intune_enabled", True)
    mocker.patch("gsap_broker.routers.authorize.settings.intune_compliance_required", True)
    mocker.patch(
        "gsap_broker.routers.authorize._check_device_compliance",
        new_callable=AsyncMock,
        return_value=True,
    )

    resp = await client.post("/governance/authorize/", json=_authorize_body())
    assert resp.status_code == 200
    ac = resp.json()["authorization_context"]
    assert ac["device_id"] == "dev-123"
    assert ac["device_compliant"] is True
    assert ac["compliance_checked_at"] is not None


# ── TEST 10: Non-compliant device → 403 ──────────────────────

@pytest.mark.asyncio
async def test_compliance_required_noncompliant_device(client, mocker):
    """Compliance required + non-compliant device → 403."""
    mocker.patch(
        "gsap_broker.drivers.keycloak.KeycloakDriver.authenticate",
        return_value=_mock_auth_result(device_id="dev-bad"),
    )
    mocker.patch("gsap_broker.routers.authorize.settings.intune_enabled", True)
    mocker.patch("gsap_broker.routers.authorize.settings.intune_compliance_required", True)
    mocker.patch(
        "gsap_broker.routers.authorize._check_device_compliance",
        new_callable=AsyncMock,
        return_value=False,
    )

    resp = await client.post("/governance/authorize/", json=_authorize_body())
    assert resp.status_code == 403
    assert "not compliant" in resp.json()["detail"].lower()


# ── TEST 11: No device_id + strict mode → 403 ────────────────

@pytest.mark.asyncio
async def test_compliance_strict_no_device_id(client, mocker):
    """Strict mode + no device_id → 403."""
    mocker.patch(
        "gsap_broker.drivers.keycloak.KeycloakDriver.authenticate",
        return_value=_mock_auth_result(device_id=None),
    )
    mocker.patch("gsap_broker.routers.authorize.settings.intune_enabled", True)
    mocker.patch("gsap_broker.routers.authorize.settings.intune_compliance_required", True)
    mocker.patch("gsap_broker.routers.authorize.settings.intune_compliance_strict", True)

    resp = await client.post("/governance/authorize/", json=_authorize_body())
    assert resp.status_code == 403
    assert "device identity required" in resp.json()["detail"].lower()


# ── TEST 12: No device_id + permissive mode → AC issued ──────

@pytest.mark.asyncio
async def test_compliance_permissive_no_device_id(client, mocker):
    """Permissive mode + no device_id → AC issued without compliance fields."""
    mocker.patch(
        "gsap_broker.drivers.keycloak.KeycloakDriver.authenticate",
        return_value=_mock_auth_result(device_id=None),
    )
    mocker.patch("gsap_broker.routers.authorize.settings.intune_enabled", True)
    mocker.patch("gsap_broker.routers.authorize.settings.intune_compliance_required", True)
    mocker.patch("gsap_broker.routers.authorize.settings.intune_compliance_strict", False)

    resp = await client.post("/governance/authorize/", json=_authorize_body())
    assert resp.status_code == 200
    ac = resp.json()["authorization_context"]
    assert ac["device_compliant"] is None


# ── TEST: Per-accord compliance override ──────────────────────

@pytest.mark.asyncio
async def test_per_accord_compliance_override(client, mocker):
    """Accord template 'infrastructure-operations' requires compliance even when global default is False."""
    mocker.patch(
        "gsap_broker.drivers.keycloak.KeycloakDriver.authenticate",
        return_value=_mock_auth_result(device_id="dev-infra"),
    )
    mocker.patch("gsap_broker.routers.authorize.settings.intune_enabled", True)
    mocker.patch("gsap_broker.routers.authorize.settings.intune_compliance_required", False)
    mocker.patch(
        "gsap_broker.routers.authorize._check_device_compliance",
        new_callable=AsyncMock,
        return_value=True,
    )

    resp = await client.post(
        "/governance/authorize/",
        json=_authorize_body(accord_template="infrastructure-operations"),
    )
    assert resp.status_code == 200
    ac = resp.json()["authorization_context"]
    assert ac["device_compliant"] is True


# ── TEST 14: Chronicle event emitted ─────────────────────────

@pytest.mark.asyncio
async def test_chronicle_event_on_compliance_check(client, mocker):
    """Compliance check emits DEVICE_COMPLIANCE_CHECKED Chronicle event."""
    mocker.patch(
        "gsap_broker.drivers.keycloak.KeycloakDriver.authenticate",
        return_value=_mock_auth_result(device_id="dev-chron"),
    )
    mocker.patch("gsap_broker.routers.authorize.settings.intune_enabled", True)
    mocker.patch("gsap_broker.routers.authorize.settings.intune_compliance_required", True)
    mocker.patch(
        "gsap_broker.routers.authorize._check_device_compliance",
        new_callable=AsyncMock,
        return_value=True,
    )
    chronicle_mock = mocker.patch("gsap_broker.routers.authorize.chronicle.emit", new_callable=AsyncMock, return_value="")

    resp = await client.post("/governance/authorize/", json=_authorize_body())
    assert resp.status_code == 200

    # Find the DEVICE_COMPLIANCE_CHECKED call
    compliance_calls = [
        call for call in chronicle_mock.call_args_list
        if call.args[0] == "DEVICE_COMPLIANCE_CHECKED"
    ]
    assert len(compliance_calls) == 1
    event_data = compliance_calls[0].args[1]
    assert event_data["device_id"] == "dev-chron"
    assert event_data["decision"] == "allowed"