Coverage for app/logic/participants.py: 90%

1from flask import g 

2from peewee import fn, JOIN 

3from datetime import date 

4from typing import List 

5 

6from app.models.user import User 

7from app.models.event import Event 

8from app.models.term import Term 

9from app.models.eventRsvp import EventRsvp 

10from app.models.program import Program 

11from app.models.eventParticipant import EventParticipant 

12from app.logic.users import isEligibleForProgram 

13from app.logic.volunteers import getEventLengthInHours 

14from app.logic.events import getEventRsvpCountsForTerm 

15from app.logic.createLogs import createRsvpLog 

16 

17def trainedParticipants(programID: int, targetTerm: Term) -> List[User]: 

18 """ 

19 This function tracks the users who have attended every Prerequisite 

20 event and adds them to a list that will not flag them when tracking hours. 

21 Returns a list of user objects who've completed all training events. 

22 """ 

23 

24 # Reset program eligibility each term for all other trainings 

25 isRelevantAllVolunteer: bool = (Event.isAllVolunteerTraining) & (Event.term.academicYear == targetTerm.academicYear) 

26 isRelevantProgramTraining: bool = (Event.program == programID) & (Event.term == targetTerm) & (Event.isTraining) 

27 allTrainings: List[Event] = (Event.select() 

28 .join(Term) 

29 .where(isRelevantAllVolunteer | isRelevantProgramTraining, 

30 Event.isCanceled == False)) 

31 

32 fullyTrainedUsers: List[User] = (User.select() 

33 .join(EventParticipant) 

34 .where(EventParticipant.event.in_(allTrainings)) 

35 .group_by(EventParticipant.user) 

36 .having(fn.Count(EventParticipant.user) == len(allTrainings)).order_by(User.username)) 

37 

38 return list(fullyTrainedUsers) 

39 

40def addBnumberAsParticipant(bnumber, eventId): 

41 """Accepts scan input and signs in the user. If user exists or is already 

42 signed in will return user and login status""" 

43 try: 

44 kioskUser = User.get(User.bnumber == bnumber) 

45 except Exception as e: 

46 print(e) 

47 return None, "does not exist" 

48 

49 event = Event.get_by_id(eventId) 

50 if not isEligibleForProgram(event.program, kioskUser): 

51 userStatus = "banned" 

52 

53 elif checkUserVolunteer(kioskUser, event): 

54 userStatus = "already signed in" 

55 

56 else: 

57 userStatus = "success" 

58 # We are not using addPersonToEvent to do this because  

59 # that function checks if the event is in the past, but 

60 # someone could start signing people up via the kiosk 

61 # before an event has started 

62 totalHours = getEventLengthInHours(event.timeStart, event.timeEnd, event.startDate) 

63 EventParticipant.create (user=kioskUser, event=event, hoursEarned=totalHours) 

64 

65 return kioskUser, userStatus 

66 

67def checkUserRsvp(user, event): 

68 return EventRsvp.select().where(EventRsvp.user==user, EventRsvp.event == event).exists() 

69 

70def checkUserVolunteer(user, event): 

71 return EventParticipant.select().where(EventParticipant.user == user, EventParticipant.event == event).exists() 

72 

73def addPersonToEvent(user, event): 

74 """ 

75 Add a user to an event. 

76 If the event is in the past, add the user as a volunteer (EventParticipant) including hours worked. 

77 If the event is in the future, rsvp for the user (EventRsvp) 

78 

79 Returns True if the operation was successful, false otherwise 

80 """ 

81 try: 

82 volunteerExists = checkUserVolunteer(user, event) 

83 rsvpExists = checkUserRsvp(user, event) 

84 if event.isPast: 

85 if not volunteerExists: 

86 # We duplicate these two lines in addBnumberAsParticipant 

87 eventHours = getEventLengthInHours(event.timeStart, event.timeEnd, event.startDate) 

88 EventParticipant.create(user = user, event = event, hoursEarned = eventHours) 

89 else: 

90 if not rsvpExists: 

91 currentRsvp = getEventRsvpCountsForTerm(event.term) 

92 waitlist = currentRsvp[event.id] >= event.rsvpLimit if event.rsvpLimit is not None else 0 

93 EventRsvp.create(user = user, event = event, rsvpWaitlist = waitlist) 

94 

95 targetList = "the waitlist" if waitlist else "the RSVP list" 

96 if g.current_user.username == user.username: 

97 createRsvpLog(event.id, f"{user.fullName} joined {targetList}.") 

98 else: 

