Travis Vasceannie
1ce24cdf7b
- Moved all hookify configuration files from `.claude/` to `.claude/hooks/` subdirectory for better organization - Added four new blocking hooks to prevent common error handling anti-patterns: - `block-broad-exception-handler`: Prevents catching generic `Exception` with only logging - `block-datetime-now-fallback`: Blocks returning `datetime.now()` as fallback on parse failures to prevent data corruption - `block-default
691 lines
26 KiB
Python
691 lines
26 KiB
Python
"""Tests for IdentityMixin gRPC endpoints.
|
|
|
|
Tests cover:
|
|
- GetCurrentUser: Returns user identity, workspace, and auth status
|
|
- ListWorkspaces: Lists user's workspaces with pagination
|
|
- SwitchWorkspace: Validates workspace access and returns workspace info
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from uuid import UUID, uuid4
|
|
|
|
import pytest
|
|
|
|
from noteflow.application.services.identity import IdentityService
|
|
from noteflow.domain.entities.integration import Integration, IntegrationType
|
|
from noteflow.domain.identity.context import OperationContext, UserContext, WorkspaceContext
|
|
from noteflow.domain.identity.entities import (
|
|
User,
|
|
Workspace,
|
|
WorkspaceMembership,
|
|
WorkspaceSettings,
|
|
)
|
|
from noteflow.domain.identity.roles import WorkspaceRole
|
|
from noteflow.grpc.mixins._types import GrpcContext
|
|
from noteflow.grpc.mixins.identity import IdentityMixin
|
|
from noteflow.grpc.proto import noteflow_pb2
|
|
|
|
if TYPE_CHECKING:
|
|
from datetime import datetime
|
|
|
|
|
|
# =============================================================================
|
|
# Mock Servicer Host
|
|
# =============================================================================
|
|
|
|
|
|
class MockIdentityServicerHost(IdentityMixin):
|
|
"""Mock servicer host implementing required protocol for IdentityMixin."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Initialize mock servicer with identity service."""
|
|
self._identity_service: IdentityService = IdentityService()
|
|
self._mock_uow: MagicMock | None = None
|
|
|
|
def create_repository_provider(self) -> MagicMock:
|
|
"""Return mock UnitOfWork."""
|
|
if self._mock_uow is None:
|
|
msg = "Mock UoW not configured"
|
|
raise RuntimeError(msg)
|
|
return self._mock_uow
|
|
|
|
def get_operation_context(self, context: GrpcContext) -> OperationContext:
|
|
"""Return mock operation context."""
|
|
return OperationContext(
|
|
user=UserContext(
|
|
user_id=uuid4(),
|
|
display_name="Test User",
|
|
),
|
|
workspace=WorkspaceContext(
|
|
workspace_id=uuid4(),
|
|
workspace_name="Test Workspace",
|
|
role=WorkspaceRole.OWNER,
|
|
),
|
|
)
|
|
|
|
@property
|
|
def identity_service(self) -> IdentityService:
|
|
"""Return identity service."""
|
|
return self._identity_service
|
|
|
|
def set_mock_uow(self, uow: MagicMock) -> None:
|
|
"""Set mock UnitOfWork for testing."""
|
|
self._mock_uow = uow
|
|
|
|
# Type stubs for mixin methods
|
|
if TYPE_CHECKING:
|
|
async def GetCurrentUser(
|
|
self,
|
|
request: noteflow_pb2.GetCurrentUserRequest,
|
|
context: GrpcContext,
|
|
) -> noteflow_pb2.GetCurrentUserResponse: ...
|
|
|
|
async def ListWorkspaces(
|
|
self,
|
|
request: noteflow_pb2.ListWorkspacesRequest,
|
|
context: GrpcContext,
|
|
) -> noteflow_pb2.ListWorkspacesResponse: ...
|
|
|
|
async def SwitchWorkspace(
|
|
self,
|
|
request: noteflow_pb2.SwitchWorkspaceRequest,
|
|
context: GrpcContext,
|
|
) -> noteflow_pb2.SwitchWorkspaceResponse: ...
|
|
|
|
|
|
# =============================================================================
|
|
# Fixtures
|
|
# =============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def identity_servicer() -> MockIdentityServicerHost:
|
|
"""Create servicer for identity mixin testing."""
|
|
return MockIdentityServicerHost()
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_identity_uow() -> MagicMock:
|
|
"""Create mock UnitOfWork with identity-related repositories."""
|
|
uow = MagicMock()
|
|
uow.__aenter__ = AsyncMock(return_value=uow)
|
|
uow.__aexit__ = AsyncMock(return_value=None)
|
|
uow.commit = AsyncMock()
|
|
|
|
# Users repository
|
|
uow.supports_users = True
|
|
uow.users = MagicMock()
|
|
uow.users.get = AsyncMock(return_value=None)
|
|
uow.users.get_default = AsyncMock(return_value=None)
|
|
uow.users.create_default = AsyncMock()
|
|
|
|
# Workspaces repository
|
|
uow.supports_workspaces = True
|
|
uow.workspaces = MagicMock()
|
|
uow.workspaces.get = AsyncMock(return_value=None)
|
|
uow.workspaces.get_default_for_user = AsyncMock(return_value=None)
|
|
uow.workspaces.get_membership = AsyncMock(return_value=None)
|
|
uow.workspaces.list_for_user = AsyncMock(return_value=[])
|
|
uow.workspaces.create = AsyncMock()
|
|
|
|
# Integrations repository
|
|
uow.supports_integrations = True
|
|
uow.integrations = MagicMock()
|
|
uow.integrations.get_by_provider = AsyncMock(return_value=None)
|
|
|
|
# Projects repository (for workspace creation)
|
|
uow.supports_projects = False
|
|
|
|
return uow
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_user_context() -> UserContext:
|
|
"""Create sample user context for testing."""
|
|
return UserContext(
|
|
user_id=uuid4(),
|
|
display_name="Test User",
|
|
email="test@example.com",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_workspace_context() -> WorkspaceContext:
|
|
"""Create sample workspace context for testing."""
|
|
return WorkspaceContext(
|
|
workspace_id=uuid4(),
|
|
workspace_name="Test Workspace",
|
|
role=WorkspaceRole.OWNER,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_workspace(sample_datetime: datetime) -> Workspace:
|
|
"""Create sample workspace for testing."""
|
|
return Workspace(
|
|
id=uuid4(),
|
|
name="Test Workspace",
|
|
slug="test-workspace",
|
|
is_default=True,
|
|
created_at=sample_datetime,
|
|
updated_at=sample_datetime,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_workspace_membership() -> WorkspaceMembership:
|
|
"""Create sample workspace membership for testing."""
|
|
return WorkspaceMembership(
|
|
workspace_id=uuid4(),
|
|
user_id=uuid4(),
|
|
role=WorkspaceRole.MEMBER,
|
|
)
|
|
|
|
|
|
def build_workspace_update_payload(
|
|
workspace_id: UUID,
|
|
user_id: UUID,
|
|
template_id: str,
|
|
) -> tuple[Workspace, WorkspaceMembership, Workspace]:
|
|
"""Create a baseline workspace, membership, and updated workspace."""
|
|
workspace = Workspace(id=workspace_id, name="Workspace")
|
|
membership = WorkspaceMembership(
|
|
workspace_id=workspace_id,
|
|
user_id=user_id,
|
|
role=WorkspaceRole.OWNER,
|
|
)
|
|
updated_workspace = Workspace(
|
|
id=workspace_id,
|
|
name="Workspace",
|
|
settings=WorkspaceSettings(default_summarization_template=template_id),
|
|
)
|
|
return workspace, membership, updated_workspace
|
|
|
|
|
|
# =============================================================================
|
|
# Test: GetCurrentUser
|
|
# =============================================================================
|
|
|
|
|
|
class TestGetCurrentUser:
|
|
"""Tests for IdentityMixin.GetCurrentUser."""
|
|
|
|
async def test_returns_default_user_in_memory_mode(
|
|
self,
|
|
identity_servicer: MockIdentityServicerHost,
|
|
mock_grpc_context: MagicMock,
|
|
) -> None:
|
|
"""GetCurrentUser returns default user when in memory mode."""
|
|
uow = MagicMock()
|
|
uow.__aenter__ = AsyncMock(return_value=uow)
|
|
uow.__aexit__ = AsyncMock(return_value=None)
|
|
uow.commit = AsyncMock()
|
|
uow.supports_users = False
|
|
uow.supports_workspaces = False
|
|
uow.supports_integrations = False
|
|
|
|
identity_servicer.set_mock_uow(uow)
|
|
|
|
request = noteflow_pb2.GetCurrentUserRequest()
|
|
response = await identity_servicer.GetCurrentUser(request, mock_grpc_context)
|
|
|
|
assert response.user_id, "should return user_id"
|
|
assert response.workspace_id, "should return workspace_id"
|
|
assert response.display_name, "should return display_name"
|
|
assert response.is_authenticated is False, "should not be authenticated in memory mode"
|
|
|
|
async def test_returns_authenticated_user_with_oauth(
|
|
self,
|
|
identity_servicer: MockIdentityServicerHost,
|
|
mock_identity_uow: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
) -> None:
|
|
"""GetCurrentUser returns authenticated user when OAuth integration exists."""
|
|
# Configure connected integration
|
|
integration = Integration.create(
|
|
workspace_id=uuid4(),
|
|
name="Google Auth",
|
|
integration_type=IntegrationType.AUTH,
|
|
config={"provider": "google"},
|
|
)
|
|
integration.connect(provider_email="test@example.com")
|
|
mock_identity_uow.integrations.get_by_provider.return_value = integration
|
|
|
|
identity_servicer.set_mock_uow(mock_identity_uow)
|
|
|
|
request = noteflow_pb2.GetCurrentUserRequest()
|
|
response = await identity_servicer.GetCurrentUser(request, mock_grpc_context)
|
|
|
|
assert response.is_authenticated is True, "should be authenticated with OAuth"
|
|
assert response.auth_provider == "google", "should return auth provider"
|
|
|
|
async def test_returns_workspace_role(
|
|
self,
|
|
identity_servicer: MockIdentityServicerHost,
|
|
mock_identity_uow: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
) -> None:
|
|
"""GetCurrentUser returns user's workspace role."""
|
|
identity_servicer.set_mock_uow(mock_identity_uow)
|
|
|
|
request = noteflow_pb2.GetCurrentUserRequest()
|
|
response = await identity_servicer.GetCurrentUser(request, mock_grpc_context)
|
|
|
|
# Role should be set (default is owner for first user)
|
|
assert response.role, "should return workspace role"
|
|
assert response.workspace_name, "should return workspace name"
|
|
|
|
|
|
# =============================================================================
|
|
# Test: ListWorkspaces
|
|
# =============================================================================
|
|
|
|
|
|
class TestListWorkspaces:
|
|
"""Tests for IdentityMixin.ListWorkspaces."""
|
|
|
|
async def test_returns_empty_list_when_no_workspaces(
|
|
self,
|
|
identity_servicer: MockIdentityServicerHost,
|
|
mock_identity_uow: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
) -> None:
|
|
"""ListWorkspaces returns empty list when user has no workspaces."""
|
|
identity_servicer.set_mock_uow(mock_identity_uow)
|
|
|
|
request = noteflow_pb2.ListWorkspacesRequest()
|
|
response = await identity_servicer.ListWorkspaces(request, mock_grpc_context)
|
|
|
|
assert response.total_count == 0, "should return zero count"
|
|
assert len(response.workspaces) == 0, "should return empty list"
|
|
|
|
async def test_returns_user_workspaces(
|
|
self,
|
|
identity_servicer: MockIdentityServicerHost,
|
|
mock_identity_uow: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
sample_workspace: Workspace,
|
|
sample_workspace_membership: WorkspaceMembership,
|
|
) -> None:
|
|
"""ListWorkspaces returns workspaces the user belongs to."""
|
|
mock_identity_uow.workspaces.list_for_user.return_value = [sample_workspace]
|
|
mock_identity_uow.workspaces.get_membership.return_value = sample_workspace_membership
|
|
|
|
identity_servicer.set_mock_uow(mock_identity_uow)
|
|
|
|
request = noteflow_pb2.ListWorkspacesRequest()
|
|
response = await identity_servicer.ListWorkspaces(request, mock_grpc_context)
|
|
|
|
assert response.total_count == 1, "should return correct count"
|
|
assert len(response.workspaces) == 1, "should return one workspace"
|
|
assert response.workspaces[0].name == "Test Workspace", "should include workspace name"
|
|
assert response.workspaces[0].is_default is True, "should include is_default flag"
|
|
|
|
async def test_respects_pagination_parameters(
|
|
self,
|
|
identity_servicer: MockIdentityServicerHost,
|
|
mock_identity_uow: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
) -> None:
|
|
"""ListWorkspaces passes pagination parameters to repository."""
|
|
identity_servicer.set_mock_uow(mock_identity_uow)
|
|
|
|
request = noteflow_pb2.ListWorkspacesRequest(limit=10, offset=5)
|
|
await identity_servicer.ListWorkspaces(request, mock_grpc_context)
|
|
|
|
# Check that list_for_user was called with pagination params
|
|
mock_identity_uow.workspaces.list_for_user.assert_called()
|
|
call_args = mock_identity_uow.workspaces.list_for_user.call_args
|
|
EXPECTED_LIMIT = 10
|
|
EXPECTED_OFFSET = 5
|
|
assert call_args[0][1] == EXPECTED_LIMIT, "should pass limit"
|
|
assert call_args[0][2] == EXPECTED_OFFSET, "should pass offset"
|
|
|
|
async def test_uses_default_pagination_values(
|
|
self,
|
|
identity_servicer: MockIdentityServicerHost,
|
|
mock_identity_uow: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
) -> None:
|
|
"""ListWorkspaces uses default pagination when not specified."""
|
|
identity_servicer.set_mock_uow(mock_identity_uow)
|
|
|
|
request = noteflow_pb2.ListWorkspacesRequest() # No limit/offset
|
|
await identity_servicer.ListWorkspaces(request, mock_grpc_context)
|
|
|
|
call_args = mock_identity_uow.workspaces.list_for_user.call_args
|
|
DEFAULT_LIMIT = 50
|
|
DEFAULT_OFFSET = 0
|
|
assert call_args[0][1] == DEFAULT_LIMIT, "should use default limit"
|
|
assert call_args[0][2] == DEFAULT_OFFSET, "should use default offset"
|
|
|
|
async def test_includes_workspace_role(
|
|
self,
|
|
identity_servicer: MockIdentityServicerHost,
|
|
mock_identity_uow: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
sample_workspace: Workspace,
|
|
) -> None:
|
|
"""ListWorkspaces includes user's role in each workspace."""
|
|
owner_membership = WorkspaceMembership(
|
|
workspace_id=sample_workspace.id,
|
|
user_id=uuid4(),
|
|
role=WorkspaceRole.OWNER,
|
|
)
|
|
mock_identity_uow.workspaces.list_for_user.return_value = [sample_workspace]
|
|
mock_identity_uow.workspaces.get_membership.return_value = owner_membership
|
|
|
|
identity_servicer.set_mock_uow(mock_identity_uow)
|
|
|
|
request = noteflow_pb2.ListWorkspacesRequest()
|
|
response = await identity_servicer.ListWorkspaces(request, mock_grpc_context)
|
|
|
|
assert response.workspaces[0].role == "owner", "should include role"
|
|
|
|
|
|
# =============================================================================
|
|
# Test: SwitchWorkspace
|
|
# =============================================================================
|
|
|
|
|
|
class TestSwitchWorkspace:
|
|
"""Tests for IdentityMixin.SwitchWorkspace."""
|
|
|
|
async def test_aborts_when_workspace_id_missing(
|
|
self,
|
|
identity_servicer: MockIdentityServicerHost,
|
|
mock_identity_uow: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
) -> None:
|
|
"""SwitchWorkspace aborts with INVALID_ARGUMENT when workspace_id not provided."""
|
|
identity_servicer.set_mock_uow(mock_identity_uow)
|
|
|
|
request = noteflow_pb2.SwitchWorkspaceRequest() # No workspace_id
|
|
|
|
with pytest.raises(AssertionError, match="Unreachable"):
|
|
await identity_servicer.SwitchWorkspace(request, mock_grpc_context)
|
|
|
|
mock_grpc_context.abort.assert_called_once()
|
|
|
|
async def test_aborts_for_invalid_uuid(
|
|
self,
|
|
identity_servicer: MockIdentityServicerHost,
|
|
mock_identity_uow: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
) -> None:
|
|
"""SwitchWorkspace aborts with INVALID_ARGUMENT for invalid workspace_id format."""
|
|
identity_servicer.set_mock_uow(mock_identity_uow)
|
|
|
|
request = noteflow_pb2.SwitchWorkspaceRequest(workspace_id="not-a-uuid")
|
|
|
|
with pytest.raises(AssertionError, match="Unreachable"):
|
|
await identity_servicer.SwitchWorkspace(request, mock_grpc_context)
|
|
|
|
mock_grpc_context.abort.assert_called_once()
|
|
|
|
async def test_aborts_when_workspace_not_found(
|
|
self,
|
|
identity_servicer: MockIdentityServicerHost,
|
|
mock_identity_uow: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
) -> None:
|
|
"""SwitchWorkspace aborts with NOT_FOUND when workspace doesn't exist."""
|
|
mock_identity_uow.workspaces.get.return_value = None
|
|
|
|
identity_servicer.set_mock_uow(mock_identity_uow)
|
|
|
|
workspace_id = uuid4()
|
|
request = noteflow_pb2.SwitchWorkspaceRequest(workspace_id=str(workspace_id))
|
|
|
|
with pytest.raises(AssertionError, match="Unreachable"):
|
|
await identity_servicer.SwitchWorkspace(request, mock_grpc_context)
|
|
|
|
mock_grpc_context.abort.assert_called_once()
|
|
|
|
async def test_aborts_when_user_not_member(
|
|
self,
|
|
identity_servicer: MockIdentityServicerHost,
|
|
mock_identity_uow: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
sample_workspace: Workspace,
|
|
) -> None:
|
|
"""SwitchWorkspace aborts with NOT_FOUND when user is not a member of workspace."""
|
|
mock_identity_uow.workspaces.get.return_value = sample_workspace
|
|
mock_identity_uow.workspaces.get_membership.return_value = None # No membership
|
|
|
|
identity_servicer.set_mock_uow(mock_identity_uow)
|
|
|
|
request = noteflow_pb2.SwitchWorkspaceRequest(workspace_id=str(sample_workspace.id))
|
|
|
|
with pytest.raises(AssertionError, match="Unreachable"):
|
|
await identity_servicer.SwitchWorkspace(request, mock_grpc_context)
|
|
|
|
mock_grpc_context.abort.assert_called_once()
|
|
|
|
async def test_switches_workspace_successfully(
|
|
self,
|
|
identity_servicer: MockIdentityServicerHost,
|
|
mock_identity_uow: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
sample_workspace: Workspace,
|
|
sample_workspace_membership: WorkspaceMembership,
|
|
) -> None:
|
|
"""SwitchWorkspace returns workspace info on success."""
|
|
mock_identity_uow.workspaces.get.return_value = sample_workspace
|
|
mock_identity_uow.workspaces.get_membership.return_value = sample_workspace_membership
|
|
|
|
identity_servicer.set_mock_uow(mock_identity_uow)
|
|
|
|
request = noteflow_pb2.SwitchWorkspaceRequest(workspace_id=str(sample_workspace.id))
|
|
response = await identity_servicer.SwitchWorkspace(request, mock_grpc_context)
|
|
|
|
expected_workspace_id = str(sample_workspace.id)
|
|
assert response.success is True, "should return success=True"
|
|
assert response.workspace.id == expected_workspace_id, "should return workspace ID"
|
|
assert response.workspace.name == "Test Workspace", "should return workspace name"
|
|
assert response.workspace.role == "member", "should return user's role"
|
|
|
|
@pytest.mark.parametrize(
|
|
("role", "expected_role_str"),
|
|
[
|
|
pytest.param(WorkspaceRole.OWNER, "owner", id="owner_role"),
|
|
pytest.param(WorkspaceRole.ADMIN, "admin", id="admin_role"),
|
|
pytest.param(WorkspaceRole.MEMBER, "member", id="member_role"),
|
|
],
|
|
)
|
|
async def test_returns_correct_role_for_membership(
|
|
self,
|
|
identity_servicer: MockIdentityServicerHost,
|
|
mock_identity_uow: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
sample_workspace: Workspace,
|
|
role: WorkspaceRole,
|
|
expected_role_str: str,
|
|
) -> None:
|
|
"""SwitchWorkspace returns correct role string for different memberships."""
|
|
membership = WorkspaceMembership(
|
|
workspace_id=sample_workspace.id,
|
|
user_id=uuid4(),
|
|
role=role,
|
|
)
|
|
mock_identity_uow.workspaces.get.return_value = sample_workspace
|
|
mock_identity_uow.workspaces.get_membership.return_value = membership
|
|
|
|
identity_servicer.set_mock_uow(mock_identity_uow)
|
|
|
|
request = noteflow_pb2.SwitchWorkspaceRequest(workspace_id=str(sample_workspace.id))
|
|
response = await identity_servicer.SwitchWorkspace(request, mock_grpc_context)
|
|
|
|
assert response.workspace.role == expected_role_str, f"should return role as {expected_role_str}"
|
|
|
|
async def test_includes_workspace_metadata(
|
|
self,
|
|
identity_servicer: MockIdentityServicerHost,
|
|
mock_identity_uow: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
sample_workspace_membership: WorkspaceMembership,
|
|
) -> None:
|
|
"""SwitchWorkspace includes workspace slug and is_default flag."""
|
|
workspace = Workspace(
|
|
id=uuid4(),
|
|
name="My Custom Workspace",
|
|
slug="my-custom-workspace",
|
|
is_default=False,
|
|
)
|
|
mock_identity_uow.workspaces.get.return_value = workspace
|
|
mock_identity_uow.workspaces.get_membership.return_value = sample_workspace_membership
|
|
|
|
identity_servicer.set_mock_uow(mock_identity_uow)
|
|
|
|
request = noteflow_pb2.SwitchWorkspaceRequest(workspace_id=str(workspace.id))
|
|
response = await identity_servicer.SwitchWorkspace(request, mock_grpc_context)
|
|
|
|
assert response.workspace.slug == "my-custom-workspace", "should include slug"
|
|
assert response.workspace.is_default is False, "should include is_default flag"
|
|
|
|
|
|
# =============================================================================
|
|
# Test: GetWorkspaceSettings
|
|
# =============================================================================
|
|
|
|
|
|
class TestGetWorkspaceSettings:
|
|
"""Tests for IdentityMixin.GetWorkspaceSettings."""
|
|
|
|
async def test_returns_workspace_settings(
|
|
self,
|
|
identity_servicer: MockIdentityServicerHost,
|
|
mock_identity_uow: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
) -> None:
|
|
"""GetWorkspaceSettings returns stored workspace settings."""
|
|
workspace_id = uuid4()
|
|
user_id = uuid4()
|
|
settings = WorkspaceSettings(
|
|
rag_enabled=True,
|
|
default_summarization_template="template-123",
|
|
)
|
|
workspace = Workspace(id=workspace_id, name="Workspace", settings=settings)
|
|
membership = WorkspaceMembership(
|
|
workspace_id=workspace_id,
|
|
user_id=user_id,
|
|
role=WorkspaceRole.MEMBER,
|
|
)
|
|
mock_identity_uow.users.get_default = AsyncMock(
|
|
return_value=User(id=user_id, display_name="Test User")
|
|
)
|
|
mock_identity_uow.workspaces.get = AsyncMock(return_value=workspace)
|
|
mock_identity_uow.workspaces.get_membership = AsyncMock(return_value=membership)
|
|
identity_servicer.set_mock_uow(mock_identity_uow)
|
|
|
|
request = noteflow_pb2.GetWorkspaceSettingsRequest(workspace_id=str(workspace_id))
|
|
response = await identity_servicer.GetWorkspaceSettings(request, mock_grpc_context)
|
|
|
|
assert response.rag_enabled is True, "rag_enabled should match workspace settings"
|
|
assert (
|
|
response.default_summarization_template == "template-123"
|
|
), "default_summarization_template should match workspace settings"
|
|
|
|
|
|
# =============================================================================
|
|
# Test: UpdateWorkspaceSettings
|
|
# =============================================================================
|
|
|
|
|
|
class TestUpdateWorkspaceSettings:
|
|
"""Tests for IdentityMixin.UpdateWorkspaceSettings."""
|
|
|
|
async def test_updates_workspace_settings(
|
|
self,
|
|
identity_servicer: MockIdentityServicerHost,
|
|
mock_identity_uow: MagicMock,
|
|
mock_grpc_context: MagicMock,
|
|
) -> None:
|
|
"""UpdateWorkspaceSettings persists updated settings."""
|
|
workspace_id = uuid4()
|
|
user_id = uuid4()
|
|
workspace, membership, updated_workspace = build_workspace_update_payload(
|
|
workspace_id,
|
|
user_id,
|
|
"template-999",
|
|
)
|
|
mock_identity_uow.users.get_default = AsyncMock(
|
|
return_value=User(id=user_id, display_name="Test User")
|
|
)
|
|
mock_identity_uow.workspaces.get = AsyncMock(return_value=workspace)
|
|
mock_identity_uow.workspaces.get_membership = AsyncMock(return_value=membership)
|
|
mock_identity_uow.workspaces.update = AsyncMock(return_value=updated_workspace)
|
|
identity_servicer.set_mock_uow(mock_identity_uow)
|
|
|
|
request = noteflow_pb2.UpdateWorkspaceSettingsRequest(
|
|
workspace_id=str(workspace_id),
|
|
settings=noteflow_pb2.WorkspaceSettingsProto(
|
|
default_summarization_template="template-999"
|
|
),
|
|
)
|
|
response = await identity_servicer.UpdateWorkspaceSettings(request, mock_grpc_context)
|
|
|
|
assert (
|
|
response.default_summarization_template == "template-999"
|
|
), "default_summarization_template should be updated"
|
|
mock_identity_uow.workspaces.update.assert_called_once()
|
|
|
|
|
|
# =============================================================================
|
|
# Test: Database Required Error
|
|
# =============================================================================
|
|
|
|
|
|
class TestDatabaseRequired:
|
|
"""Tests for database requirement handling in identity endpoints."""
|
|
|
|
async def test_list_workspaces_aborts_without_database(
|
|
self,
|
|
identity_servicer: MockIdentityServicerHost,
|
|
mock_grpc_context: MagicMock,
|
|
) -> None:
|
|
"""ListWorkspaces aborts when workspaces not supported."""
|
|
uow = MagicMock()
|
|
uow.__aenter__ = AsyncMock(return_value=uow)
|
|
uow.__aexit__ = AsyncMock(return_value=None)
|
|
uow.commit = AsyncMock()
|
|
uow.supports_users = False
|
|
uow.supports_workspaces = False
|
|
|
|
identity_servicer.set_mock_uow(uow)
|
|
|
|
request = noteflow_pb2.ListWorkspacesRequest()
|
|
|
|
# abort helpers raise AssertionError after mock context.abort()
|
|
with pytest.raises(AssertionError, match="Unreachable"):
|
|
await identity_servicer.ListWorkspaces(request, mock_grpc_context)
|
|
|
|
mock_grpc_context.abort.assert_called_once()
|
|
|
|
async def test_switch_workspace_aborts_without_database(
|
|
self,
|
|
identity_servicer: MockIdentityServicerHost,
|
|
mock_grpc_context: MagicMock,
|
|
) -> None:
|
|
"""SwitchWorkspace aborts when workspaces not supported."""
|
|
uow = MagicMock()
|
|
uow.__aenter__ = AsyncMock(return_value=uow)
|
|
uow.__aexit__ = AsyncMock(return_value=None)
|
|
uow.commit = AsyncMock()
|
|
uow.supports_users = False
|
|
uow.supports_workspaces = False
|
|
|
|
identity_servicer.set_mock_uow(uow)
|
|
|
|
workspace_id = uuid4()
|
|
request = noteflow_pb2.SwitchWorkspaceRequest(workspace_id=str(workspace_id))
|
|
|
|
# abort helpers raise AssertionError after mock context.abort()
|
|
with pytest.raises(AssertionError, match="Unreachable"):
|
|
await identity_servicer.SwitchWorkspace(request, mock_grpc_context)
|
|
|
|
mock_grpc_context.abort.assert_called_once()
|