diff --git a/backend/app/services/knowledge/knowledge_service.py b/backend/app/services/knowledge/knowledge_service.py index a3761626f..db97a8c25 100644 --- a/backend/app/services/knowledge/knowledge_service.py +++ b/backend/app/services/knowledge/knowledge_service.py @@ -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. @@ -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 diff --git a/backend/app/services/readers/__init__.py b/backend/app/services/readers/__init__.py index 1e2533329..e23ebc9ed 100644 --- a/backend/app/services/readers/__init__.py +++ b/backend/app/services/readers/__init__.py @@ -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 @@ -35,6 +35,6 @@ "groupReader", "groupMemberReader", "sharedTeamReader", - "kbPermissionResolver", + "kb_permission_resolver", "KindType", ] diff --git a/backend/app/services/readers/kb_permissions.py b/backend/app/services/readers/kb_permissions.py index 73e52fc00..bc91e7a64 100644 --- a/backend/app/services/readers/kb_permissions.py +++ b/backend/app/services/readers/kb_permissions.py @@ -171,4 +171,4 @@ def __getattr__(self, name: str): # Export # ============================================================================= -kbPermissionResolver: IKbPermissionResolver = _LazyReader() # type: ignore +kb_permission_resolver: IKbPermissionResolver = _LazyReader() # type: ignore diff --git a/backend/app/services/share/knowledge_share_service.py b/backend/app/services/share/knowledge_share_service.py index 43f9d781e..577b93521 100644 --- a/backend/app/services/share/knowledge_share_service.py +++ b/backend/app/services/share/knowledge_share_service.py @@ -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 diff --git a/backend/tests/services/readers/test_kb_permission_resolver.py b/backend/tests/services/readers/test_kb_permission_resolver.py index f6443e728..a9e83814d 100644 --- a/backend/tests/services/readers/test_kb_permission_resolver.py +++ b/backend/tests/services/readers/test_kb_permission_resolver.py @@ -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 # --------------------------------------------------------------------------- @@ -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() @@ -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() @@ -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" @@ -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