infrasynth-backend-kit/tests/test_gates.py

"""Tests for the composable per-endpoint gate layer."""

import pytest

from infrasynth.gates import (
    AltchaGate,
    EntitlementGate,
    FeatureGate,
    GatePermission,
    PermissionGate,
    TwoFactorGate,
    evaluate_gates,
    gated,
)
from infrasynth.shared.exceptions import AppError, AuthError, EntitlementError, NotFoundError

pytestmark = pytest.mark.django_db


class _Anon:
    is_authenticated = False


class FakeRequest:
    def __init__(self, user=None, *, auth=None, session=None, data=None, headers=None, query_params=None):
        self.user = user if user is not None else _Anon()
        self.auth = auth
        self.session = session
        self.data = data or {}
        self.headers = headers or {}
        self.query_params = query_params or {}


class FakeView:
    def __init__(self, *gates):
        self.infrasynth_gates = list(gates)


class TestGatePermission:
    def test_no_gates_allows(self, user):
        assert GatePermission().has_permission(FakeRequest(user), FakeView()) is True

    def test_decorator_merges_gates(self):
        class V:
            pass

        @gated(FeatureGate("a"))
        @gated(FeatureGate("b"))
        def action(self):
            return None

        assert len(action.infrasynth_gates) == 2


class TestTwoFactorGate:
    def test_no_config_passes_by_default(self, user):
        evaluate_gates(FakeRequest(user), FakeView(TwoFactorGate()))

    def test_no_config_required_denies(self, user):
        with pytest.raises(AuthError) as exc:
            evaluate_gates(FakeRequest(user), FakeView(TwoFactorGate(require_configured=True)))
        assert exc.value.code == "AUTH_2FA_SETUP_REQUIRED"

    def test_configured_without_proof_denies(self, user):
        from infrasynth.security.models import TwoFactorConfig

        TwoFactorConfig.objects.create(user=user, is_enabled=True, is_configured=True, secret_key_encrypted="x")
        with pytest.raises(AuthError) as exc:
            evaluate_gates(FakeRequest(user), FakeView(TwoFactorGate()))
        assert exc.value.code == "AUTH_2FA_REQUIRED"

    def test_configured_with_token_claim_passes(self, user):
        from infrasynth.security.models import TwoFactorConfig

        TwoFactorConfig.objects.create(user=user, is_enabled=True, is_configured=True, secret_key_encrypted="x")
        request = FakeRequest(user, auth={"2fa": True})
        evaluate_gates(request, FakeView(TwoFactorGate()))

    def test_configured_with_session_passes(self, user):
        from infrasynth.security.models import TwoFactorConfig

        TwoFactorConfig.objects.create(user=user, is_enabled=True, is_configured=True, secret_key_encrypted="x")
        request = FakeRequest(user, session={"_2fa_verified": True})
        evaluate_gates(request, FakeView(TwoFactorGate()))


class TestAltchaGate:
    def test_missing_token_denies(self):
        with pytest.raises(AppError) as exc:
            evaluate_gates(FakeRequest(), FakeView(AltchaGate()))
        assert exc.value.code == "VALIDATION_ALTCHA_REQUIRED"
        assert exc.value.status == 400

    def test_valid_solution_passes(self):
        from infrasynth.security.altcha.services import ALTCHAService

        svc = ALTCHAService()
        challenge = svc.create_challenge()
        solution, number = svc.compute_solution(challenge["salt"], challenge["difficulty"])
        request = FakeRequest(
            data={"altcha": {"challenge_id": challenge["challenge_id"], "solution": solution, "number": number}}
        )
        evaluate_gates(request, FakeView(AltchaGate()))

    def test_header_solution_passes(self):
        from infrasynth.security.altcha.services import ALTCHAService

        svc = ALTCHAService()
        challenge = svc.create_challenge()
        solution, number = svc.compute_solution(challenge["salt"], challenge["difficulty"])
        token = f"{challenge['challenge_id']}:{solution}:{number}"
        evaluate_gates(FakeRequest(headers={"X-Altcha": token}), FakeView(AltchaGate()))

    def test_bad_solution_denies(self):
        from infrasynth.security.altcha.services import ALTCHAService

        challenge = ALTCHAService().create_challenge()
        request = FakeRequest(
            data={"altcha": {"challenge_id": challenge["challenge_id"], "solution": "deadbeef", "number": 1}}
        )
        with pytest.raises(AppError) as exc:
            evaluate_gates(request, FakeView(AltchaGate()))
        assert exc.value.code == "VALIDATION_ALTCHA_INVALID"


