Coverage for mcpgateway/routers/teams.py: 100%

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

2"""Location: ./mcpgateway/routers/teams.py

4SPDX-License-Identifier: Apache-2.0

5Authors: Mihai Criveti

7Team Management Router.

8This module provides FastAPI routes for team management including

9team creation, member management, and invitation handling.

11Examples:

12 >>> from fastapi import FastAPI

13 >>> from mcpgateway.routers.teams import teams_router

14 >>> app = FastAPI()

15 >>> app.include_router(teams_router, prefix="/teams", tags=["Teams"])

16 >>> isinstance(teams_router, APIRouter)

17 True

18 >>> len(teams_router.routes) > 10 # Multiple team management endpoints

19 True

20"""

22# Standard

23from typing import Any, cast, List, Optional, Union

25# Third-Party

26from fastapi import APIRouter, Depends, HTTPException, Query, status

27from sqlalchemy.orm import Session

29# First-Party

30from mcpgateway.common.validators import SecurityValidator

31from mcpgateway.config import settings

32from mcpgateway.db import get_db

33from mcpgateway.middleware.rbac import _ACCESS_DENIED_MSG, get_current_user_with_permissions, require_permission

34from mcpgateway.schemas import (

35 CursorPaginatedTeamsResponse,

36 PaginatedTeamMembersResponse,

37 SuccessResponse,

38 TeamCreateRequest,

39 TeamDiscoveryResponse,

40 TeamInvitationResponse,

41 TeamInviteRequest,

42 TeamJoinRequest,

43 TeamJoinRequestResponse,

44 TeamListResponse,

45 TeamMemberAddRequest,

46 TeamMemberResponse,

47 TeamMemberUpdateRequest,

48 TeamResponse,

49 TeamUpdateRequest,

50)

51from mcpgateway.services.logging_service import LoggingService

52from mcpgateway.services.team_invitation_service import TeamInvitationService

53from mcpgateway.services.team_management_service import (

54 InvalidRoleError,

55 MemberAlreadyExistsError,

56 TeamManagementError,

57 TeamManagementService,

58 TeamMemberAddError,

59 TeamMemberLimitExceededError,

60 TeamNotFoundError,

61 UserNotFoundError,

62)

64# Initialize logging

65logging_service = LoggingService()

66logger = logging_service.get_logger(__name__)

68# Create router

69teams_router = APIRouter()

72# ---------------------------------------------------------------------------

73# Team CRUD Operations

74# ---------------------------------------------------------------------------

77@teams_router.post("/", response_model=TeamResponse, status_code=status.HTTP_201_CREATED)

78@require_permission("teams.create")

79async def create_team(request: TeamCreateRequest, current_user_ctx: dict = Depends(get_current_user_with_permissions), db: Session = Depends(get_db)) -> TeamResponse:

80 """Create a new team.

82 Args:

83 request: Team creation request data

84 current_user_ctx: Currently authenticated user context

85 db: Database session

87 Returns:

88 TeamResponse: Created team data

90 Raises:

91 HTTPException: If team creation fails

93 Examples:

94 >>> import asyncio

95 >>> asyncio.iscoroutinefunction(create_team)

96 True

97 """

98 try:

99 is_admin = bool(current_user_ctx.get("is_admin"))

100

101 if not settings.allow_team_creation and not is_admin:

102 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Team creation is currently disabled")

103

104 service = TeamManagementService(db)

105 team = await service.create_team(

106 name=request.name,

107 description=request.description,

108 created_by=current_user_ctx["email"],

109 visibility=request.visibility,

110 max_members=request.max_members,

111 skip_limits=is_admin,

112 )

113

114 # Build response BEFORE closing session to avoid lazy-load issues with get_member_count()

115 response = TeamResponse(

116 id=team.id,

117 name=team.name,

118 slug=team.slug,

119 description=team.description,

120 created_by=team.created_by,

121 is_personal=team.is_personal,

122 visibility=team.visibility,

123 max_members=team.max_members,

124 member_count=team.get_member_count(),

125 created_at=team.created_at,

126 updated_at=team.updated_at,

127 is_active=team.is_active,

128 )

129 db.commit()

130 db.close()

131 return response

132 except HTTPException:

133 raise

134 except ValueError as e:

135 logger.error(f"Team creation failed: {e}")

136 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))

137 except Exception as e:

138 logger.error(f"Unexpected error creating team: {e}")

139 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to create team")

140

141

142@teams_router.get("/", response_model=Union[TeamListResponse, CursorPaginatedTeamsResponse])

143@require_permission("teams.read")

144async def list_teams(

145 skip: int = Query(0, ge=0, description="Number of teams to skip"),

146 limit: int = Query(50, ge=1, le=settings.pagination_max_page_size, description="Number of teams to return"),

147 cursor: Optional[str] = Query(None, description="Pagination cursor"),

148 include_pagination: bool = Query(False, description="Include pagination metadata (cursor)"),

149 current_user_ctx: dict = Depends(get_current_user_with_permissions),

150 db: Session = Depends(get_db),

151) -> Union[TeamListResponse, CursorPaginatedTeamsResponse]:

