Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion backend/app/services/knowledge/knowledge_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -422,9 +422,17 @@ def list_knowledge_bases(

# External resolver: returns additional KB IDs the user can access
# via extension rules (e.g. department / employee bindings).
# Errors are caught so a faulty extension cannot break the core listing path.
from app.services.readers.kb_permissions import kb_permission_resolver

ext_kb_ids = kb_permission_resolver.get_accessible_kb_ids(db, user_id)
try:
ext_kb_ids = kb_permission_resolver.get_accessible_kb_ids(db, user_id)
except Exception as e:
logger.warning(
f"kb_permissions extension get_accessible_kb_ids failed: {e}; "
"falling back to empty list"
)
ext_kb_ids = []

# Single query to get personal, team, organization, shared, bound, and
# externally-accessible knowledge bases.
Expand Down
7 changes: 6 additions & 1 deletion backend/app/services/readers/kb_permissions.py
Original file line number Diff line number Diff line change
Expand Up @@ -134,7 +134,12 @@ def _create_resolver() -> IKbPermissionResolver:
try:
ext = importlib.import_module(f"{settings.SERVICE_EXTENSION}.kb_permissions")
result = ext.wrap(base)
if result:
if result is None:
logger.warning(
"kb_permissions extension wrap() returned None; "
f"using default resolver ({settings.SERVICE_EXTENSION})"
)
else:
logger.info("KB permission resolver extension loaded")
return result
except ImportError:
Expand Down
22 changes: 7 additions & 15 deletions backend/tests/services/readers/test_kb_permission_resolver.py
Original file line number Diff line number Diff line change
Expand Up @@ -81,16 +81,10 @@ def test_create_resolver_returns_default_when_no_extension(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""When SERVICE_EXTENSION is empty, _create_resolver returns DefaultKbPermissionResolver."""
with patch("app.services.readers.kb_permissions.importlib") as _mock_importlib:
with patch("app.core.config.settings") as mock_settings:
mock_settings.SERVICE_EXTENSION = ""
# Re-import to pick up the patched settings
from app.services.readers import kb_permissions

with patch.object(kb_permissions, "_create_resolver") as mock_create:
mock_create.return_value = DefaultKbPermissionResolver()
result = mock_create()
from app.core.config import settings

monkeypatch.setattr(settings, "SERVICE_EXTENSION", "")
result = _create_resolver()
assert isinstance(result, DefaultKbPermissionResolver)


Expand Down Expand Up @@ -237,7 +231,7 @@ def test_get_user_kb_permission_calls_external_resolver_as_last_resort(
test_db: Session,
) -> None:
"""
When all built-in checks deny access, kbPermissionResolver.resolve is called
When all built-in checks deny access, kb_permission_resolver.resolve is called
and its returned role is used.
"""
from unittest.mock import MagicMock, patch
Expand All @@ -263,9 +257,7 @@ def test_get_user_kb_permission_calls_external_resolver_as_last_resort(
patch.object(_kss_module, "is_organization_namespace", return_value=False),
patch.object(_kss_module, "is_restricted_analyst", return_value=False),
):
# Simulate KB query returning mock_kb
mock_query.return_value.filter.return_value.first.return_value = mock_kb
# Simulate no explicit permission
# KB lookup returns mock_kb, subsequent explicit-permission lookup returns None
mock_query.return_value.filter.return_value.first.side_effect = [
mock_kb, # KB lookup
None, # no explicit ResourceMember
Expand All @@ -287,7 +279,7 @@ def test_get_user_kb_permission_external_resolver_not_called_when_creator(
test_db: Session,
) -> None:
"""
When the user is the KB creator, kbPermissionResolver.resolve must NOT be called.
When the user is the KB creator, kb_permission_resolver.resolve must NOT be called.
"""
from unittest.mock import MagicMock, patch

Expand All @@ -312,7 +304,7 @@ def test_get_user_kb_permission_external_resolver_not_called_when_creator(
):
mock_query.return_value.filter.return_value.first.return_value = mock_kb

has_access, role, is_creator = service.get_user_kb_permission(
has_access, _role, is_creator = service.get_user_kb_permission(
db, knowledge_base_id=55, user_id=7
)

Expand Down