99 createRsvpLog(event.id, f"Added {user.fullName} to {targetList}.") 

100 

101 if volunteerExists or rsvpExists: 

102 return "already in" 

103 except Exception as e: 

104 print(e) 

105 return False 

106 

107 return True 

108 

109def unattendedRequiredEvents(program, user): 

110 

111 # Check for events that are prerequisite for program 

112 requiredEvents = (Event.select(Event) 

113 .where(Event.isTraining == True, Event.program == program)) 

114 

115 if requiredEvents: 

116 attendedRequiredEventsList = [] 

117 for event in requiredEvents: 

118 attendedRequirement = (EventParticipant.select().where(EventParticipant.user == user, EventParticipant.event == event)) 

119 if not attendedRequirement: 

120 attendedRequiredEventsList.append(event.name) 

121 if attendedRequiredEventsList is not None: 

122 return attendedRequiredEventsList 

123 else: 

124 return [] 

125 

126 

127def getEventParticipants(event): 

128 eventParticipants = (EventParticipant.select(EventParticipant, User) 

129 .join(User) 

130 .where(EventParticipant.event == event)) 

131 

132 return [p for p in eventParticipants] 

133 

134def getParticipationStatusForTrainings(program, userList, term): 

135 """ 

136 This function returns a dictionary of all trainings for a program and 

137 whether the current user participated in them. 

138 

139 :returns: trainings for program and if the user participated 

140 """ 

141 isRelevantAllVolunteer = (Event.isAllVolunteerTraining) & (Event.term.academicYear == term.academicYear) 

142 isRelevantProgramTraining = (Event.program == program) & (Event.term == term) & (Event.isTraining) 

143 programTrainings = (Event.select(Event, Term, EventParticipant, EventRsvp) 

144 .join(EventParticipant, JOIN.LEFT_OUTER).switch() 

145 .join(EventRsvp, JOIN.LEFT_OUTER).switch() 

146 .join(Term) 

147 .where(isRelevantAllVolunteer | isRelevantProgramTraining, (Event.isCanceled != True)).order_by(Event.startDate)) 

148 

149 # Create a dictionary where the keys are trainings and values are a list of those who attended 

150 trainingData = {} 

151 for training in programTrainings: 

152 try: 

153 if training.isPast: 

154 trainingData[training] = trainingData.get(training, []) + [training.eventparticipant.user_id] 

155 else: # The training has yet to happen 

156 trainingData[training] = trainingData.get(training, []) + [training.eventrsvp.user_id] 

157 except AttributeError: 

158 trainingData[training] = trainingData.get(training, []) 

159 # Create a dictionary binding usernames to tuples. The tuples consist of the training (event object) and whether or not they attended it (bool) 

160 userParticipationStatus = {} 

161 for user in userList: 

162 for training, attendeeList in trainingData.items(): 

163 userParticipationStatus[user.username] = userParticipationStatus.get(user.username, []) + [(training, user.username in attendeeList)] 

164 

165 return userParticipationStatus 

166 

167def sortParticipantsByStatus(event): 

168 """ 

169 Takes in an event object, queries all participants, and then filters those 

170 participants by their attendee status. 

171 

172 return: a list of participants who didn't attend, a list of participants who are waitlisted, 

173 a list of participants who attended, and a list of all participants who have some status for the  

174 event. 

175 """ 

176 eventParticipants = getEventParticipants(event) 

177 

178 # get all RSVPs for event and filter out those that did not attend into separate list 

179 eventRsvpData = list(EventRsvp.select(EventRsvp, User).join(User).where(EventRsvp.event==event).order_by(EventRsvp.rsvpTime)) 

180 eventNonAttendedData = [rsvp for rsvp in eventRsvpData if rsvp.user not in eventParticipants] 

181 

182 if event.isPast: 

183 eventVolunteerData = eventParticipants 

184 

185 # if the event date has passed disregard the waitlist 

186 eventWaitlistData = [] 

187 else: 

188 # if rsvp is required for the event, grab all volunteers that are in the waitlist 

189 eventWaitlistData = [volunteer for volunteer in (eventParticipants + eventRsvpData) if volunteer.rsvpWaitlist and event.isRsvpRequired] 

190 

191 # put the rest of the users that are not on the waitlist into the volunteer data 

192 eventVolunteerData = [volunteer for volunteer in eventNonAttendedData if volunteer not in eventWaitlistData] 

193 eventNonAttendedData = [] 

194 

195 return eventNonAttendedData, eventWaitlistData, eventVolunteerData, eventParticipants