152 """List teams visible to the caller.

153

154 - Administrators see all non-personal teams (paginated)

155 - Regular users see only teams they are a member of (paginated client-side)

156

157 Args:

158 skip: Number of teams to skip for pagination

159 limit: Maximum number of teams to return

160 cursor: Pagination cursor

161 include_pagination: Include pagination metadata

162 current_user_ctx: Current user context with permissions and database session

163 db: Database session

164

165 Returns:

166 Union[TeamListResponse, CursorPaginatedTeamsResponse]: List of teams

167

168 Raises:

169 HTTPException: If there's an error listing teams

170 """

171 try:

172 service = TeamManagementService(db)

173

174 teams_data = []

175 next_cursor = None

176 total = 0

177

178 if current_user_ctx.get("is_admin"):

179 # Use updated list_teams logic

180 # If current request uses offset (skip), mapped to offset.

181 # If cursor, mapped to cursor.

182 # page is None, so returns Tuple

183 result = await service.list_teams(

184 limit=limit,

185 offset=skip,

186 cursor=cursor,

187 )

188 # Result is tuple (list, next_cursor)

189 teams_data, next_cursor = result

190

191 # Get accurate total count for API consumers

192 total = await service.get_teams_count()

193 else:

194 # Fallback to user teams and apply pagination locally

195 user_teams = await service.get_user_teams(current_user_ctx["email"], include_personal=True)

196 total = len(user_teams)

197 teams_data = user_teams[skip : skip + limit]

198

199 # Batch fetch member counts with caching (N+1 elimination)

200 team_ids = [str(team.id) for team in teams_data]

201 member_counts = await service.get_member_counts_batch_cached(team_ids)

202

203 team_responses = [

204 TeamResponse(

205 id=team.id,

206 name=team.name,

207 slug=team.slug,

208 description=team.description,

209 created_by=team.created_by,

210 is_personal=team.is_personal,

211 visibility=team.visibility,

212 max_members=team.max_members,

213 member_count=member_counts.get(str(team.id), 0),

214 created_at=team.created_at,

215 updated_at=team.updated_at,

216 is_active=team.is_active,

217 )

218 for team in teams_data

219 ]

220

221 # Release transaction before response serialization

222 db.commit()

223 db.close()

224

225 if include_pagination:

226 return CursorPaginatedTeamsResponse(teams=team_responses, nextCursor=next_cursor)

227

228 return TeamListResponse(teams=team_responses, total=total)

229 except Exception as e:

230 logger.error(f"Error listing teams: {e}")

231 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to list teams")

232

233

234@teams_router.get("/discover", response_model=List[TeamDiscoveryResponse])

235@require_permission("teams.read")

236async def discover_public_teams(

237 skip: int = Query(0, ge=0, description="Number of teams to skip"),

238 limit: int = Query(50, ge=1, le=settings.pagination_max_page_size, description="Number of teams to return"),

239 current_user_ctx: dict = Depends(get_current_user_with_permissions),

240 db: Session = Depends(get_db),

241) -> List[TeamDiscoveryResponse]:

242 """Discover public teams that can be joined.

243

244 Returns public teams that are discoverable to all authenticated users.

245 Only shows teams where the current user is not already a member.

246

247 Args:

248 skip: Number of teams to skip for pagination

249 limit: Maximum number of teams to return

250 current_user_ctx: Current user context with permissions and database session

251 db: Database session

252

253 Returns:

254 List[TeamDiscoveryResponse]: List of discoverable public teams

255

256 Raises:

257 HTTPException: If there's an error discovering teams

258 """

259 try:

260 team_service = TeamManagementService(db)

261

262 # Get public teams where user is not already a member

263 public_teams = await team_service.discover_public_teams(current_user_ctx["email"], skip=skip, limit=limit)

264

265 # Batch fetch member counts with caching (N+1 elimination)

266 team_ids = [str(team.id) for team in public_teams]

267 member_counts = await team_service.get_member_counts_batch_cached(team_ids)

268

269 discovery_responses = []

270 for team in public_teams:

271 discovery_responses.append(

272 TeamDiscoveryResponse(

273 id=team.id,

274 name=team.name,

275 description=team.description,

276 member_count=member_counts.get(str(team.id), 0),

277 created_at=team.created_at,

278 is_joinable=True, # All returned teams are joinable

279 )

280 )

281

282 # Release transaction before response serialization

283 db.commit()

284 db.close()

285

286 return discovery_responses

287 except Exception as e:

288 logger.error(f"Error discovering public teams: {e}")

289 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to discover teams")

290

291

292@teams_router.get("/{team_id}", response_model=TeamResponse)

293@require_permission("teams.read")

294async def get_team(team_id: str, current_user: dict = Depends(get_current_user_with_permissions), db: Session = Depends(get_db)) -> TeamResponse:

