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
8 changes: 5 additions & 3 deletions backend/app/services/knowledge/knowledge_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -422,9 +422,9 @@ def list_knowledge_bases(

# External resolver: returns additional KB IDs the user can access
# via extension rules (e.g. department / employee bindings).
from app.services.readers.kb_permissions import kbPermissionResolver
from app.services.readers.kb_permissions import kb_permission_resolver

ext_kb_ids = kbPermissionResolver.get_accessible_kb_ids(db, user_id)
ext_kb_ids = kb_permission_resolver.get_accessible_kb_ids(db, user_id)

# Single query to get personal, team, organization, shared, bound, and
# externally-accessible knowledge bases.
Expand Down Expand Up @@ -2038,7 +2038,9 @@ def _get_user_kb_permission(
return True, BaseRole.Owner, False

has_access, role, is_creator = knowledge_share_service.get_user_kb_permission(
db, knowledge_base_id, user_id,
db,
knowledge_base_id,
user_id,
)

effective_role = BaseRole(role) if role is not None else None
Expand Down
8 changes: 4 additions & 4 deletions backend/app/services/readers/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,17 +14,17 @@
from app.services.readers.groups import groupReader
from app.services.readers.group_members import groupMemberReader
from app.services.readers.shared_teams import sharedTeamReader
from app.services.readers.kb_permissions import kbPermissionResolver
from app.services.readers.kb_permissions import kb_permission_resolver

bot = kindReader.get_by_name_and_namespace(db, user_id, KindType.BOT, "default", "mybot")
user = userReader.get_by_id(db, user_id)
is_member = groupMemberReader.is_member(db, "group-name", user_id)
ids = kbPermissionResolver.get_accessible_kb_ids(db, user_id)
ids = kb_permission_resolver.get_accessible_kb_ids(db, user_id)
"""

from app.services.readers.group_members import groupMemberReader
from app.services.readers.groups import groupReader
from app.services.readers.kb_permissions import kbPermissionResolver
from app.services.readers.kb_permissions import kb_permission_resolver
from app.services.readers.kinds import KindType, kindReader
from app.services.readers.shared_teams import sharedTeamReader
from app.services.readers.users import userReader
Expand All @@ -35,6 +35,6 @@
"groupReader",
"groupMemberReader",
"sharedTeamReader",
"kbPermissionResolver",
"kb_permission_resolver",
"KindType",
]
2 changes: 1 addition & 1 deletion backend/app/services/readers/kb_permissions.py
Original file line number Diff line number Diff line change
Expand Up @@ -171,4 +171,4 @@ def __getattr__(self, name: str):
# Export
# =============================================================================

kbPermissionResolver: IKbPermissionResolver = _LazyReader() # type: ignore
kb_permission_resolver: IKbPermissionResolver = _LazyReader() # type: ignore
4 changes: 2 additions & 2 deletions backend/app/services/share/knowledge_share_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -382,9 +382,9 @@ def get_user_kb_permission(

# External permission resolver (e.g. department / employee bindings).
# Called last so it never interferes with built-in access control.
from app.services.readers.kb_permissions import kbPermissionResolver
from app.services.readers.kb_permissions import kb_permission_resolver

ext_role = kbPermissionResolver.resolve(db, knowledge_base_id, user_id, kb)
ext_role = kb_permission_resolver.resolve(db, knowledge_base_id, user_id, kb)
if ext_role is not None:
return True, ext_role, False

Expand Down
42 changes: 19 additions & 23 deletions backend/tests/services/readers/test_kb_permission_resolver.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,17 +21,20 @@
from typing import Optional
from unittest.mock import MagicMock, patch

# Use importlib.import_module to get the actual module object, not the
# instance exported by app.services.share.__init__.py under the same name.
_kss_module = importlib.import_module("app.services.share.knowledge_share_service")

import pytest
from sqlalchemy.orm import Session

from app.services.readers.kb_permissions import (
DefaultKbPermissionResolver,
IKbPermissionResolver,
_LazyReader,
_create_resolver,
_LazyReader,
)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
Expand Down Expand Up @@ -139,12 +142,12 @@ def test_create_resolver_falls_back_on_import_error(
"""ImportError during extension load logs a warning and returns the default resolver."""
with patch("app.core.config.settings") as mock_settings:
mock_settings.SERVICE_EXTENSION = "missing_ext"
with patch(
"importlib.import_module", side_effect=ImportError("no module")
):
with patch("importlib.import_module", side_effect=ImportError("no module")):
import logging

with caplog.at_level(logging.WARNING, logger="app.services.readers.kb_permissions"):
with caplog.at_level(
logging.WARNING, logger="app.services.readers.kb_permissions"
):
from app.services.readers import kb_permissions

result = kb_permissions._create_resolver()
Expand All @@ -171,7 +174,9 @@ def test_create_resolver_falls_back_on_general_exception(
with patch("importlib.import_module", return_value=fake_ext):
import logging

with caplog.at_level(logging.WARNING, logger="app.services.readers.kb_permissions"):
with caplog.at_level(
logging.WARNING, logger="app.services.readers.kb_permissions"
):
from app.services.readers import kb_permissions

result = kb_permissions._create_resolver()
Expand Down Expand Up @@ -253,23 +258,17 @@ def test_get_user_kb_permission_calls_external_resolver_as_last_resort(
with (
patch.object(db, "query") as mock_query,
patch(
"app.services.readers.kb_permissions.kbPermissionResolver"
"app.services.readers.kb_permissions.kb_permission_resolver"
) as mock_resolver,
patch(
"app.services.share.knowledge_share_service.is_organization_namespace",
return_value=False,
),
patch(
"app.services.share.knowledge_share_service.is_restricted_analyst",
return_value=False,
),
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
mock_query.return_value.filter.return_value.first.side_effect = [
mock_kb, # KB lookup
None, # no explicit ResourceMember
mock_kb, # KB lookup
None, # no explicit ResourceMember
]
# External resolver grants Reporter access
mock_resolver.resolve.return_value = "Reporter"
Expand Down Expand Up @@ -307,12 +306,9 @@ def test_get_user_kb_permission_external_resolver_not_called_when_creator(
with (
patch.object(db, "query") as mock_query,
patch(
"app.services.readers.kb_permissions.kbPermissionResolver"
"app.services.readers.kb_permissions.kb_permission_resolver"
) as mock_resolver,
patch(
"app.services.share.knowledge_share_service.is_restricted_analyst",
return_value=False,
),
patch.object(_kss_module, "is_restricted_analyst", return_value=False),
):
mock_query.return_value.filter.return_value.first.return_value = mock_kb

Expand Down