Coverage for model/post.py: 98%

1from fastapi import APIRouter, Depends 

2from mongo import * 

3from mongo import engine 

4from .auth import login_required 

5from .utils import * 

6from .schemas import ModifyPostBody 

7from mongo.utils import * 

8from mongo.post import * 

9from mongo.course import * 

10 

11__all__ = ['post_router'] 

12 

13post_router = APIRouter() 

14 

15 

16@post_router.get('/{course}') 

17def get_post(course: str, user=Depends(login_required)): 

18 target_course = Course(course) 

19 if not target_course: 

20 return HTTPError("Course not found.", 404) 

21 if not target_course.permission(user, Course.Permission.VIEW): 

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

23 data = Post.found_post(target_course) 

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

25 

26 

27@post_router.get('/view/{course}/{target_thread_id}') 

28def get_single_post(course: str, 

29 target_thread_id: str, 

30 user=Depends(login_required)): 

31 target_course = Course(course) 

32 if not target_course: 

33 return HTTPError("Course not found.", 404) 

34 if not target_course.permission(user, Course.Permission.VIEW): 

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

36 data = Post.found_post(target_course, target_thread_id) 

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

38 

39 

40def _resolve_post_target(body: ModifyPostBody, user): 

41 """Shared logic to resolve course/thread context and check permissions.""" 

42 course = body.course 

43 target_thread_id = body.target_thread_id 

44 

45 if course == 'Public': 

46 return HTTPError('You can not add post in system.', 

47 403), None, None, None 

48 

49 if course and target_thread_id: 

50 return HTTPError( 

51 'Request is fail,course or target_thread_id must be none.', 

52 400), None, None, None 

53 

54 if course: 

55 course_obj = Course(course) 

56 if not course_obj: 

57 return HTTPError('Course not exist.', 404), None, None, None 

58 target_course = course_obj 

59 target_thread = None 

60 elif target_thread_id: 

61 try: 

62 target_thread = engine.PostThread.objects.get(id=target_thread_id) 

63 except engine.DoesNotExist: 

64 try: 

65 target_post = engine.Post.objects.get(id=target_thread_id) 

66 except engine.DoesNotExist: 

67 return HTTPError('Post/reply not exist.', 

68 404), None, None, None 

69 target_thread = target_post.thread 

70 target_thread_id = target_thread.id 

71 if target_thread.status: 

72 return HTTPResponse('Forbidden,the post/reply is deleted.', 

73 403), None, None, None 

74 target_course = Course(target_thread.course_id) 

75 else: 

76 return HTTPError( 

77 'Request is fail,course and target_thread_id are both none.', 

78 400), None, None, None 

79 

80 capability = target_course.own_permission(user) 

81 if not (capability & Course.Permission.VIEW): 

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

83 

84 return None, target_course, target_thread, capability 

85 

86 

87@post_router.post('') 

88def add_post(body: ModifyPostBody, user=Depends(login_required)): 

89 err, target_course, target_thread, capability = _resolve_post_target( 

90 body, user) 

91 if err is not None: 

92 return err 

93 course = body.course 

94 target_thread_id = body.target_thread_id 

95 if course: 

96 r = Post.add_post(course, user, body.content, body.title) 

97 else: 

98 r = Post.add_reply(target_thread, user, body.content) 

99 if r is not None: 

100 return HTTPError(r, 403) 

101 return HTTPResponse('success.') 

102 

103 

104@post_router.put('') 

105def edit_post(body: ModifyPostBody, user=Depends(login_required)): 

106 err, target_course, target_thread, capability = _resolve_post_target( 

107 body, user) 

108 if err is not None: 

109 return err 

110 if body.course: 

111 return HTTPError( 

112 "Request is fail,you should provide target_thread_id replace course.", 

113 400) 

114 r = Post.edit_post(target_thread, user, body.content, body.title, 

115 capability) 

116 if r is not None: 

117 return HTTPError(r, 403) 

118 return HTTPResponse('success.') 

119 

120 

121@post_router.delete('') 

122def delete_post(body: ModifyPostBody, user=Depends(login_required)): 

123 err, target_course, target_thread, capability = _resolve_post_target( 

124 body, user) 

125 if err is not None: 

126 return err 

127 if body.course: 

128 return HTTPError( 

129 "Request is fail,you should provide target_thread_id replace course.", 

130 400) 

131 r = Post.delete_post(target_thread, user, capability) 

132 if r is not None: 

133 return HTTPError(r, 403) 

134 return HTTPResponse('success.')