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.config import settings 

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

29from mcpgateway.services.logging_service import LoggingService 

30from mcpgateway.utils.create_slug import slugify 

31 

32# Initialize logging 

33logging_service = LoggingService() 

34logger = logging_service.get_logger(__name__) 

35 

36 

37class PersonalTeamService: 

38 """Service for managing personal teams. 

39 

40 This service handles automatic creation of personal teams for users 

41 and manages team membership for personal workspaces. 

42 

43 Attributes: 

44 db (Session): SQLAlchemy database session 

45 

46 Examples: 

47 >>> from unittest.mock import Mock 

48 >>> service = PersonalTeamService(Mock()) 

49 >>> service.__class__.__name__ 

50 'PersonalTeamService' 

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

52 True 

53 """ 

54 

55 def __init__(self, db: Session): 

56 """Initialize the personal team service. 

57 

58 Args: 

59 db: SQLAlchemy database session 

60 

61 Examples: 

62 >>> from unittest.mock import Mock 

63 >>> service = PersonalTeamService(Mock()) 

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

65 True 

66 """ 

67 self.db = db 

68 

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

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

71 

72 Args: 

73 user: EmailUser instance for whom to create personal team 

74 

75 Returns: 

76 EmailTeam: The created personal team 

77 

78 Raises: 

79 ValueError: If user already has a personal team 

80 Exception: If team creation fails 

81 

82 Examples: 

83 Personal team creation is handled automatically during user registration. 

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

85 

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

87 

88 Note: 

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

90 

91 # Example (not executable in doctest): 

92 # import asyncio 

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

94 """ 

95 try: 

96 # Check if user already has a personal team 

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

98 

99 if existing_team: 

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

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

102 

103 # Generate team name from user's display name 

104 display_name = user.get_display_name() 

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

106 

107 # Create team slug — use prefix from config if set, otherwise derive from display name 

108 prefix = slugify(settings.personal_team_prefix.strip()) if settings.personal_team_prefix.strip() else "" 

109 if prefix: 

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

111 team_slug = f"{prefix}-{email_slug}" 

112 else: 

113 team_slug = slugify(team_name) 

114 

115 # Create the personal team 

116 team = EmailTeam( 

117 name=team_name, 

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

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

120 created_by=user.email, 

121 is_personal=True, 

122 visibility="private", 

123 is_active=True, 

124 ) 

125 

126 self.db.add(team) 

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

128 

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

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

131 

132 self.db.add(membership) 

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

134 # Insert history record 

135 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()) 

136 self.db.add(history) 

137 self.db.commit() 

138 

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

140 return team 

141 

142 except Exception as e: 

143 self.db.rollback() 

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

145 raise 

146 

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

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

149 

150 Args: 

151 user_email: Email address of the user 

152 

153 Returns: 

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

155 

156 Examples: 

157 >>> import asyncio 

158 >>> from unittest.mock import Mock 

159 >>> service = PersonalTeamService(Mock()) 

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

161 True 

162 """ 

163 try: 

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

165 return team 

166 

167 except Exception as e: 

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

169 return None 

170 

171 async def ensure_personal_team(self, user: EmailUser) -> Optional[EmailTeam]: 

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

173 

174 Args: 

175 user: EmailUser instance 

176 

177 Returns: 

178 EmailTeam: The user's personal team (existing or newly created), or None if auto-creation is disabled 

179 

180 Raises: 

181 Exception: If team creation or retrieval fails 

182 

183 Examples: 

184 >>> import asyncio 

185 >>> from unittest.mock import Mock 

186 >>> service = PersonalTeamService(Mock()) 

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

188 True 

189 """ 

190 try: 

191 # Try to get existing personal team 

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

193 

194 if team is None: 

195 if not getattr(settings, "auto_create_personal_teams", True): 

196 return None 

197 # Create personal team if it doesn't exist 

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

199 team = await self.create_personal_team(user) 

200 

201 return team 

202 

203 except ValueError: 

204 # User already has a team, get it 

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

206 if team is None: 

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

208 return team 

209 

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

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

212 

213 Args: 

214 team_id: ID of the team to check 

215 

216 Returns: 

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

218 

219 Examples: 

220 >>> from unittest.mock import Mock 

221 >>> service = PersonalTeamService(Mock()) 

222 >>> callable(service.is_personal_team) 

223 True 

224 """ 

225 try: 

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

227 return team is not None and team.is_personal 

228 

229 except Exception as e: 

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

231 return False 

232 

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

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

235 

236 Personal teams cannot be deleted, only deactivated. 

237 

238 Args: 

239 team_id: ID of the team to delete 

240 

241 Returns: 

242 bool: False (personal teams cannot be deleted) 

243 

244 Raises: 

245 ValueError: Always, as personal teams cannot be deleted 

246 

247 Examples: 

248 >>> import asyncio 

249 >>> from unittest.mock import Mock 

250 >>> service = PersonalTeamService(Mock()) 

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

252 True 

253 """ 

254 if self.is_personal_team(team_id): 

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

256 return False 

257 

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

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

260 

261 Args: 

262 team_id: ID of the personal team 

263 

264 Returns: 

265 str: Owner email address or None if not found 

266 

267 Examples: 

268 Used for access control and team management operations. 

269 """ 

270 try: 

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

272 return team.created_by if team else None 

273 

274 except Exception as e: 

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

276 return None