freemo
b5f56a6fb8
Implemented multi-user connection handling with user identity tracking (TeamMember with owner/admin/member/viewer roles), role-based access control (TeamPermission with read/write/admin/manage_members), concurrent session support (SessionRegistry with thread-safe locking and per-user/ per-project queries), and optimistic-locking conflict resolution (VersionStamp with last-writer-wins, reject, and merge strategies). Added TeamCollaborationService as the central orchestrator for all collaboration operations: team membership management, permission enforcement, session lifecycle, and version-stamp conflict detection/ resolution. The service cleans up user sessions when members are removed. Domain models follow existing patterns: Pydantic BaseModel with ConfigDict, StrEnum enums, ULID identifiers, and an error hierarchy rooted in TeamCollaborationError. Includes 48 Behave BDD scenarios (178 steps) covering all models, service operations, and edge cases, plus 15 Robot Framework integration tests for end-to-end workflow validation. ISSUES CLOSED: #863
141 lines
11 KiB
Plaintext
141 lines
11 KiB
Plaintext
*** Settings ***
|
|
Documentation Integration tests for team collaboration features.
|
|
... Verifies multi-user session management, role-based access
|
|
... control, concurrent sessions, and conflict resolution.
|
|
... Based on issue #863.
|
|
Library Process
|
|
Library OperatingSystem
|
|
Library Collections
|
|
|
|
Suite Setup Set Suite Variable ${PYTHON} python
|
|
|
|
*** Variables ***
|
|
${PYTHON} python
|
|
|
|
*** Keywords ***
|
|
Run Python Script
|
|
[Documentation] Run a Python snippet and return stdout.
|
|
[Arguments] ${script}
|
|
${result}= Run Process ${PYTHON} -c ${script}
|
|
... env:PYTHONPATH=src
|
|
RETURN ${result}
|
|
|
|
*** Test Cases ***
|
|
TeamRole Enum Contains Expected Members
|
|
[Documentation] Verify TeamRole enum has owner, admin, member, viewer.
|
|
${result}= Run Python Script
|
|
... from cleveragents.domain.models.core.team_collab import TeamRole; roles = sorted([r.value for r in TeamRole]); print(','.join(roles))
|
|
Should Be Equal As Strings ${result.rc} 0
|
|
Should Contain ${result.stdout} admin
|
|
Should Contain ${result.stdout} member
|
|
Should Contain ${result.stdout} owner
|
|
Should Contain ${result.stdout} viewer
|
|
|
|
TeamPermission Enum Contains Expected Members
|
|
[Documentation] Verify TeamPermission enum has read, write, admin, manage_members.
|
|
${result}= Run Python Script
|
|
... from cleveragents.domain.models.core.team_collab import TeamPermission; perms = sorted([p.value for p in TeamPermission]); print(','.join(perms))
|
|
Should Be Equal As Strings ${result.rc} 0
|
|
Should Contain ${result.stdout} read
|
|
Should Contain ${result.stdout} write
|
|
Should Contain ${result.stdout} admin
|
|
Should Contain ${result.stdout} manage_members
|
|
|
|
Team Role Permissions Mapping Is Correct
|
|
[Documentation] Verify role-to-permission mapping counts.
|
|
${result}= Run Python Script
|
|
... from cleveragents.domain.models.core.team_collab import TEAM_ROLE_PERMISSIONS, TeamRole; print(len(TEAM_ROLE_PERMISSIONS[TeamRole.OWNER]), len(TEAM_ROLE_PERMISSIONS[TeamRole.ADMIN]), len(TEAM_ROLE_PERMISSIONS[TeamRole.MEMBER]), len(TEAM_ROLE_PERMISSIONS[TeamRole.VIEWER]))
|
|
Should Be Equal As Strings ${result.rc} 0
|
|
Should Contain ${result.stdout} 4 4 2 1
|
|
|
|
TeamMember Model Creation And Validation
|
|
[Documentation] Create a TeamMember and verify fields.
|
|
${result}= Run Python Script
|
|
... from cleveragents.domain.models.core.team_collab import TeamMember, TeamRole; m = TeamMember(user_id='alice', display_name='Alice', email='alice@example.com', role=TeamRole.OWNER); print(f'{m.user_id},{m.role},{m.email}')
|
|
Should Be Equal As Strings ${result.rc} 0
|
|
Should Contain ${result.stdout} alice,owner,alice@example.com
|
|
|
|
TeamMember Email Validation Rejects Invalid
|
|
[Documentation] Creating a TeamMember with invalid email fails.
|
|
${result}= Run Python Script
|
|
... from cleveragents.domain.models.core.team_collab import TeamMember, TeamRole;\ntry:\n\ \ \ \ TeamMember(user_id='x', display_name='X', email='bad', role=TeamRole.MEMBER)\n\ \ \ \ print('NO_ERROR')\nexcept Exception as e:\n\ \ \ \ print('ERROR')
|
|
Should Be Equal As Strings ${result.rc} 0
|
|
Should Contain ${result.stdout} ERROR
|
|
|
|
TeamMember Permission Check
|
|
[Documentation] Verify has_permission for different roles.
|
|
${result}= Run Python Script
|
|
... from cleveragents.domain.models.core.team_collab import TeamMember, TeamRole, TeamPermission; m = TeamMember(user_id='v', display_name='V', email='v@x.com', role=TeamRole.VIEWER); print(m.has_permission(TeamPermission.READ), m.has_permission(TeamPermission.WRITE))
|
|
Should Be Equal As Strings ${result.rc} 0
|
|
Should Contain ${result.stdout} True False
|
|
|
|
UserSession Creation And Deactivation
|
|
[Documentation] Create a session, verify active, then deactivate.
|
|
${result}= Run Python Script
|
|
... from cleveragents.domain.models.core.team_collab import UserSession; s = UserSession(session_id='s1', user_id='alice', display_name='Alice', project_id='p1'); print(s.is_active); s.deactivate(); print(s.is_active)
|
|
Should Be Equal As Strings ${result.rc} 0
|
|
Should Contain ${result.stdout} True
|
|
Should Contain ${result.stdout} False
|
|
|
|
VersionStamp Creation And Increment
|
|
[Documentation] Create a version stamp and increment it.
|
|
${result}= Run Python Script
|
|
... from cleveragents.domain.models.core.team_collab import VersionStamp; v = VersionStamp(resource_id='r1', updated_by='alice'); v2 = v.increment('bob'); print(f'{v.version},{v2.version},{v2.updated_by}')
|
|
Should Be Equal As Strings ${result.rc} 0
|
|
Should Contain ${result.stdout} 1,2,bob
|
|
|
|
SessionRegistry Concurrent Session Management
|
|
[Documentation] Register multiple sessions and query by user/project.
|
|
${result}= Run Python Script
|
|
... from cleveragents.domain.models.core.team_collab import SessionRegistry, UserSession; r = SessionRegistry(); r.register(UserSession(session_id='s1', user_id='alice', display_name='Alice', project_id='p1')); r.register(UserSession(session_id='s2', user_id='bob', display_name='Bob', project_id='p1')); r.register(UserSession(session_id='s3', user_id='alice', display_name='Alice', project_id='p2')); print(f'active={r.active_count},alice={len(r.get_sessions_for_user("alice"))},p1={len(r.get_sessions_for_project("p1"))}')
|
|
Should Be Equal As Strings ${result.rc} 0
|
|
Should Contain ${result.stdout} active=3,alice=2,p1=2
|
|
|
|
TeamCollaborationService End To End Workflow
|
|
[Documentation] Full workflow: add members, register sessions, check perms, handle conflicts.
|
|
${result}= Run Python Script
|
|
... from cleveragents.application.services.team_collab_service import TeamCollaborationService; from cleveragents.domain.models.core.team_collab import TeamRole, TeamPermission, ConflictResolutionStrategy, VersionConflictError; svc = TeamCollaborationService(); svc.add_member('alice', 'Alice', 'alice@x.com', TeamRole.OWNER); svc.add_member('bob', 'Bob', 'bob@x.com', TeamRole.MEMBER); svc.add_member('carol', 'Carol', 'carol@x.com', TeamRole.VIEWER); s1 = svc.register_session('alice', 'p1'); s2 = svc.register_session('bob', 'p1'); sessions = svc.get_active_sessions_for_project('p1'); print(f'sessions={len(sessions)}'); print(f'alice_write={svc.check_permission("alice", TeamPermission.WRITE)}'); print(f'carol_write={svc.check_permission("carol", TeamPermission.WRITE)}'); svc.create_version_stamp('r1', 'alice'); svc.update_version('r1', 1, 'bob'); conflict = svc.detect_conflict('r1', 1); print(f'conflict={conflict}'); result = svc.update_version('r1', 1, 'carol', ConflictResolutionStrategy.LAST_WRITER_WINS); print(f'resolved={result.resolved}'); stamp = svc.get_version_stamp('r1'); print(f'version={stamp.version}')
|
|
Should Be Equal As Strings ${result.rc} 0
|
|
Should Contain ${result.stdout} sessions=2
|
|
Should Contain ${result.stdout} alice_write=True
|
|
Should Contain ${result.stdout} carol_write=False
|
|
Should Contain ${result.stdout} conflict=True
|
|
Should Contain ${result.stdout} resolved=True
|
|
Should Contain ${result.stdout} version=3
|
|
|
|
VersionConflictError With Reject Strategy
|
|
[Documentation] Verify reject strategy raises VersionConflictError on conflict.
|
|
${result}= Run Python Script
|
|
... from cleveragents.application.services.team_collab_service import TeamCollaborationService; from cleveragents.domain.models.core.team_collab import TeamRole, ConflictResolutionStrategy, VersionConflictError; svc = TeamCollaborationService(); svc.add_member('alice', 'Alice', 'alice@x.com', TeamRole.OWNER); svc.create_version_stamp('r1', 'alice'); svc.update_version('r1', 1, 'alice');\ntry:\n\ \ \ \ svc.update_version('r1', 1, 'alice', ConflictResolutionStrategy.REJECT)\n\ \ \ \ print('NO_ERROR')\nexcept VersionConflictError:\n\ \ \ \ print('CONFLICT_ERROR')
|
|
Should Be Equal As Strings ${result.rc} 0
|
|
Should Contain ${result.stdout} CONFLICT_ERROR
|
|
|
|
Error Hierarchy Verification
|
|
[Documentation] Verify error class hierarchy.
|
|
${result}= Run Python Script
|
|
... from cleveragents.domain.models.core.team_collab import TeamCollaborationError, TeamMemberNotFoundError, TeamMemberExistsError, VersionConflictError; print(issubclass(TeamMemberNotFoundError, TeamCollaborationError), issubclass(TeamMemberExistsError, TeamCollaborationError), issubclass(VersionConflictError, TeamCollaborationError))
|
|
Should Be Equal As Strings ${result.rc} 0
|
|
Should Contain ${result.stdout} True True True
|
|
|
|
Remove Member Cleans Up Sessions
|
|
[Documentation] Verify removing a member cleans up their sessions.
|
|
${result}= Run Python Script
|
|
... from cleveragents.application.services.team_collab_service import TeamCollaborationService; from cleveragents.domain.models.core.team_collab import TeamRole; svc = TeamCollaborationService(); svc.add_member('alice', 'Alice', 'alice@x.com', TeamRole.OWNER); svc.register_session('alice', 'p1'); print(f'before={len(svc.get_active_sessions_for_project("p1"))}'); svc.remove_member('alice'); print(f'after={len(svc.get_active_sessions_for_project("p1"))}')
|
|
Should Be Equal As Strings ${result.rc} 0
|
|
Should Contain ${result.stdout} before=1
|
|
Should Contain ${result.stdout} after=0
|
|
|
|
Enforce Permission Raises PermissionError
|
|
[Documentation] Verify enforce_permission raises PermissionError for insufficient role.
|
|
${result}= Run Python Script
|
|
... from cleveragents.application.services.team_collab_service import TeamCollaborationService; from cleveragents.domain.models.core.team_collab import TeamRole, TeamPermission;\nsvc = TeamCollaborationService()\nsvc.add_member('viewer', 'Viewer', 'v@x.com', TeamRole.VIEWER)\ntry:\n\ \ \ \ svc.enforce_permission('viewer', TeamPermission.WRITE)\n\ \ \ \ print('NO_ERROR')\nexcept PermissionError:\n\ \ \ \ print('PERM_ERROR')
|
|
Should Be Equal As Strings ${result.rc} 0
|
|
Should Contain ${result.stdout} PERM_ERROR
|
|
|
|
Merge Strategy Resolves Conflict
|
|
[Documentation] Verify merge strategy resolves version conflicts.
|
|
${result}= Run Python Script
|
|
... from cleveragents.application.services.team_collab_service import TeamCollaborationService; from cleveragents.domain.models.core.team_collab import TeamRole, ConflictResolutionStrategy; svc = TeamCollaborationService(); svc.add_member('alice', 'Alice', 'alice@x.com', TeamRole.OWNER); svc.create_version_stamp('r1', 'alice'); svc.update_version('r1', 1, 'alice'); result = svc.update_version('r1', 1, 'alice', ConflictResolutionStrategy.MERGE); print(f'detected={result.detected},resolved={result.resolved},strategy={result.strategy}')
|
|
Should Be Equal As Strings ${result.rc} 0
|
|
Should Contain ${result.stdout} detected=True,resolved=True,strategy=merge
|