295 """Get a specific team by ID.

296

297 Args:

298 team_id: Team UUID

299 current_user: Authenticated user context dict with email and permissions

300 db: Database session

301

302 Returns:

303 TeamResponse: Team data

304

305 Raises:

306 HTTPException: If team not found or access denied

307 """

308 try:

309 service = TeamManagementService(db)

310 team = await service.get_team_by_id(team_id)

311

312 if not team:

313 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Team not found")

314

315 # Check if user has access to the team

316 user_role = await service.get_user_role_in_team(current_user["email"], team_id)

317 if not user_role:

318 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=_ACCESS_DENIED_MSG)

319

320 team_obj = cast(Any, team)

321 # Build response BEFORE closing session to avoid lazy-load issues with get_member_count()

322 response = TeamResponse(

323 id=team_obj.id,

324 name=team_obj.name,

325 slug=team_obj.slug,

326 description=team_obj.description,

327 created_by=team_obj.created_by,

328 is_personal=team_obj.is_personal,

329 visibility=team_obj.visibility,

330 max_members=team_obj.max_members,

331 member_count=team_obj.get_member_count(),

332 created_at=team_obj.created_at,

333 updated_at=team_obj.updated_at,

334 is_active=team_obj.is_active,

335 )

336 db.commit()

337 db.close()

338 return response

339 except HTTPException:

340 raise

341 except Exception as e:

342 logger.error(f"Error getting team {SecurityValidator.sanitize_log_message(team_id)}: {e}")

343 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to get team")

344

345

346@teams_router.put("/{team_id}", response_model=TeamResponse)

347@require_permission("teams.update")

348async def update_team(team_id: str, request: TeamUpdateRequest, current_user: dict = Depends(get_current_user_with_permissions), db: Session = Depends(get_db)) -> TeamResponse:

349 """Update a team.

350

351 Args:

352 team_id: Team UUID

353 request: Team update request data

354 current_user: Authenticated user context dict with email and permissions

355 db: Database session

356

357 Returns:

358 TeamResponse: Updated team data

359

360 Raises:

361 HTTPException: If team not found, access denied, or update fails

362 """

363 try:

364 is_admin = bool(current_user.get("is_admin"))

365 service = TeamManagementService(db)

366

367 # Check if user is team owner

368 role = await service.get_user_role_in_team(current_user["email"], team_id)

369 if role != "owner":

370 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=_ACCESS_DENIED_MSG)

371

372 # Only pass max_members when explicitly provided in the request body

373 # (including explicit null) so update_team can distinguish "not provided"

374 # from "clear the per-team override".

375 update_kwargs: dict[str, Any] = dict(team_id=team_id, name=request.name, description=request.description, visibility=request.visibility, skip_limits=is_admin)

376 if "max_members" in request.model_fields_set:

377 update_kwargs["max_members"] = request.max_members

378 success = await service.update_team(**update_kwargs)

379

380 if not success:

381 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Team not found or update failed")

382

383 # Fetch the updated team to build the response

384 team = await service.get_team_by_id(team_id)

385 if not team:

386 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Team not found after update")

387

388 team_obj = cast(Any, team)

389 # Build response BEFORE closing session to avoid lazy-load issues with get_member_count()

390 response = TeamResponse(

391 id=team_obj.id,

392 name=team_obj.name,

393 slug=team_obj.slug,

394 description=team_obj.description,

395 created_by=team_obj.created_by,

396 is_personal=team_obj.is_personal,

397 visibility=team_obj.visibility,

398 max_members=team_obj.max_members,

399 member_count=team_obj.get_member_count(),

400 created_at=team_obj.created_at,

401 updated_at=team_obj.updated_at,

402 is_active=team_obj.is_active,

403 )

404 db.commit()

405 db.close()

406 return response

407 except HTTPException:

408 raise

409 except ValueError as e:

410 logger.error(f"Team update failed: {e}")

411 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))

412 except Exception as e:

413 logger.error(f"Error updating team {SecurityValidator.sanitize_log_message(team_id)}: {e}")

414 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to update team")

415

416

417@teams_router.delete("/{team_id}", response_model=SuccessResponse)

418@require_permission("teams.delete")

419async def delete_team(team_id: str, current_user: dict = Depends(get_current_user_with_permissions), db: Session = Depends(get_db)) -> SuccessResponse:

420 """Delete a team.

421

422 Args:

423 team_id: Team UUID

424 current_user: Authenticated user context dict with email and permissions

425 db: Database session

426

427 Returns:

428 SuccessResponse: Success confirmation

429

430 Raises:

431 HTTPException: If team not found, access denied, or deletion fails

432 """

433 try:

434 service = TeamManagementService(db)

435

436 # Check if user is team owner

437 role = await service.get_user_role_in_team(current_user["email"], team_id)

438 if role != "owner":

439 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Only team owners can delete teams")

440

441 success = await service.delete_team(team_id, current_user["email"])

442 if not success:

443 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Team not found")

444

445 db.commit()

446 db.close()

447 return SuccessResponse(message="Team deleted successfully")

448 except HTTPException:

449 raise

450 except Exception as e:

451 logger.error(f"Error deleting team {SecurityValidator.sanitize_log_message(team_id)}: {e}")

452 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to delete team")

453

454

455# ---------------------------------------------------------------------------

456# Team Member Management

457# ---------------------------------------------------------------------------

458

459

460@teams_router.get("/{team_id}/members", response_model=Union[PaginatedTeamMembersResponse, List[TeamMemberResponse]])

461@require_permission("teams.read")

462async def list_team_members(

463 team_id: str,

464 cursor: Optional[str] = Query(None, description="Cursor for pagination"),

465 limit: Optional[int] = Query(None, ge=1, le=settings.pagination_max_page_size, description="Maximum number of members to return (default: 50)"),

466 include_pagination: bool = Query(False, description="Include cursor pagination metadata in response"),

467 current_user: dict = Depends(get_current_user_with_permissions),

468 db: Session = Depends(get_db),

469) -> Union[PaginatedTeamMembersResponse, List[TeamMemberResponse]]:

470 """List team members with cursor-based pagination.

471

472 Args:

473 team_id: Team UUID

474 cursor: Pagination cursor for fetching the next set of results

475 limit: Maximum number of members to return (default: 50)

476 include_pagination: Whether to include cursor pagination metadata in the response (default: false)

477 current_user: Authenticated user context dict with email and permissions

478 db: Database session

479

480 Returns:

481 PaginatedTeamMembersResponse with members and nextCursor if include_pagination=true, or

482 List of team members if include_pagination=false

483

484 Raises:

485 HTTPException: If team not found or access denied

486 """

487 try:

488 service = TeamManagementService(db)

489

490 # Check if user has access to the team

491 user_role = await service.get_user_role_in_team(current_user["email"], team_id)

492 if not user_role:

493 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=_ACCESS_DENIED_MSG)

494

495 # Get members - service returns different types based on parameters:

496 # - cursor=None, limit=None: List[Tuple] (backward compat)

497 # - cursor or limit provided: Tuple[List[Tuple], next_cursor]

498 result = await service.get_team_members(team_id, cursor=cursor, limit=limit)

499

500 # Handle different return types from service

501 if cursor is not None or limit is not None:

502 # Cursor pagination was used - result is a tuple

503 members, next_cursor = result

504 else:

505 # No pagination - result is a plain list

506 members = result

507 next_cursor = None

508

509 # Convert to response objects

510 member_responses = []

511 for user, membership in members:

512 member_responses.append(

513 TeamMemberResponse(

514 id=membership.id,

515 team_id=membership.team_id,

516 user_email=membership.user_email,

517 role=membership.role,

518 joined_at=membership.joined_at,

519 invited_by=membership.invited_by,

520 is_active=membership.is_active,

521 )

522 )

523

524 # Return with pagination metadata if requested

525 db.commit()

526 db.close()

527 if include_pagination:

528 return PaginatedTeamMembersResponse(members=member_responses, nextCursor=next_cursor)

529

530 return member_responses

531 except HTTPException:

532 raise

533 except Exception as e:

534 logger.error(f"Error listing team members for team {SecurityValidator.sanitize_log_message(team_id)}: {e}")

535 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to list team members")

536

537

538@teams_router.post("/{team_id}/members", response_model=TeamMemberResponse, status_code=status.HTTP_201_CREATED)

539@require_permission("teams.manage_members")

540async def add_team_member(team_id: str, request: TeamMemberAddRequest, current_user: dict = Depends(get_current_user_with_permissions), db: Session = Depends(get_db)) -> TeamMemberResponse:

541 """Add a new member to a team.

542

543 Args:

544 team_id: Team UUID

545 request: Member add request data with email and role

546 current_user: Authenticated user context dict with email and permissions

547 db: Database session

548

549 Returns:

550 TeamMemberResponse: New member data

551

552 Raises:

553 HTTPException: If team not found, access denied, or add fails

554 """

555 try:

556 service = TeamManagementService(db)

557

558 # Check if user is team owner

559 role = await service.get_user_role_in_team(current_user["email"], team_id)

560 if role != "owner":

561 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=_ACCESS_DENIED_MSG)

562

563 # Add member to team and get the created member directly

564 member = await service.add_member_to_team(team_id, request.email, request.role, invited_by=current_user["email"])

565

566 db.commit()

567 db.close()

568 return TeamMemberResponse.model_validate(member)

569 except HTTPException:

570 raise

571 except InvalidRoleError as e:

572 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))

573 except TeamNotFoundError as e:

574 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))

575 except UserNotFoundError as e:

576 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))

577 except MemberAlreadyExistsError as e:

578 raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e))

579 except TeamMemberLimitExceededError as e:

580 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))

581 except TeamMemberAddError as e:

582 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))

583 except TeamManagementError as e:

584 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))

585 except Exception as e:

586 logger.error(f"Error adding team member {SecurityValidator.sanitize_log_message(request.email)} to team {SecurityValidator.sanitize_log_message(team_id)}: {e}")

587 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to add team member")

588

589

