Coverage for mcpgateway / services / personal_team_service.py: 100%

1# -*- coding: utf-8 -*- 

2"""Location: ./mcpgateway/services/personal_team_service.py 

3Copyright 2025 

4SPDX-License-Identifier: Apache-2.0 

5Authors: Mihai Criveti 

6 

7Personal Team Service. 

8This module provides automatic personal team creation and management 

9for email-based user authentication system. 

10 

11Examples: 

12 >>> from unittest.mock import Mock 

13 >>> service = PersonalTeamService(Mock()) 

14 >>> isinstance(service, PersonalTeamService) 

15 True 

16 >>> hasattr(service, 'db') 

17 True 

18""" 

19 

20# Standard 

21from typing import Optional 

22 

23# Third-Party 

24from sqlalchemy.orm import Session 

25 

26# First-Party 

27from mcpgateway.db import EmailTeam, EmailTeamMember, EmailTeamMemberHistory, EmailUser, utc_now 

28from mcpgateway.services.logging_service import LoggingService 

29 

30# Initialize logging 

31logging_service = LoggingService() 

32logger = logging_service.get_logger(__name__) 

33 

34 

35class PersonalTeamService: 

36 """Service for managing personal teams. 

37 

38 This service handles automatic creation of personal teams for users 

39 and manages team membership for personal workspaces. 

40 

41 Attributes: 

42 db (Session): SQLAlchemy database session 

43 

44 Examples: 

45 >>> from unittest.mock import Mock 

46 >>> service = PersonalTeamService(Mock()) 

47 >>> service.__class__.__name__ 

48 'PersonalTeamService' 

49 >>> hasattr(service, 'db') 

50 True 

51 """ 

52 

53 def __init__(self, db: Session): 

54 """Initialize the personal team service. 

55 

56 Args: 

57 db: SQLAlchemy database session 

58 

59 Examples: 

60 >>> from unittest.mock import Mock 

61 >>> service = PersonalTeamService(Mock()) 

62 >>> hasattr(service, 'db') and service.db is not None or service.db is None 

63 True 

64 """ 

65 self.db = db 

66 

67 async def create_personal_team(self, user: EmailUser) -> EmailTeam: 

68 """Create a personal team for a user. 

69 

70 Args: 

71 user: EmailUser instance for whom to create personal team 

72 

73 Returns: 

74 EmailTeam: The created personal team 

75 

76 Raises: 

77 ValueError: If user already has a personal team 

78 Exception: If team creation fails 

79 

80 Examples: 

81 Personal team creation is handled automatically during user registration. 

82 The team name is derived from the user's full name or email. 

83 

84 After creation, a record is inserted into EmailTeamMemberHistory to track the membership event. 

85 

86 Note: 

87 This method is async and cannot be directly called with 'await' in doctest. To test async methods, use an event loop in real tests. 

88 

89 # Example (not executable in doctest): 

90 # import asyncio 

91 # team = asyncio.run(service.create_personal_team(user)) 

92 """ 

93 try: 

94 # Check if user already has a personal team 

95 existing_team = self.db.query(EmailTeam).filter(EmailTeam.created_by == user.email, EmailTeam.is_personal.is_(True), EmailTeam.is_active.is_(True)).first() 

96 

97 if existing_team: 

98 logger.warning(f"User {user.email} already has a personal team: {existing_team.id}") 

99 raise ValueError(f"User {user.email} already has a personal team") 

100 

101 # Generate team name from user's display name 

102 display_name = user.get_display_name() 

103 team_name = f"{display_name}'s Team" 

104 

105 # Create team slug from email to ensure uniqueness 

106 email_slug = user.email.replace("@", "-").replace(".", "-").lower() 

107 team_slug = f"personal-{email_slug}" 

108 

109 # Create the personal team 

110 team = EmailTeam( 

111 name=team_name, 

112 slug=team_slug, # Will be auto-generated by event listener if not set 

113 description=f"Personal workspace for {user.email}", 

114 created_by=user.email, 

115 is_personal=True, 

116 visibility="private", 

117 is_active=True, 

118 ) 

119 

120 self.db.add(team) 

121 self.db.flush() # Get the team ID 

122 

123 # Add the user as the owner of their personal team 

124 membership = EmailTeamMember(team_id=team.id, user_email=user.email, role="owner", joined_at=utc_now(), is_active=True) 

125 

126 self.db.add(membership) 

127 self.db.flush() # Get the membership ID 

128 # Insert history record 

129 history = EmailTeamMemberHistory(team_member_id=membership.id, team_id=team.id, user_email=user.email, role="owner", action="added", action_by=user.email, action_timestamp=utc_now()) 