class TestEntitlementGate:
    @pytest.fixture
    def entitled(self, tenant):
        from infrasynth.billing.models import App, Entitlement, Plan
        from infrasynth.shared.enums import EntitlementStatus, MonetizationModel

        app = App.objects.create(slug="messenger", name="Messenger", monetization=MonetizationModel.SUBSCRIPTION)
        plan = Plan.objects.create(app=app, slug="pro", name="Pro", price_amount=0, features={"payouts": True})
        Entitlement.objects.create(tenant=tenant, app=app, plan=plan, status=EntitlementStatus.ACTIVE)
        return app

    def test_entitled_passes(self, entitled):
        evaluate_gates(FakeRequest(), FakeView(EntitlementGate("messenger", feature="payouts")))

    def test_feature_not_in_plan_denies(self, entitled):
        with pytest.raises(EntitlementError) as exc:
            evaluate_gates(FakeRequest(), FakeView(EntitlementGate("messenger", feature="broadcast")))
        assert exc.value.code == "ENTITLEMENT_PLAN_UPGRADE_REQUIRED"
        assert exc.value.status == 402

    def test_no_entitlement_denies(self, tenant):
        with pytest.raises(EntitlementError) as exc:
            evaluate_gates(FakeRequest(), FakeView(EntitlementGate("messenger")))
        assert exc.value.code == "ENTITLEMENT_APP_NOT_OWNED"


class TestFeatureGate:
    def test_disabled_hides_as_404(self):
        from infrasynth.features.models import FeatureFlag

        FeatureFlag.objects.create(slug="ticketing", is_active=False)
        with pytest.raises(NotFoundError) as exc:
            evaluate_gates(FakeRequest(), FakeView(FeatureGate("ticketing")))
        assert exc.value.status == 404

    def test_enabled_passes(self):
        from infrasynth.features.models import FeatureFlag

        FeatureFlag.objects.create(slug="ticketing", is_active=True)
        evaluate_gates(FakeRequest(), FakeView(FeatureGate("ticketing")))


class TestPermissionGate:
    def test_missing_permission_denies(self, user):
        with pytest.raises(AuthError) as exc:
            evaluate_gates(FakeRequest(user), FakeView(PermissionGate("billing.payout")))
        assert exc.value.code == "AUTH_FORBIDDEN"

    def test_grant_passes(self, user):
        from infrasynth.security.models import Grant

        Grant.objects.create(user=user, codename="billing.payout")
        evaluate_gates(FakeRequest(user), FakeView(PermissionGate("billing.payout")))


class TestHybridPermissionIntegration:
    def test_kit_permission_evaluates_declared_gates(self, user):
        from infrasynth.features.models import FeatureFlag
        from infrasynth.security.permissions import HybridPermission

        FeatureFlag.objects.create(slug="ticketing", is_active=False)
        view = FakeView(FeatureGate("ticketing"))
        with pytest.raises(NotFoundError):
            HybridPermission().has_permission(FakeRequest(user), view)

    def test_kit_permission_allows_when_gate_passes(self, user):
        from infrasynth.features.models import FeatureFlag
        from infrasynth.security.permissions import HybridPermission

        FeatureFlag.objects.create(slug="ticketing", is_active=True)
        assert HybridPermission().has_permission(FakeRequest(user), FakeView(FeatureGate("ticketing"))) is True