Coverage for services/team_manager_service.py: 14.49%

1""" 

2Team Manager Service for Missing Table 

3Handles team manager assignments and permissions 

4""" 

5 

6import logging 

7 

8from supabase import Client 

9 

10logger = logging.getLogger(__name__) 

11 

12 

13class TeamManagerService: 

14 """Service for managing team manager assignments""" 

15 

16 def __init__(self, supabase_client: Client): 

17 self.supabase = supabase_client 

18 

19 def assign_team_manager( 

20 self, user_id: str, team_id: int, age_group_id: int, assigned_by_user_id: str 

21 ) -> dict | None: 

22 """ 

23 Assign a user as team manager for a specific team. 

24 

25 Note: age_group_id and assigned_by_user_id are accepted for API compatibility 

26 but not stored. The table schema only has: id, user_id, team_id, created_at. 

27 

28 Args: 

29 user_id: ID of the user to assign as team manager 

30 team_id: ID of the team 

31 age_group_id: ID of the age group (not stored) 

32 assigned_by_user_id: ID of the admin making the assignment (not stored) 

33 

34 Returns: 

35 Assignment record if successful, None otherwise 

36 """ 

37 try: 

38 # Check if assignment already exists (unique constraint on user_id, team_id) 

39 existing = ( 

40 self.supabase.table("team_manager_assignments") 

41 .select("id") 

42 .eq("user_id", user_id) 

43 .eq("team_id", team_id) 

44 .execute() 

45 ) 

46 

47 if existing.data: 

48 logger.warning(f"Team manager assignment already exists for user {user_id}") 

49 return existing.data[0] 

50 

51 # Create assignment (table schema: id, user_id, team_id, created_at) 

52 assignment_data = { 

53 "user_id": user_id, 

54 "team_id": team_id, 

55 } 

56 

57 response = self.supabase.table("team_manager_assignments").insert(assignment_data).execute() 

58 

59 if response.data: 

60 logger.info(f"Assigned user {user_id} as team manager for team {team_id}") 

61 return response.data[0] 

62 

63 return None 

64 

65 except Exception as e: 

66 logger.error(f"Error assigning team manager: {e}") 

67 return None 

68 

69 def get_user_team_assignments(self, user_id: str) -> list[dict]: 

70 """ 

71 Get all team assignments for a user 

72 

73 Args: 

74 user_id: ID of the user 

75 

76 Returns: 

77 List of team assignments 

78 """ 

79 try: 

80 response = ( 

81 self.supabase.table("team_manager_assignments") 

82 .select("*, teams(id, name), age_groups(id, name)") 

83 .eq("user_id", user_id) 

84 .execute() 

85 ) 

86 

87 return response.data if response.data else [] 

88 

89 except Exception as e: 

90 logger.error(f"Error getting team assignments for user {user_id}: {e}") 

91 return [] 

92 

93 def can_manage_team(self, user_id: str, team_id: int, age_group_id: int) -> bool: 

94 """ 

95 Check if a user can manage a specific team. 

96 

97 Note: age_group_id is accepted for API compatibility but not checked. 

98 The team_manager_assignments table grants access to the entire team. 

99 

100 Args: 

101 user_id: ID of the user 

102 team_id: ID of the team 

103 age_group_id: ID of the age group (not used - kept for API compatibility) 

104 

105 Returns: 

106 True if user can manage, False otherwise 

107 """ 

108 try: 

109 # Check if user is admin 

110 user_response = self.supabase.table("user_profiles").select("role").eq("id", user_id).single().execute() 

111 

112 if user_response.data and user_response.data["role"] == "admin": 

113 return True 

114 

115 # Check if user has team manager assignment (by team_id only) 

116 response = ( 

117 self.supabase.table("team_manager_assignments") 

118 .select("id") 

119 .eq("user_id", user_id) 

120 .eq("team_id", team_id) 

121 .execute() 

122 ) 

123 

124 return bool(response.data) 

125 

126 except Exception as e: 

127 logger.error(f"Error checking team management permission: {e}") 

128 return False 

129 

130 def remove_team_manager(self, user_id: str, team_id: int, age_group_id: int) -> bool: 

131 """ 

132 Remove a team manager assignment. 

133 

134 Note: age_group_id is accepted for API compatibility but not used. 

135 Deletion is by user_id and team_id only. 

136 

137 Args: 

138 user_id: ID of the user 

139 team_id: ID of the team 

140 age_group_id: ID of the age group (not used) 

141 

142 Returns: 

143 True if successful, False otherwise 

144 """ 

145 try: 

146 response = ( 

147 self.supabase.table("team_manager_assignments") 

148 .delete() 

149 .eq("user_id", user_id) 

150 .eq("team_id", team_id) 

151 .execute() 

152 ) 

153 

154 if response.data: 

155 logger.info(f"Removed team manager assignment for user {user_id}") 

156 return True 

157 

158 return False 

159 

160 except Exception as e: 

161 logger.error(f"Error removing team manager: {e}") 

162 return False 

163 

164 def get_team_managers(self, team_id: int, age_group_id: int | None = None) -> list[dict]: 

165 """ 

166 Get all team managers for a specific team 

167 

168 Args: 

169 team_id: ID of the team 

170 age_group_id: Optional age group filter 

171 

172 Returns: 

173 List of team managers 

174 """ 

175 try: 

176 select_fields = ( 

177 "*, user:user_profiles!team_manager_assignments_user_id_fkey" 

178 "(id, email, first_name, last_name), age_groups(name)" 

179 ) 

180 query = self.supabase.table("team_manager_assignments").select(select_fields).eq("team_id", team_id) 

181 

182 if age_group_id: 

183 query = query.eq("age_group_id", age_group_id) 

184 

185 response = query.execute() 

186 

187 return response.data if response.data else [] 

188 

189 except Exception as e: 

190 logger.error(f"Error getting team managers: {e}") 

191 return []