ss-tools/backend/tests/services/test_rbac_permission_catalog.py

# #region Test.RbacPermissionCatalog [C:3] [TYPE Module] [SEMANTICS test,rbac,permission,discovery,route]
# @BRIEF Tests for services/rbac_permission_catalog.py — permission discovery, route scanning, sync.
# @RELATION BINDS_TO -> [rbac_permission_catalog]

from pathlib import Path
import sys

sys.path.insert(0, str(Path(__file__).parent.parent / "src"))

import os
import tempfile
from unittest.mock import MagicMock, patch, PropertyMock
import pytest


class TestIterRouteFiles:
    """_iter_route_files — discover Python files in routes directory."""

    def test_directory_not_found(self):
        from src.services.rbac_permission_catalog import _iter_route_files
        with patch("src.services.rbac_permission_catalog.ROUTES_DIR", Path("/nonexistent/path")):
            result = _iter_route_files()
            assert result == []

    def test_finds_py_files(self):
        from src.services.rbac_permission_catalog import _iter_route_files
        with tempfile.TemporaryDirectory() as tmpdir:
            routes_dir = Path(tmpdir) / "routes"
            routes_dir.mkdir()
            (routes_dir / "test_route.py").write_text("")
            (routes_dir / "__init__.py").write_text("")
            (routes_dir / "subdir").mkdir()
            (routes_dir / "subdir" / "nested.py").write_text("")
            # These should be excluded
            (routes_dir / "__tests__").mkdir()
            (routes_dir / "__tests__" / "test_excluded.py").write_text("")
            (routes_dir / "__pycache__").mkdir()
            (routes_dir / "__pycache__" / "cached.py").write_text("")

            with patch("src.services.rbac_permission_catalog.ROUTES_DIR", routes_dir):
                files = _iter_route_files()
                filenames = [f.name for f in files]
                assert "test_route.py" in filenames
                assert "__init__.py" in filenames
                assert "nested.py" in filenames
                assert "test_excluded.py" not in filenames
                assert "cached.py" not in filenames


class TestDiscoverRoutePermissions:
    """_discover_route_permissions — extract has_permission() calls from route files."""

    def test_extracts_permissions(self):
        from src.services.rbac_permission_catalog import _discover_route_permissions
        with tempfile.TemporaryDirectory() as tmpdir:
            route_file = Path(tmpdir) / "test_route.py"
            route_file.write_text("""
                has_permission("dashboards", "READ")
                has_permission("datasets", "WRITE")
                # Not a permission call:
                x = has_permission("users", "ADMIN")
            """)

            with patch("src.services.rbac_permission_catalog.ROUTES_DIR", Path(tmpdir)):
                result = _discover_route_permissions()
                assert ("dashboards", "READ") in result
                assert ("datasets", "WRITE") in result
                assert ("users", "ADMIN") in result

    def test_empty_file(self):
        from src.services.rbac_permission_catalog import _discover_route_permissions
        with tempfile.TemporaryDirectory() as tmpdir:
            route_file = Path(tmpdir) / "empty.py"
            route_file.write_text("# No permissions here")

            with patch("src.services.rbac_permission_catalog.ROUTES_DIR", Path(tmpdir)):
                result = _discover_route_permissions()
                assert result == set()

    def test_multiple_files(self):
        from src.services.rbac_permission_catalog import _discover_route_permissions
        with tempfile.TemporaryDirectory() as tmpdir:
            f1 = Path(tmpdir) / "routes_a.py"
            f1.write_text('has_permission("dashboard", "READ")')
            f2 = Path(tmpdir) / "routes_b.py"
            f2.write_text('has_permission("dataset", "WRITE")')

            with patch("src.services.rbac_permission_catalog.ROUTES_DIR", Path(tmpdir)):
                result = _discover_route_permissions()
                assert ("dashboard", "READ") in result
                assert ("dataset", "WRITE") in result

    def test_single_quote_variants(self):
        from src.services.rbac_permission_catalog import _discover_route_permissions
        with tempfile.TemporaryDirectory() as tmpdir:
            f = Path(tmpdir) / "test.py"
            f.write_text("""has_permission('dashboard', 'READ')""")

            with patch("src.services.rbac_permission_catalog.ROUTES_DIR", Path(tmpdir)):
                result = _discover_route_permissions()
                assert ("dashboard", "READ") in result


class TestDiscoverDeclaredPermissions:
    """discover_declared_permissions — cached public API."""

    def test_returns_permissions(self):
        from src.services.rbac_permission_catalog import discover_declared_permissions
        with patch("src.services.rbac_permission_catalog._discover_route_permissions",
                   return_value={("dashboard", "READ"), ("dataset", "WRITE")}):
            result = discover_declared_permissions()
            assert ("dashboard", "READ") in result
            assert ("dataset", "WRITE") in result

    def test_caching(self):
        from src.services.rbac_permission_catalog import discover_declared_permissions
        mock_fn = MagicMock(return_value={("dashboard", "READ")})
        with patch("src.services.rbac_permission_catalog._discover_route_permissions", mock_fn):
            result1 = discover_declared_permissions()
            result2 = discover_declared_permissions()
            assert result1 == result2
            # With no plugin_loader, it calls _discover_route_permissions
            # The lru_cache should only call once if args match
            # discover_declared_permissions takes optional plugin_loader

    def test_with_plugin_loader(self):
        from src.services.rbac_permission_catalog import discover_declared_permissions
        mock_loader = MagicMock()
        mock_loader.list_plugins.return_value = []
        mock_fn = MagicMock(return_value={("dashboard", "READ")})
        with patch("src.services.rbac_permission_catalog._discover_route_permissions", mock_fn):
            result = discover_declared_permissions(plugin_loader=mock_loader)
            # Should call plugin loader to discover plugin permissions
            assert ("dashboard", "READ") in result


class TestSyncPermissionCatalog:
    """sync_permission_catalog — sync discovered permissions to DB."""

    def test_creates_new_permissions(self):
        from src.services.rbac_permission_catalog import sync_permission_catalog

        db = MagicMock()
        db.query.return_value.all.return_value = []  # No existing permissions

        discovered = {("dashboard", "READ"), ("dataset", "WRITE")}
        result = sync_permission_catalog(db, discovered)
        assert result == 2  # 2 new permissions
        assert db.add.call_count >= 2
        db.commit.assert_called_once()

    def test_skips_existing_permissions(self):
        from src.services.rbac_permission_catalog import sync_permission_catalog

        existing = MagicMock()
        existing.resource = "dashboard"
        existing.action = "READ"

        db = MagicMock()
        db.query.return_value.all.return_value = [existing]

        discovered = {("dashboard", "READ"), ("dataset", "WRITE")}
        result = sync_permission_catalog(db, discovered)
        assert result == 1  # 1 new (dataset/WRITE), 1 skipped
        db.commit.assert_called_once()

    def test_empty_discovered(self):
        from src.services.rbac_permission_catalog import sync_permission_catalog
        db = MagicMock()
        result = sync_permission_catalog(db, set())
        assert result == 0
        db.commit.assert_not_called()
# #endregion Test.RbacPermissionCatalog