130 self.db.add(history) 

131 self.db.commit() 

132 

133 logger.info(f"Created personal team '{team.name}' for user {user.email}") 

134 return team 

135 

136 except Exception as e: 

137 self.db.rollback() 

138 logger.error(f"Failed to create personal team for {user.email}: {e}") 

139 raise 

140 

141 async def get_personal_team(self, user_email: str) -> Optional[EmailTeam]: 

142 """Get the personal team for a user. 

143 

144 Args: 

145 user_email: Email address of the user 

146 

147 Returns: 

148 EmailTeam: The user's personal team or None if not found 

149 

150 Examples: 

151 >>> import asyncio 

152 >>> from unittest.mock import Mock 

153 >>> service = PersonalTeamService(Mock()) 

154 >>> asyncio.iscoroutinefunction(service.get_personal_team) 

155 True 

156 """ 

157 try: 

158 team = self.db.query(EmailTeam).filter(EmailTeam.created_by == user_email, EmailTeam.is_personal.is_(True), EmailTeam.is_active.is_(True)).first() 

159 return team 

160 

161 except Exception as e: 

162 logger.error(f"Failed to get personal team for {user_email}: {e}") 

163 return None 

164 

165 async def ensure_personal_team(self, user: EmailUser) -> EmailTeam: 

166 """Ensure a user has a personal team, creating one if needed. 

167 

168 Args: 

169 user: EmailUser instance 

170 

171 Returns: 

172 EmailTeam: The user's personal team (existing or newly created) 

173 

174 Raises: 

175 Exception: If team creation or retrieval fails 

176 

177 Examples: 

178 >>> import asyncio 

179 >>> from unittest.mock import Mock 

180 >>> service = PersonalTeamService(Mock()) 

181 >>> asyncio.iscoroutinefunction(service.ensure_personal_team) 

182 True 

183 """ 

184 try: 

185 # Try to get existing personal team 

186 team = await self.get_personal_team(user.email) 

187 

188 if team is None: 

189 # Create personal team if it doesn't exist 

190 logger.info(f"Creating missing personal team for user {user.email}") 

191 team = await self.create_personal_team(user) 

192 

193 return team 

194 

195 except ValueError: 

196 # User already has a team, get it 

197 team = await self.get_personal_team(user.email) 

198 if team is None: 

199 raise Exception(f"Failed to get or create personal team for {user.email}") 

200 return team 

201 

202 def is_personal_team(self, team_id: str) -> bool: 

203 """Check if a team is a personal team. 

204 

205 Args: 

206 team_id: ID of the team to check 

207 

208 Returns: 

209 bool: True if the team is a personal team, False otherwise 

210 

211 Examples: 

212 >>> from unittest.mock import Mock 

213 >>> service = PersonalTeamService(Mock()) 

214 >>> callable(service.is_personal_team) 

215 True 

216 """ 

217 try: 

218 team = self.db.query(EmailTeam).filter(EmailTeam.id == team_id, EmailTeam.is_active.is_(True)).first() 

219 return team is not None and team.is_personal 

220 

221 except Exception as e: 

222 logger.error(f"Failed to check if team {team_id} is personal: {e}") 

223 return False 

224 

225 async def delete_personal_team(self, team_id: str) -> bool: 

226 """Delete a personal team (not allowed). 

227 

228 Personal teams cannot be deleted, only deactivated. 

229 

230 Args: 

231 team_id: ID of the team to delete 

232 

233 Returns: 

234 bool: False (personal teams cannot be deleted) 

235 

236 Raises: 

237 ValueError: Always, as personal teams cannot be deleted 

238 

239 Examples: 

240 >>> import asyncio 

241 >>> from unittest.mock import Mock 

242 >>> service = PersonalTeamService(Mock()) 

243 >>> asyncio.iscoroutinefunction(service.delete_personal_team) 

244 True 

245 """ 

246 if self.is_personal_team(team_id): 

247 raise ValueError("Personal teams cannot be deleted") 

248 return False 

249 

250 async def get_personal_team_owner(self, team_id: str) -> Optional[str]: 

251 """Get the owner email of a personal team. 

252 

253 Args: 

254 team_id: ID of the personal team 

255 

256 Returns: 

257 str: Owner email address or None if not found 

258 

259 Examples: 

260 Used for access control and team management operations. 

261 """ 

262 try: 

263 team = self.db.query(EmailTeam).filter(EmailTeam.id == team_id, EmailTeam.is_personal.is_(True), EmailTeam.is_active.is_(True)).first() 

264 return team.created_by if team else None 

265 

266 except Exception as e: 

267 logger.error(f"Failed to get personal team owner for {team_id}: {e}") 

268 return None