Coverage for model/course.py: 90%

1from typing import Optional 

2from fastapi import APIRouter, Depends 

3from datetime import datetime 

4 

5from mongo import * 

6from .auth import identity_verify, login_required 

7from .utils import * 

8from .schemas import ( 

9 ModifyCoursesBody, 

10 UpdateCourseBody, 

11 AddGradeBody, 

12 UpdateGradeBody, 

13 DeleteGradeBody, 

14 GetCourseScoreboardQuery, 

15) 

16from mongo.utils import * 

17from mongo.course import * 

18from mongo import engine 

19 

20__all__ = ['course_router'] 

21 

22course_router = APIRouter() 

23 

24 

25@course_router.get('') 

26def get_courses(user=Depends(login_required)): 

27 data = [{ 

28 'course': c.course_name, 

29 'teacher': c.teacher.info, 

30 } for c in Course.get_user_courses(user)] 

31 return HTTPResponse('Success.', data=data) 

32 

33 

34@course_router.get('/summary') 

35def get_courses_summary(user: User = identity_verify(0)): 

36 courses = [Course(c) for c in Course.get_all()] 

37 summary = {"courseCount": len(courses), "breakdown": []} 

38 for course in courses: 

39 problems = Problem.get_problem_list(user, course=course.course_name) 

40 course_summary = course.get_course_summary(problems) 

41 course_summary["problemCount"] = len(problems) 

42 summary["breakdown"].append(course_summary) 

43 return HTTPResponse("Success.", data=summary) 

44 

45 

46@course_router.post('') 

47def create_course(body: ModifyCoursesBody, user: User = identity_verify(0, 1)): 

48 teacher = body.teacher 

49 if user.role == 1: 

50 teacher = user.username 

51 try: 

52 Course.add_course(body.course, teacher) 

53 except ValueError: 

54 return HTTPError('Not allowed name.', 400) 

55 except NotUniqueError: 

56 return HTTPError('Course exists.', 400) 

57 except PermissionError: 

58 return HTTPError('Forbidden.', 403) 

59 except engine.DoesNotExist as e: 

60 return HTTPError(f'{e} not found.', 404) 

61 return HTTPResponse('Success.') 

62 

63 

64@course_router.put('') 

65def update_course_meta(body: ModifyCoursesBody, 

66 user: User = identity_verify(0, 1)): 

67 teacher = body.teacher 

68 if user.role == 1: 

69 teacher = user.username 

70 try: 

71 co = Course(body.course) 

72 co.edit_course(user, body.new_course, teacher) 

73 except ValueError: 

74 return HTTPError('Not allowed name.', 400) 

75 except NotUniqueError: 

76 return HTTPError('Course exists.', 400) 

77 except PermissionError: 

78 return HTTPError('Forbidden.', 403) 

79 except engine.DoesNotExist as e: 

80 return HTTPError(f'{e} not found.', 404) 

81 return HTTPResponse('Success.') 

82 

83 

84@course_router.delete('') 

85def delete_course(body: ModifyCoursesBody, user: User = identity_verify(0, 1)): 

86 try: 

87 co = Course(body.course) 

88 co.delete_course(user) 

89 except ValueError: 

90 return HTTPError('Not allowed name.', 400) 

91 except PermissionError: 

92 return HTTPError('Forbidden.', 403) 

93 except engine.DoesNotExist as e: 

94 return HTTPError(f'{e} not found.', 404) 

95 return HTTPResponse('Success.') 

96 

97 

98@course_router.get('/{course_name}') 

99def get_course(course_name: str, user=Depends(login_required)): 

100 course = Course(course_name) 

101 if not course: 

102 return HTTPError('Course not found.', 404) 

103 if not course.permission(user, Course.Permission.VIEW): 

104 return HTTPError('You are not in this course.', 403) 

105 return HTTPResponse( 

106 'Success.', 

107 data={ 

108 "teacher": course.teacher.info, 

109 "TAs": [ta.info for ta in course.tas], 

110 "students": [User(name).info for name in course.student_nicknames], 

111 }, 

112 ) 

113 

114 

115@course_router.put('/{course_name}') 

116def update_course(course_name: str, 

117 body: UpdateCourseBody, 

118 user=Depends(login_required)): 

119 TAs = body.TAs 

120 student_nicknames = body.student_nicknames 

121 course = Course(course_name) 

122 

123 if not course: 

124 return HTTPError('Course not found.', 404) 

125 if not course.permission(user, Course.Permission.VIEW): 

126 return HTTPError('You are not in this course.', 403) 

127 if not course.permission(user, Course.Permission.MODIFY): 

128 return HTTPError('Forbidden.', 403) 

129 

130 tas = [] 

131 for ta in TAs: 

132 permit_user = User(ta).obj 

133 if not User(ta): 

134 return HTTPResponse(f'User: {ta} not found.', 404) 

135 tas.append(permit_user) 

136 

137 for permit_user in set(course.tas) - set(tas): 

138 course.remove_user(permit_user) 

139 for permit_user in set(tas) - set(course.tas): 

140 course.add_user(permit_user) 

141 course.tas = tas 

142 

143 try: 

144 course.update_student_namelist(student_nicknames) 

145 except engine.DoesNotExist as e: 

146 return HTTPError(str(e), 404) 

147 return HTTPResponse('Success.') 

148 

149 

150@course_router.get('/{course_name}/grade/{student}') 

151def get_grade(course_name: str, student: str, user=Depends(login_required)): 