590@teams_router.put("/{team_id}/members/{user_email}", response_model=TeamMemberResponse)

591@require_permission("teams.manage_members")

592async def update_team_member(

593 team_id: str, user_email: str, request: TeamMemberUpdateRequest, current_user: dict = Depends(get_current_user_with_permissions), db: Session = Depends(get_db)

594) -> TeamMemberResponse:

595 """Update a team member's role.

596

597 Args:

598 team_id: Team UUID

599 user_email: Email of the member to update

600 request: Member update request data

601 current_user: Authenticated user context dict with email and permissions

602 db: Database session

603

604 Returns:

605 TeamMemberResponse: Updated member data

606

607 Raises:

608 HTTPException: If member not found, access denied, or update fails

609 """

610 try:

611 service = TeamManagementService(db)

612

613 # Check if user is team owner

614 role = await service.get_user_role_in_team(current_user["email"], team_id)

615 if role != "owner":

616 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=_ACCESS_DENIED_MSG)

617

618 success = await service.update_member_role(team_id, user_email, request.role)

619 if not success:

620 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Team member not found or update failed")

621

622 # Fetch the updated member to build the response

623 member = await service.get_member(team_id, user_email)

624 if not member:

625 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Team member not found after update")

626

627 db.commit()

628 db.close()

629 return TeamMemberResponse.model_validate(member)

630 except HTTPException:

631 raise

632 except ValueError as e:

633 logger.error(f"Member update failed: {e}")

634 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))

635 except Exception as e:

636 logger.error(f"Error updating team member {SecurityValidator.sanitize_log_message(user_email)} in team {SecurityValidator.sanitize_log_message(team_id)}: {e}")

637 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to update team member")

638

639

640@teams_router.delete("/{team_id}/members/{user_email}", response_model=SuccessResponse)

641@require_permission("teams.manage_members")

642async def remove_team_member(team_id: str, user_email: str, current_user: dict = Depends(get_current_user_with_permissions), db: Session = Depends(get_db)) -> SuccessResponse:

643 """Remove a team member.

644

645 Args:

646 team_id: Team UUID

647 user_email: Email of the member to remove

648 current_user: Authenticated user context dict with email and permissions

649 db: Database session

650

651 Returns:

652 SuccessResponse: Success confirmation

653

654 Raises:

655 HTTPException: If member not found, access denied, or removal fails

656 """

657 try:

658 service = TeamManagementService(db)

659

660 # Users can remove themselves, or owners can remove others

661 current_user_role = await service.get_user_role_in_team(current_user["email"], team_id)

662 if current_user["email"] != user_email and current_user_role != "owner":

663 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=_ACCESS_DENIED_MSG)

664

665 success = await service.remove_member_from_team(team_id, user_email)

666 if not success:

667 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Team member not found")

668

669 db.commit()

670 db.close()

671 return SuccessResponse(message="Team member removed successfully")

672 except HTTPException:

673 raise

674 except Exception as e:

675 logger.error(f"Error removing team member {SecurityValidator.sanitize_log_message(user_email)} from team {SecurityValidator.sanitize_log_message(team_id)}: {e}")

676 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to remove team member")

677

678

679# ---------------------------------------------------------------------------

680# Team Invitations

681# ---------------------------------------------------------------------------

682

683

684@teams_router.post("/{team_id}/invitations", response_model=TeamInvitationResponse, status_code=status.HTTP_201_CREATED)

685@require_permission("teams.manage_members")

686async def invite_team_member(team_id: str, request: TeamInviteRequest, current_user: dict = Depends(get_current_user_with_permissions), db: Session = Depends(get_db)) -> TeamInvitationResponse:

687 """Invite a user to join a team.

688

689 Args:

690 team_id: Team UUID

691 request: Invitation request data

692 current_user: Authenticated user context dict with email and permissions

693 db: Database session

694

695 Returns:

696 TeamInvitationResponse: Created invitation data

697

698 Raises:

699 HTTPException: If team not found, access denied, or invitation fails

700 """

701 try:

702 if not settings.allow_team_invitations:

703 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Team invitations are currently disabled")

704

705 team_service = TeamManagementService(db)

706 invitation_service = TeamInvitationService(db)

707

708 # Check if user is team owner

709 role = await team_service.get_user_role_in_team(current_user["email"], team_id)

710 if role != "owner":

711 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=_ACCESS_DENIED_MSG)

712

713 invitation = await invitation_service.create_invitation(team_id=team_id, email=str(request.email), role=request.role, invited_by=current_user["email"])

714 if not invitation:

715 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to create invitation")

716

717 # Get team name for response

718 team = await team_service.get_team_by_id(team_id)

719 team_name = team.name if team else "Unknown Team"

720

721 db.commit()

722 db.close()

723 return TeamInvitationResponse(

724 id=invitation.id,

725 team_id=invitation.team_id,

726 team_name=team_name,

727 email=invitation.email,

728 role=invitation.role,

729 invited_by=invitation.invited_by,

730 invited_at=invitation.invited_at,

731 expires_at=invitation.expires_at,

732 token=invitation.token,

733 is_active=invitation.is_active,

734 is_expired=invitation.is_expired(),

735 )

