Coverage for app/controllers/admin/volunteers.py: 27%

153 statements  

« prev     ^ index     » next       coverage.py v7.2.7, created at 2024-03-11 22:25 +0000

1from flask import request, render_template, redirect, url_for, flash, abort, g, json, jsonify 

2from peewee import DoesNotExist, JOIN 

3from playhouse.shortcuts import model_to_dict 

4from typing import List 

5 

6from app.controllers.admin import admin_bp 

7from app.models.event import Event 

8from app.models.program import Program 

9from app.models.user import User 

10from app.models.eventParticipant import EventParticipant 

11from app.models.emergencyContact import EmergencyContact 

12 

13from app.logic.userManagement import getAllowedPrograms 

14from app.logic.searchUsers import searchUsers 

15from app.logic.volunteers import updateEventParticipants, getEventLengthInHours, addUserBackgroundCheck, setProgramManager 

16from app.logic.participants import trainedParticipants, addPersonToEvent, getParticipationStatusForTrainings, sortParticipantsByStatus 

17from app.logic.events import getPreviousRecurringEventData, getEventRsvpCount 

18from app.models.eventRsvp import EventRsvp 

19from app.models.backgroundCheck import BackgroundCheck 

20from app.logic.createLogs import createAdminLog, createRsvpLog 

21from app.logic.users import getBannedUsers, isBannedFromEvent 

22 

23 

24@admin_bp.route('/searchVolunteers/<query>', methods = ['GET']) 

25def getVolunteers(query: str) -> str: 

26 '''Accepts user input and queries the database returning results that matches user search''' 

27 

28 return json.dumps(searchUsers(query)) 

29 

30@admin_bp.route('/event/<eventID>/manage_volunteers', methods=['GET', 'POST']) 

31def manageVolunteersPage(eventID) -> str: 

32 """ 

33 Controller that handles POST and GET requests regarding the Manage Volunteers page. 

34 

35 POST: updates the event participants for a particular event by calling  

36 updateEventParticipants on the form. 

37 

38 GET: retrieves all necessary participant lists and dictionaries and categorizes 

39 the participants/volunteers into their respective participation statuses. Then  

40 renders the manageVolunteers.html template. 

41 """ 

42 try: 

43 event: Event = Event.get_by_id(eventID) 

44 except DoesNotExist as e: 

45 print(f"No event found for {eventID}", e) 

46 abort(404) 

47 

48 # ------------ POST request ------------ 

49 if request.method == "POST": 

50 volunteerUpdated: bool = updateEventParticipants(request.form) 

51 

52 # error handling depending on the boolean returned from updateEventParticipants 

53 if volunteerUpdated: 

54 flash("Volunteer table succesfully updated", "success") 

55 else: 

56 flash("Error adding volunteer", "danger") 

57 return redirect(url_for("admin.manageVolunteersPage", eventID=eventID)) 

58 

59 # ------------ GET request ------------ 

60 canAccessProgram: bool = event.program in getAllowedPrograms(g.current_user) 

61 if not canAccessProgram: 

62 abort(403) 

63 

64 # ------- Grab the different lists of participants ------- 

65 trainedParticipantsForProgramAndTerm: List[User] = trainedParticipants(event.program, event.term) 

66 

67 bannedUsersForProgram: List[User] = [bannedUser.user for bannedUser in getBannedUsers(event.program)] 

68 

69 eventNonAttendedData, eventWaitlistData, eventVolunteerData, eventParticipants = sortParticipantsByStatus(event) 

70 

71 allRelevantUsers = [participant.user for participant in (eventParticipants + eventNonAttendedData + eventWaitlistData)] 

72 

73 # ----------- Get miscellaneous data ----------- 

74 

75 participationStatusForTrainings = getParticipationStatusForTrainings(event.program, allRelevantUsers, event.term) 

76 

77 eventLengthInHours: float = getEventLengthInHours(event.timeStart, event.timeEnd, event.startDate) 

78 

79 recurringVolunteers = getPreviousRecurringEventData(event.recurringId) 

