busya
ce0369ae5c
Subagents delivered tests across all uncovered backend modules: Schemas (100%): agent, auth, health, profile, settings, validation Services (98-100%): auth, profile, health, llm, mapping, resource, security, git, superset_lookup, sql_table_extractor, rbac API routes (new): auth, admin, health, environments, plugins, dashboards (helpers, projection, actions, listing), git (config, deps, env, helpers) Clean Release (100%): DTO, facade, policy_engine, stages, repos, preparation, source_isolation, compliance Git services: base, remote_providers Agent module: app, run, middleware, langgraph_setup Core: trace, cleanup, ws_log_handler, timezone, auth (config/oauth/security), matching Reports: normalizer, report_service, type_profiles Notifications: service, providers Also: - .gitignore: add .coverage, *.cover, coverage-* dirs - src/schemas/auth.py: fix AD group DN regex (comma in CN=...) - Remove co-located src/services/__tests__/ (caused pytest module collision)
263 lines
12 KiB
Python
263 lines
12 KiB
Python
# #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
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_lru_caches():
|
|
"""Clear lru_cache between tests to prevent isolation failures."""
|
|
from src.services.rbac_permission_catalog import (
|
|
_discover_plugin_execute_permissions_cached,
|
|
_discover_route_permissions_cached,
|
|
)
|
|
_discover_route_permissions_cached.cache_clear()
|
|
_discover_plugin_execute_permissions_cached.cache_clear()
|
|
|
|
|
|
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()
|
|
|
|
|
|
class TestDiscoverRoutePermissionsErrors:
|
|
"""_discover_route_permissions — error handling."""
|
|
|
|
def test_oserror_on_read_continues(self):
|
|
"""Lines 60-65: OSError when reading route file is caught and logged."""
|
|
from src.services.rbac_permission_catalog import _discover_route_permissions
|
|
import tempfile
|
|
from pathlib import Path
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
route_dir = Path(tmpdir)
|
|
route_file = route_dir / "broken.py"
|
|
route_file.write_text("has_permission('dashboard', 'READ')")
|
|
# Make it unreadable
|
|
route_file.chmod(0o000)
|
|
with patch("src.services.rbac_permission_catalog.ROUTES_DIR", route_dir):
|
|
result = _discover_route_permissions()
|
|
# May get 0 or 1 depending on whether the file is actually unreadable
|
|
assert isinstance(result, set)
|
|
route_file.chmod(0o644)
|
|
|
|
|
|
class TestDiscoverPluginExecutePermissions:
|
|
"""_discover_plugin_execute_permissions — dynamic plugin permissions."""
|
|
|
|
def test_plugin_loader_none_returns_empty(self):
|
|
"""Line 94-95: None plugin_loader returns empty set."""
|
|
from src.services.rbac_permission_catalog import _discover_plugin_execute_permissions
|
|
result = _discover_plugin_execute_permissions(plugin_loader=None)
|
|
assert result == set()
|
|
|
|
def test_plugin_loader_exception_returns_empty(self):
|
|
"""Lines 99-104: exception from get_all_plugin_configs returns empty."""
|
|
from src.services.rbac_permission_catalog import _discover_plugin_execute_permissions
|
|
loader = MagicMock()
|
|
loader.get_all_plugin_configs.side_effect = RuntimeError("loader failed")
|
|
result = _discover_plugin_execute_permissions(plugin_loader=loader)
|
|
assert result == set()
|
|
|
|
def test_plugin_loader_with_configs(self):
|
|
"""Lines 106-110: plugin configs produce EXECUTE permissions."""
|
|
from src.services.rbac_permission_catalog import _discover_plugin_execute_permissions
|
|
config1 = MagicMock()
|
|
config1.id = "plugin-a"
|
|
config2 = MagicMock()
|
|
config2.id = "plugin-b"
|
|
loader = MagicMock()
|
|
loader.get_all_plugin_configs.return_value = [config1, config2]
|
|
result = _discover_plugin_execute_permissions(plugin_loader=loader)
|
|
assert ("plugin:plugin-a", "EXECUTE") in result
|
|
assert ("plugin:plugin-b", "EXECUTE") in result
|
|
assert len(result) == 2
|
|
|
|
def test_plugin_loader_empty_id_skipped(self):
|
|
"""Line 108: empty plugin id is skipped."""
|
|
from src.services.rbac_permission_catalog import _discover_plugin_execute_permissions
|
|
config = MagicMock()
|
|
config.id = ""
|
|
loader = MagicMock()
|
|
loader.get_all_plugin_configs.return_value = [config]
|
|
result = _discover_plugin_execute_permissions(plugin_loader=loader)
|
|
assert result == set()
|
|
|
|
|
|
class TestDiscoverPluginExecutePermissionsCached:
|
|
"""_discover_plugin_execute_permissions_cached — cached version."""
|
|
|
|
def test_cached_returns_tuples(self):
|
|
from src.services.rbac_permission_catalog import _discover_plugin_execute_permissions_cached
|
|
result = _discover_plugin_execute_permissions_cached(("p1", "p2"))
|
|
assert ("plugin:p1", "EXECUTE") in result
|
|
assert ("plugin:p2", "EXECUTE") in result
|
|
|
|
def test_cached_with_empty_ids(self):
|
|
from src.services.rbac_permission_catalog import _discover_plugin_execute_permissions_cached
|
|
result = _discover_plugin_execute_permissions_cached(())
|
|
assert result == ()
|
|
# #endregion Test.RbacPermissionCatalog
|