govoplan-policy/tests/test_impact_preview.py

from __future__ import annotations

import unittest
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace

from fastapi import HTTPException
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker

from govoplan_core.core.policy import (
    CAPABILITY_POLICY_IMPACT_SUBJECT_PREFIX,
    PolicyImpactPopulationRequest,
    PolicyImpactSubject,
    PolicyImpactSubjectBatch,
)
from govoplan_policy.backend.db.models import PolicyOverride
from govoplan_policy.backend.impact_preview import (
    PolicyImpactPopulationSpec,
    preview_policy_impact,
)
from govoplan_policy.backend.api.v1.routes import (
    _require_recent_policy_authentication,
)


class _SubjectProvider:
    provider_id = "example"
    supported_policy_families = ("view",)

    def collect_policy_impact_subjects(
        self,
        session: object | None = None,
        *,
        request: PolicyImpactPopulationRequest,
    ) -> PolicyImpactSubjectBatch:
        del session
        subjects = (
            PolicyImpactSubject(
                module_id="views",
                resource_type="view",
                resource_id="view-1",
                action="edit",
                label="First" if request.allow_sensitive_details else None,
            ),
            PolicyImpactSubject(
                module_id="views",
                resource_type="view",
                resource_id="view-2",
                action="view",
                label="Second" if request.allow_sensitive_details else None,
            ),
        )
        return PolicyImpactSubjectBatch(
            provider_id=self.provider_id,
            subjects=subjects[: request.limit],
            state="truncated" if request.limit < len(subjects) else "complete",
            total_available=len(subjects),
            explanation="Explicit test population.",
        )


class _Registry:
    def __init__(self, provider: object | None = None) -> None:
        self.provider = provider

    def has_capability(self, name: str) -> bool:
        return (
            self.provider is not None
            and name == f"{CAPABILITY_POLICY_IMPACT_SUBJECT_PREFIX}example"
        )

    def capability(self, name: str) -> object | None:
        del name
        return self.provider


class PolicyImpactPreviewTests(unittest.TestCase):
    def setUp(self) -> None:
        self.engine = create_engine("sqlite+pysqlite:///:memory:")
        PolicyOverride.__table__.create(self.engine)
        self.session: Session = sessionmaker(
            bind=self.engine,
            expire_on_commit=False,
        )()

    def tearDown(self) -> None:
        self.session.close()
        self.engine.dispose()

    def test_preview_compares_effective_view_policy_without_persistence(self) -> None:
        preview = preview_policy_impact(
            self.session,
            registry=_Registry(_SubjectProvider()),
            tenant_id="tenant-1",
            policy_family="view",
            scope_type="tenant",
            scope_id=None,
            proposed_policy={
                "allow_edit": False,
                "allowed_view_ids": ["view-1"],
            },
            populations=(PolicyImpactPopulationSpec(provider_id="example"),),
            actor_scopes=("admin:policies:read", "policy:impact:details"),
            include_details=True,
            details_allowed=True,
        )

        self.assertEqual(2, preview.counts["newly_denied"])
        self.assertEqual(2, len(preview.effects))
        self.assertEqual(
            {"view.allow_edit", "view.allow_view"},
            {effect.rule for effect in preview.effects},
        )
        self.assertTrue(
            all(
                effect.proposed_sources[-1].label.startswith("Proposed Tenant")
                for effect in preview.effects
            )
        )
        self.assertEqual(0, self.session.query(PolicyOverride).count())

    def test_details_are_hidden_but_permission_filtered_counts_remain(self) -> None:
        preview = preview_policy_impact(
            self.session,
            registry=_Registry(_SubjectProvider()),
            tenant_id="tenant-1",
            policy_family="view",
            scope_type="system",
            scope_id=None,
            proposed_policy={"allow_view": False},
            populations=(
                PolicyImpactPopulationSpec(provider_id="example", limit=1),
            ),
            include_details=True,
            details_allowed=False,
        )

        self.assertEqual(1, preview.counts["newly_denied"])
        self.assertEqual((), preview.effects)
        self.assertTrue(preview.details_hidden)
        self.assertIn("policy:impact:details", preview.details_explanation or "")
        self.assertEqual("truncated", preview.populations[0]["state"])
        self.assertTrue(preview.high_impact)

    def test_unavailable_provider_is_explained_instead_of_counted_as_zero(self) -> None:
        preview = preview_policy_impact(
            self.session,
            registry=_Registry(),
            tenant_id="tenant-1",
            policy_family="view",
            scope_type="tenant",
            scope_id=None,
            proposed_policy={},
            populations=(PolicyImpactPopulationSpec(provider_id="missing"),),
        )

        self.assertEqual("unavailable", preview.populations[0]["state"])
        self.assertIn("not enabled", preview.populations[0]["explanation"])

    def test_system_policy_guard_requires_a_recent_interactive_session(self) -> None:
        fresh = SimpleNamespace(
            auth_session=SimpleNamespace(
                created_at=datetime.now(timezone.utc) - timedelta(minutes=2)
            )
        )
        _require_recent_policy_authentication(fresh)  # type: ignore[arg-type]

        stale = SimpleNamespace(
            auth_session=SimpleNamespace(
                created_at=datetime.now(timezone.utc) - timedelta(minutes=30)
            )
        )
        with self.assertRaises(HTTPException) as context:
            _require_recent_policy_authentication(stale)  # type: ignore[arg-type]
        self.assertEqual(403, context.exception.status_code)
        self.assertEqual(
            "recent_authentication_required",
            context.exception.detail["code"],
        )


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