80 

81 currentRsvpAmount: int = getEventRsvpCount(event.id) 

82 

83 # ----------- Render template with all of the data ------------ 

84 

85 return render_template("/events/manageVolunteers.html", 

86 eventVolunteerData = eventVolunteerData, 

87 eventNonAttendedData = eventNonAttendedData, 

88 eventWaitlistData = eventWaitlistData, 

89 eventLength = eventLengthInHours, 

90 event = event, 

91 recurringVolunteers = recurringVolunteers, 

92 bannedUsersForProgram = bannedUsersForProgram, 

93 trainedParticipantsForProgramAndTerm = trainedParticipantsForProgramAndTerm, 

94 participationStatusForTrainings = participationStatusForTrainings, 

95 currentRsvpAmount = currentRsvpAmount) 

96 

97@admin_bp.route('/event/<eventID>/volunteer_details', methods=['GET']) 

98def volunteerDetailsPage(eventID) -> str: 

99 try: 

100 event: Event = Event.get_by_id(eventID) 

101 except DoesNotExist as e: 

102 print(f"No event found for {eventID}", e) 

103 abort(404) 

104 canAccessProgram: bool = event.program in getAllowedPrograms(g.current_user) 

105 if not (canAccessProgram): 

106 abort(403) 

107 

108 eventRsvpData: List = list(EventRsvp.select(EmergencyContact, EventRsvp) 

109 .join(EmergencyContact, JOIN.LEFT_OUTER, on=(EmergencyContact.user==EventRsvp.user)) 

110 .where(EventRsvp.event==event)) 

111 eventParticipantData = list(EventParticipant.select(EmergencyContact, EventParticipant) 

112 .join(EmergencyContact, JOIN.LEFT_OUTER, on=(EmergencyContact.user==EventParticipant.user)) 

113 .where(EventParticipant.event==event)) 

114 

115 waitlistUser = list(set([obj for obj in eventRsvpData if obj.rsvpWaitlist])) 

116 rsvpUser = list(set([obj for obj in eventRsvpData if not obj.rsvpWaitlist ])) 

117 

118 return render_template("/events/volunteerDetails.html", 

119 waitlistUser = waitlistUser, 

120 attendedUser= eventParticipantData, 

121 rsvpUser= rsvpUser, 

122 event = event) 

123 

124 

125@admin_bp.route('/addVolunteersToEvent/<eventId>', methods = ['POST']) 

126def addVolunteer(eventId): 

127 event = Event.get_by_id(eventId) 

128 successfullyAddedVolunteer = False 

129 usernameList = [] 

130 usernameList = request.form.getlist("volunteer[]") 

131 

132 successfullyAddedVolunteer = False 

133 alreadyAddedList = [] 

134 addedSuccessfullyList = [] 

135 errorList = [] 

136 

137 for user in usernameList: 

138 userObj = User.get_by_id(user) 

139 successfullyAddedVolunteer = addPersonToEvent(userObj, event) 

140 if successfullyAddedVolunteer == "already in": 

141 alreadyAddedList.append(userObj.fullName) 

142 else: 

143 if successfullyAddedVolunteer: 

144 addedSuccessfullyList.append(userObj.fullName) 

145 else: 

146 errorList.append(userObj.fullName) 

147 

148 volunteers = "" 

149 if alreadyAddedList: 

150 volunteers = ", ".join(vol for vol in alreadyAddedList) 

151 flash(f"{volunteers} already in table.", "warning") 

152 

153 if addedSuccessfullyList: 

154 volunteers = ", ".join(vol for vol in addedSuccessfullyList) 

155 flash(f"{volunteers} added successfully.", "success") 

156 

157 if errorList: 

158 volunteers = ", ".join(vol for vol in errorList) 

159 flash(f"Error when adding {volunteers} to event.", "danger") 

160 

161 if 'ajax' in request.form and request.form['ajax']: 

162 return '' 

163 

164 return redirect(url_for('admin.manageVolunteersPage', eventID = eventId)) 

