Coverage for model/homework.py: 89%

1from fastapi import APIRouter, Depends 

2from mongo import * 

3from mongo import engine 

4from .utils import * 

5from .auth import login_required 

6from .course import course_router 

7from .schemas import CreateHomeworkBody, UpdateHomeworkBody, PatchIpFiltersBody 

8 

9__all__ = ['homework_router'] 

10 

11homework_router = APIRouter() 

12 

13 

14@homework_router.post('') 

15def create_homework(body: CreateHomeworkBody, user=Depends(login_required)): 

16 try: 

17 Homework.add( 

18 user=user, 

19 hw_name=body.name, 

20 markdown=body.markdown, 

21 scoreboard_status=body.scoreboard_status, 

22 course_name=body.course_name, 

23 problem_ids=body.problem_ids or [], 

24 start=body.start, 

25 end=body.end, 

26 penalty=body.penalty, 

27 ) 

28 except NameError: 

29 return HTTPError('user must be the teacher or ta of this course', 403) 

30 except FileExistsError: 

31 return HTTPError('homework exists in this course', 400) 

32 except engine.DoesNotExist as e: 

33 return HTTPError(str(e), 404) 

34 except (PermissionError, engine.NotUniqueError) as e: 

35 return HTTPError(str(e), 403) 

36 return HTTPResponse('Add homework Success') 

37 

38 

39@homework_router.get('/{homework_id}') 

40def get_homework(homework_id: str, user=Depends(login_required)): 

41 try: 

42 homework = Homework.get_by_id(homework_id) 

43 ret = { 

44 'name': 

45 homework.homework_name, 

46 'start': 

47 int(homework.duration.start.timestamp()), 

48 'end': 

49 int(homework.duration.end.timestamp()), 

50 'problemIds': 

51 homework.problem_ids, 

52 'markdown': 

53 homework.markdown, 

54 'studentStatus': (homework.student_status if user.role < 2 else 

55 homework.student_status.get(user.username)), 

56 'penalty': 

57 homework.penalty if hasattr(homework, 'penalty') else None, 

58 } 

59 except engine.DoesNotExist as e: 

60 return HTTPError(str(e), 404) 

61 except (PermissionError, engine.NotUniqueError) as e: 

62 return HTTPError(str(e), 403) 

63 return HTTPResponse('get homework', data=ret) 

64 

65 

66@homework_router.put('/{homework_id}') 

67def update_homework(homework_id: str, 

68 body: UpdateHomeworkBody, 

69 user=Depends(login_required)): 

70 try: 

71 Homework.update( 

72 user=user, 

73 homework_id=homework_id, 

74 markdown=body.markdown, 

75 new_hw_name=body.name, 

76 start=body.start, 

77 end=body.end, 

78 problem_ids=body.problem_ids, 

79 scoreboard_status=body.scoreboard_status, 

80 penalty=body.penalty, 

81 ) 

82 except engine.DoesNotExist as e: 

83 return HTTPError(str(e), 404) 

84 except (PermissionError, engine.NotUniqueError) as e: 

85 return HTTPError(str(e), 403) 

86 return HTTPResponse('Update homework Success') 

87 

88 

89@homework_router.delete('/{homework_id}') 

90def delete_homework(homework_id: str, user=Depends(login_required)): 

91 try: 

92 homework = Homework(homework_id) 

93 homework.delete_problems(user=user, course=homework.course_id) 

94 except engine.DoesNotExist as e: 

95 return HTTPError(str(e), 404) 

96 except (PermissionError, engine.NotUniqueError) as e: 

97 return HTTPError(str(e), 403) 

98 return HTTPResponse('Delete homework Success') 

99 

100 

101@course_router.get('/{course_name}/homework') 

102def get_homework_list(course_name: str, user=Depends(login_required)): 

103 '''Get a list of homework for a course.''' 

104 try: 

105 homeworks = Homework.get_homeworks(course_name=course_name) 

106 data = [] 

107 for homework in homeworks: 

108 new = { 

109 'name': homework.homework_name, 

110 'start': int(homework.duration.start.timestamp()), 

111 'end': int(homework.duration.end.timestamp()), 

112 'problemIds': homework.problem_ids, 

113 'markdown': homework.markdown, 

114 'id': str(homework.id), 

115 } 

116 if user.role < 2: 

117 new['studentStatus'] = homework.student_status 

118 else: 

119 new['studentStatus'] = homework.student_status.get( 

120 user.username) 

121 data.append(new) 

122 except DoesNotExist: 

123 return HTTPError('course not exists', 

124 404, 

125 data={'courseName': course_name}) 

126 return HTTPResponse('get homeworks', data=data) 

127 

128 

129@homework_router.get('/{course}/{homework_name}/ip-filters') 

130def get_ip_filters(course: str, 

131 homework_name: str, 

132 user=Depends(login_required)): 

133 if user.role != 0: 

134 return HTTPError('Not admin!', 403) 

135 try: 

136 hw = Homework.get_by_name(course, homework_name) 

137 except DoesNotExist: 

138 return HTTPError('Homework does not exist', 404) 

139 return HTTPResponse(data={'ipFilters': hw.ip_filters}) 

140 

141 

142@homework_router.patch('/{course}/{homework_name}/ip-filters') 

143def patch_ip_filters( 

144 course: str, 

145 homework_name: str, 

146 body: PatchIpFiltersBody, 

147 user=Depends(login_required), 

148): 

149 if user.role != 0: 

150 return HTTPError('Not admin!', 403) 

151 try: 

152 hw = Homework.get_by_name(course, homework_name) 

153 except DoesNotExist: 

154 return HTTPError('Homework does not exist', 404) 

155 adds = [] 

156 dels = [] 

157 for patch in body.patches: 

158 op = patch.get('op') 

159 if op not in {'add', 'del'}: 

160 return HTTPError('Invalid operation', 400, data={'op': op}) 

161 value = patch.get('value') 

162 if value is None: 

163 return HTTPError('Value not found', 400) 

164 if op == 'add': 

165 adds.append(value) 

166 else: 

167 dels.append(value) 

168 try: 

169 IPFilter(value) 

170 except ValueError as e: 

171 return HTTPError(str(e), 400) 

172 hw.update(push_all__ip_filters=adds) 

173 hw.update(pull_all__ip_filters=dels) 

174 return HTTPResponse()