736 except HTTPException:

737 raise

738 except (ValueError, TeamMemberLimitExceededError) as e:

739 logger.error(f"Team invitation failed: {e}")

740 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))

741 except Exception as e:

742 logger.error(f"Error creating team invitation for team {SecurityValidator.sanitize_log_message(team_id)}: {e}")

743 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to create invitation")

744

745

746@teams_router.get("/{team_id}/invitations", response_model=List[TeamInvitationResponse])

747@require_permission("teams.read")

748async def list_team_invitations(team_id: str, current_user: dict = Depends(get_current_user_with_permissions), db: Session = Depends(get_db)) -> List[TeamInvitationResponse]:

749 """List team invitations.

750

751 Args:

752 team_id: Team UUID

753 current_user: Authenticated user context dict with email and permissions

754 db: Database session

755

756 Returns:

757 List[TeamInvitationResponse]: List of team invitations

758

759 Raises:

760 HTTPException: If team not found or access denied

761 """

762 try:

763 team_service = TeamManagementService(db)

764 invitation_service = TeamInvitationService(db)

765

766 # Check if user is team owner

767 role = await team_service.get_user_role_in_team(current_user["email"], team_id)

768 if role != "owner":

769 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=_ACCESS_DENIED_MSG)

770

771 invitations = await invitation_service.get_team_invitations(team_id)

772

773 # Get team name for responses

774 team = await team_service.get_team_by_id(team_id)

775 team_name = team.name if team else "Unknown Team"

776

777 invitation_responses = []

778 for invitation in invitations:

779 invitation_responses.append(

780 TeamInvitationResponse(

781 id=invitation.id,

782 team_id=invitation.team_id,

783 team_name=team_name,

784 email=invitation.email,

785 role=invitation.role,

786 invited_by=invitation.invited_by,

787 invited_at=invitation.invited_at,

788 expires_at=invitation.expires_at,

789 token=invitation.token,

790 is_active=invitation.is_active,

791 is_expired=invitation.is_expired(),

792 )

793 )

794

795 db.commit()

796 db.close()

797 return invitation_responses

798 except HTTPException:

799 raise

800 except Exception as e:

801 logger.error(f"Error listing team invitations for team {SecurityValidator.sanitize_log_message(team_id)}: {e}")

802 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to list invitations")

803

804

805@teams_router.post("/invitations/{token}/accept", response_model=TeamMemberResponse)

806@require_permission("teams.join")

807async def accept_team_invitation(token: str, current_user: dict = Depends(get_current_user_with_permissions), db: Session = Depends(get_db)) -> TeamMemberResponse:

808 """Accept a team invitation.

809

810 Args:

811 token: Invitation token

812 current_user: Authenticated user context dict with email and permissions

813 db: Database session

814

815 Returns:

816 TeamMemberResponse: New team member data

817

818 Raises:

819 HTTPException: If invitation not found, expired, or acceptance fails

820 """

821 try:

822 invitation_service = TeamInvitationService(db)

823

824 member = await invitation_service.accept_invitation(token, current_user["email"])

825 if not member or not hasattr(member, "id"):

826 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Invalid or expired invitation")

827

828 db.commit()

829 db.close()

830 return TeamMemberResponse.model_validate(member)

831 except HTTPException:

832 raise

833 except (ValueError, TeamMemberLimitExceededError) as e:

834 logger.error(f"Invitation acceptance failed: {e}")

835 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))

836 except Exception as e:

837 logger.error("Error accepting invitation: %s", e)

838 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to accept invitation")

839

840

841@teams_router.delete("/invitations/{invitation_id}", response_model=SuccessResponse)

842@require_permission("teams.manage_members")

843async def cancel_team_invitation(invitation_id: str, current_user: dict = Depends(get_current_user_with_permissions), db: Session = Depends(get_db)) -> SuccessResponse:

844 """Cancel a team invitation.

845

846 Args:

847 invitation_id: Invitation UUID

848 current_user: Authenticated user context dict with email and permissions

849 db: Database session

850

851 Returns:

852 SuccessResponse: Success confirmation

853

854 Raises:

855 HTTPException: If invitation not found, access denied, or cancellation fails

856 """

857 try:

858 team_service = TeamManagementService(db)

859 invitation_service = TeamInvitationService(db)

860

861 # Get invitation to check team permissions

862 # First-Party

863 from mcpgateway.db import EmailTeamInvitation

864

865 invitation = db.query(EmailTeamInvitation).filter(EmailTeamInvitation.id == invitation_id).first()

866 if not invitation:

867 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Invitation not found")

868

869 # Check if user is team owner or the inviter

870 role = await team_service.get_user_role_in_team(current_user["email"], invitation.team_id)

871 if role != "owner" and current_user["email"] != invitation.invited_by:

872 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=_ACCESS_DENIED_MSG)

873

874 success = await invitation_service.revoke_invitation(invitation_id, current_user["email"])