165 

166@admin_bp.route('/rsvpFromWaitlist/<username>/<eventId>', methods = ['POST']) 

167def rsvpFromWaitlist(username, eventId): 

168 event = Event.get_by_id(eventId) 

169 isProgramManager = g.current_user.isProgramManagerFor(event.program) 

170 if g.current_user.isCeltsAdmin or (g.current_user.isCeltsStudentStaff and isProgramManager): 

171 waitlistUsers = EventRsvp.select(EventRsvp, User).join(User).where(EventRsvp.user == username, EventRsvp.event==eventId).execute() 

172 if (waitlistUsers): 

173 createRsvpLog(event.id, f"Moved {waitlistUsers[0].user.fullName} from waitlist to RSVP.") 

174 (EventRsvp.update(rsvpWaitlist = False).where(EventRsvp.event_id == eventId, EventRsvp.user_id == username)).execute() 

175 return "" 

176 

177@admin_bp.route('/addVolunteersToEvent/<username>/<eventId>/isBanned', methods = ['GET']) 

178def isVolunteerBanned(username, eventId): 

179 return {"banned":1} if isBannedFromEvent(username, eventId) else {"banned":0} 

180 

181@admin_bp.route('/removeVolunteerFromEvent', methods = ['POST']) 

182def removeVolunteerFromEvent(): 

183 user = request.form.get('username') 

184 eventID = request.form.get('eventId') 

185 if g.current_user.isAdmin: 

186 userInRsvpTable = EventRsvp.select(EventRsvp, User).join(User).where(EventRsvp.user == user, EventRsvp.event==eventID).execute() 

187 if (userInRsvpTable): 

188 rsvpUser = userInRsvpTable[0] 

189 if rsvpUser.rsvpWaitlist: 

190 createRsvpLog(eventID, f"Removed {rsvpUser.user.fullName} from waitlist.") 

191 else: 

192 createRsvpLog(eventID, f"Removed {rsvpUser.user.fullName} from RSVP list.") 

193 (EventParticipant.delete().where(EventParticipant.user==user, EventParticipant.event==eventID)).execute() 

194 (EventRsvp.delete().where(EventRsvp.user==user, EventRsvp.event==eventID)).execute() 

195 flash("Volunteer successfully removed", "success") 

196 return "" 

197 

198@admin_bp.route('/addBackgroundCheck', methods = ['POST']) 

199def addBackgroundCheck(): 

200 if g.current_user.isCeltsAdmin: 

201 eventData = request.form 

202 user = eventData['user'] 

203 bgStatus = eventData['bgStatus'] 

204 type = eventData['bgType'] 

205 dateCompleted = eventData['bgDate'] 

206 addUserBackgroundCheck(user, type, bgStatus, dateCompleted) 

207 return " " 

208 

209@admin_bp.route('/deleteBackgroundCheck', methods = ['POST']) 

210def deleteBackgroundCheck(): 

211 if g.current_user.isCeltsAdmin: 

212 eventData = request.form 

213 bgToDelete = BackgroundCheck.get_by_id(eventData['bgID']) 

214 BackgroundCheck.delete().where(BackgroundCheck.id == bgToDelete).execute() 

215 return "" 

216 

217@admin_bp.route('/updateProgramManager', methods=["POST"]) 

218def updateProgramManager(): 

219 if g.current_user.isCeltsAdmin: 

220 data =request.form 

221 username = User.get(User.username == data["user_name"]) 

222 program = Program.get_by_id(data['program_id']) 

223 setProgramManager(data["user_name"], data["program_id"], data["action"]) 

224 createAdminLog(f'{username.firstName} has been {data["action"]}ed as a Program Manager for {program.programName}') 

225 return "" 

226 else: 

227 abort(403) 

228 

229@admin_bp.route("/updatePhone", methods=["POST"]) 

230def updatePhone(): 

231 newinfo=request.form 

232 User.update(phoneNumber=newinfo["phoneNumber"]).where(User.username==newinfo["username"]).execute() 

233 return ""