152 course = Course(course_name) 

153 if not course: 

154 return HTTPError('Course not found.', 404) 

155 if not course.permission(user, Course.Permission.VIEW): 

156 return HTTPError('You are not in this course.', 403) 

157 if student not in course.student_nicknames.keys(): 

158 return HTTPError('The student is not in the course.', 404) 

159 if course.permission(user, 

160 Course.Permission.SCORE) and user.username != student: 

161 return HTTPError('You can only view your score.', 403) 

162 return HTTPResponse( 

163 'Success.', 

164 data=[{ 

165 'title': score['title'], 

166 'content': score['content'], 

167 'score': score['score'], 

168 'timestamp': score['timestamp'].timestamp(), 

169 } for score in course.student_scores.get(student, [])], 

170 ) 

171 

172 

173@course_router.post('/{course_name}/grade/{student}') 

174def add_grade( 

175 course_name: str, 

176 student: str, 

177 body: AddGradeBody, 

178 user=Depends(login_required), 

179): 

180 course = Course(course_name) 

181 if not course: 

182 return HTTPError('Course not found.', 404) 

183 if not course.permission(user, Course.Permission.VIEW): 

184 return HTTPError('You are not in this course.', 403) 

185 if student not in course.student_nicknames.keys(): 

186 return HTTPError('The student is not in the course.', 404) 

187 if course.permission(user, Course.Permission.SCORE): 

188 return HTTPError('You can only view your score.', 403) 

189 

190 score_list = course.student_scores.get(student, []) 

191 if body.title in [s['title'] for s in score_list]: 

192 return HTTPError('This title is taken.', 400) 

193 score_list.append({ 

194 'title': body.title, 

195 'content': body.content, 

196 'score': body.score, 

197 'timestamp': datetime.now(), 

198 }) 

199 course.student_scores[student] = score_list 

200 course.save() 

201 return HTTPResponse('Success.') 

202 

203 

204@course_router.put('/{course_name}/grade/{student}') 

205def update_grade( 

206 course_name: str, 

207 student: str, 

208 body: UpdateGradeBody, 

209 user=Depends(login_required), 

210): 

211 course = Course(course_name) 

212 if not course: 

213 return HTTPError('Course not found.', 404) 

214 if not course.permission(user, Course.Permission.VIEW): 

215 return HTTPError('You are not in this course.', 403) 

216 if student not in course.student_nicknames.keys(): 

217 return HTTPError('The student is not in the course.', 404) 

218 if course.permission(user, Course.Permission.SCORE): 

219 return HTTPError('You can only view your score.', 403) 

220 

221 score_list = course.student_scores.get(student, []) 

222 title_list = [s['title'] for s in score_list] 

223 if body.title not in title_list: 

224 return HTTPError('Score not found.', 404) 

225 index = title_list.index(body.title) 

226 title = body.new_title if body.new_title is not None else body.title 

227 if body.new_title is not None and body.new_title in title_list: 

228 return HTTPError('This title is taken.', 400) 

229 score_list[index] = { 

230 'title': title, 

231 'content': body.content, 

232 'score': body.score, 

233 'timestamp': datetime.now(), 

234 } 

235 course.student_scores[student] = score_list 

236 course.save() 

237 return HTTPResponse('Success.') 

238 

239 

240@course_router.delete('/{course_name}/grade/{student}') 

241def delete_grade( 

242 course_name: str, 

243 student: str, 

244 body: DeleteGradeBody, 

245 user=Depends(login_required), 

246): 

247 course = Course(course_name) 

248 if not course: 

249 return HTTPError('Course not found.', 404) 

250 if not course.permission(user, Course.Permission.VIEW): 

251 return HTTPError('You are not in this course.', 403) 

252 if student not in course.student_nicknames.keys(): 

253 return HTTPError('The student is not in the course.', 404) 

254 if course.permission(user, Course.Permission.SCORE): 

255 return HTTPError('You can only view your score.', 403) 

256 

257 score_list = course.student_scores.get(student, []) 

258 title_list = [s['title'] for s in score_list] 

259 if body.title not in title_list: 

260 return HTTPError('Score not found.', 404) 

261 index = title_list.index(body.title) 

262 del score_list[index] 

263 course.student_scores[student] = score_list 

264 course.save() 

265 return HTTPResponse('Success.') 

266 

267 

268@course_router.get('/{course_name}/scoreboard') 

269def get_course_scoreboard( 

270 course_name: str, 

271 query: GetCourseScoreboardQuery = Depends(), 

272 user=Depends(login_required), 

273 course: Course = get_doc('course_name', Course), 

274): 

275 pids = query.pids 

276 start = query.start 

277 end = query.end 

278 try: 

279 pids = pids.split(',') 

280 pids = [int(pid.strip()) for pid in pids] 

281 except Exception: 

282 return HTTPError('Error occurred when parsing `pids`.', 400) 

283 

284 if start: 

285 try: 

286 start = float(start) 

287 except Exception: 

288 return HTTPError('Type of `start` should be float.', 400) 

289 if end: 

290 try: 

291 end = float(end) 

292 except Exception: 

293 return HTTPError('Type of `end` should be float.', 400) 

294 

295 if not course.permission(user, Course.Permission.GRADE): 

296 return HTTPError('Permission denied', 403) 

297 

298 ret = course.get_scoreboard(pids, start, end) 

299 return HTTPResponse('Success.', data=ret)