875 if not success:

876 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Invitation not found")

877

878 db.commit()

879 db.close()

880 return SuccessResponse(message="Team invitation cancelled successfully")

881 except HTTPException:

882 raise

883 except Exception as e:

884 logger.error(f"Error cancelling invitation {invitation_id}: {e}")

885 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to cancel invitation")

886

887

888@teams_router.post("/{team_id}/join", response_model=TeamJoinRequestResponse)

889@require_permission("teams.join")

890async def request_to_join_team(

891 team_id: str,

892 join_request: TeamJoinRequest,

893 current_user: dict = Depends(get_current_user_with_permissions),

894 db: Session = Depends(get_db),

895) -> TeamJoinRequestResponse:

896 """Request to join a public team.

897

898 Allows users to request membership in public teams. The request will be

899 pending until approved by a team owner.

900

901 Args:

902 team_id: ID of the team to join

903 join_request: Join request details including optional message

904 current_user: Currently authenticated user

905 db: Database session

906

907 Returns:

908 TeamJoinRequestResponse: Created join request details

909

910 Raises:

911 HTTPException: If team not found, not public, user already member, or request fails

912 """

913 try:

914 if not settings.allow_team_join_requests:

915 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Team join requests are currently disabled")

916

917 team_service = TeamManagementService(db)

918

919 # Validate team exists and is public

920 team = await team_service.get_team_by_id(team_id)

921 if not team:

922 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Team not found")

923

924 if team.visibility != "public":

925 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Can only request to join public teams")

926

927 # Check if user is already a member

928 user_role = await team_service.get_user_role_in_team(current_user["email"], team_id)

929 if user_role:

930 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="User is already a member of this team")

931

932 # Create join request

933 join_req = await team_service.create_join_request(team_id=team_id, user_email=current_user["email"], message=join_request.message)

934

935 db.commit()

936 db.close()

937 return TeamJoinRequestResponse(

938 id=join_req.id,

939 team_id=join_req.team_id,

940 team_name=team.name,

941 user_email=join_req.user_email,

942 message=join_req.message,

943 status=join_req.status,

944 requested_at=join_req.requested_at,

945 expires_at=join_req.expires_at,

946 )

947 except ValueError as e:

948 # Handle validation errors with 400 Bad Request

949 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))

950 except HTTPException:

951 raise

952 except Exception as e:

953 logger.error(f"Error creating join request for team {SecurityValidator.sanitize_log_message(team_id)}: {e}")

954 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to create join request")

955

956

957@teams_router.delete("/{team_id}/leave", response_model=SuccessResponse)

958@require_permission("teams.join")

959async def leave_team(

960 team_id: str,

961 current_user: dict = Depends(get_current_user_with_permissions),

962 db: Session = Depends(get_db),

963) -> SuccessResponse:

964 """Leave a team.

965

966 Allows users to remove themselves from a team. Cannot leave personal teams

967 or if they are the last owner of a team.

968

969 Args:

970 team_id: ID of the team to leave

971 current_user: Currently authenticated user

972 db: Database session

973

974 Returns:

975 SuccessResponse: Confirmation of leaving the team

976

977 Raises:

978 HTTPException: If team not found, user not member, cannot leave personal team, or last owner

979 """

980 try:

981 team_service = TeamManagementService(db)

982

983 # Validate team exists

984 team = await team_service.get_team_by_id(team_id)

985 if not team:

986 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Team not found")

987

988 # Cannot leave personal team

989 if team.is_personal:

990 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Cannot leave personal team")

991

992 # Check if user is member

993 user_role = await team_service.get_user_role_in_team(current_user["email"], team_id)

994 if not user_role:

995 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="User is not a member of this team")

996

997 # Remove user from team

998 success = await team_service.remove_member_from_team(team_id, current_user["email"], removed_by=current_user["email"])

999 if not success:

1000 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot leave team - you may be the last owner")

1001

1002 db.commit()

1003 db.close()

1004 return SuccessResponse(message="Successfully left the team")

1005 except HTTPException:

1006 raise

1007 except Exception as e:

1008 logger.error(f"Error leaving team {SecurityValidator.sanitize_log_message(team_id)}: {e}")

1009 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to leave team")

1010

1011

1012@teams_router.get("/{team_id}/join-requests", response_model=List[TeamJoinRequestResponse])

1013@require_permission("teams.manage_members")

1014async def list_team_join_requests(

1015 team_id: str,

1016 current_user: dict = Depends(get_current_user_with_permissions),

1017 db: Session = Depends(get_db),

1018) -> List[TeamJoinRequestResponse]:

1019 """List pending join requests for a team.

1020

1021 Only team owners can view join requests for their teams.

1022

1023 Args:

1024 team_id: ID of the team

1025 current_user: Authenticated user context dict with email and permissions

1026 db: Database session

1027

1028 Returns:

1029 List[TeamJoinRequestResponse]: List of pending join requests

1030

1031 Raises:

1032 HTTPException: If team not found or user not authorized

1033 """

1034 try:

1035 team_service = TeamManagementService(db)

1036

1037 # Validate team exists and user is owner

1038 team = await team_service.get_team_by_id(team_id)

1039 if not team:

1040 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Team not found")

1041

1042 user_role = await team_service.get_user_role_in_team(current_user["email"], team_id)

1043 if user_role != "owner":

1044 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Only team owners can view join requests")

1045

1046 # Get join requests

1047 join_requests = await team_service.list_join_requests(team_id)

1048

1049 result = [

1050 TeamJoinRequestResponse(

1051 id=req.id,

1052 team_id=req.team_id,

1053 team_name=team.name,

1054 user_email=req.user_email,

1055 message=req.message,

1056 status=req.status,

1057 requested_at=req.requested_at,

1058 expires_at=req.expires_at,

1059 )

1060 for req in join_requests

1061 ]

1062 db.commit()

1063 db.close()

1064 return result

1065 except HTTPException:

1066 raise

1067 except Exception as e:

1068 logger.error(f"Error listing join requests for team {SecurityValidator.sanitize_log_message(team_id)}: {e}")

1069 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to list join requests")

1070

1071

1072@teams_router.post("/{team_id}/join-requests/{request_id}/approve", response_model=TeamMemberResponse)

1073@require_permission("teams.manage_members")

1074async def approve_join_request(

1075 team_id: str,

1076 request_id: str,

1077 current_user: dict = Depends(get_current_user_with_permissions),

1078 db: Session = Depends(get_db),

1079) -> TeamMemberResponse:

1080 """Approve a team join request.

1081

1082 Only team owners can approve join requests for their teams.

1083

1084 Args:

1085 team_id: ID of the team

1086 request_id: ID of the join request

1087 current_user: Authenticated user context dict with email and permissions

1088 db: Database session

1089

1090 Returns:

1091 TeamMemberResponse: New team member data

1092

1093 Raises:

1094 HTTPException: If request not found or user not authorized

1095 """

1096 try:

1097 team_service = TeamManagementService(db)

1098

1099 # Validate team exists and user is owner

1100 team = await team_service.get_team_by_id(team_id)

1101 if not team:

1102 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Team not found")

1103

1104 user_role = await team_service.get_user_role_in_team(current_user["email"], team_id)

1105 if user_role != "owner":

1106 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Only team owners can approve join requests")

1107

1108 # Approve join request

1109 member = await team_service.approve_join_request(request_id, approved_by=current_user["email"])

1110 if not member:

1111 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Join request not found")

1112

1113 db.commit()

1114 db.close()

1115 return TeamMemberResponse(

1116 id=member.id,

1117 team_id=member.team_id,

1118 user_email=member.user_email,

1119 role=member.role,

1120 joined_at=member.joined_at,

1121 invited_by=member.invited_by,

1122 is_active=member.is_active,

1123 )

1124 except (ValueError, TeamMemberLimitExceededError) as e:

1125 # Handle validation errors with 400 Bad Request

1126 error_msg = str(e)

1127 if "maximum team limit" in error_msg:

1128 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"Cannot approve: {error_msg.lower()}")

1129 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=error_msg)

1130 except HTTPException:

1131 raise

1132 except Exception as e:

1133 logger.error(f"Error approving join request {request_id}: {e}")

1134 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to approve join request")

1135

1136

1137@teams_router.delete("/{team_id}/join-requests/{request_id}", response_model=SuccessResponse)

1138@require_permission("teams.manage_members")

1139async def reject_join_request(

1140 team_id: str,

1141 request_id: str,

1142 current_user: dict = Depends(get_current_user_with_permissions),

1143 db: Session = Depends(get_db),

1144) -> SuccessResponse:

1145 """Reject a team join request.

1146

1147 Only team owners can reject join requests for their teams.

1148

1149 Args:

1150 team_id: ID of the team

1151 request_id: ID of the join request

1152 current_user: Authenticated user context dict with email and permissions

1153 db: Database session

1154

1155 Returns:

1156 SuccessResponse: Confirmation of rejection

1157

1158 Raises:

1159 HTTPException: If request not found or user not authorized

1160 """

1161 try:

1162 team_service = TeamManagementService(db)

1163

1164 # Validate team exists and user is owner

1165 team = await team_service.get_team_by_id(team_id)

1166 if not team:

1167 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Team not found")

1168

1169 user_role = await team_service.get_user_role_in_team(current_user["email"], team_id)

1170 if user_role != "owner":

1171 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Only team owners can reject join requests")

1172

1173 # Reject join request

1174 success = await team_service.reject_join_request(request_id, rejected_by=current_user["email"])

1175 if not success:

1176 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Join request not found")

1177

1178 db.commit()

1179 db.close()

1180 return SuccessResponse(message="Join request rejected successfully")

1181 except HTTPException:

1182 raise

1183 except Exception as e:

1184 logger.error(f"Error rejecting join request {request_id}: {e}")

1185 raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to reject join request")

Coverage for mcpgateway / routers / teams.py: 100%

463 statements