Coverage for model/user.py: 100%

1import logging 

2from urllib import parse 

3from typing import Optional 

4from fastapi import APIRouter, Depends 

5from mongo import engine 

6from mongo.utils import drop_none 

7from mongo import * 

8from .utils import * 

9from .auth import identity_verify, login_required 

10from .schemas import AddUserBody, UpdateUserBody, GetUserListQuery 

11 

12__all__ = ['user_router', 'user_options_router'] 

13 

14logger = logging.getLogger(__name__) 

15 

16# Handles CORS preflight (OPTIONS) without authentication, matching original Flask before_request behavior 

17user_options_router = APIRouter() 

18 

19 

20@user_options_router.options('/{path:path}') 

21@user_options_router.options('') 

22def _user_options_handler(): 

23 from fastapi.responses import Response 

24 return Response(status_code=200) 

25 

26 

27# All user-management endpoints require admin access 

28user_router = APIRouter(dependencies=[identity_verify(0)]) 

29 

30 

31@user_router.get('') 

32def get_user_list(query: GetUserListQuery = Depends()): 

33 offset = query.offset 

34 count = query.count 

35 course = query.course 

36 role = query.role 

37 try: 

38 if offset is not None: 

39 offset = int(offset) 

40 if count is not None: 

41 count = int(count) 

42 if role is not None: 

43 role = int(role) 

44 except (TypeError, ValueError): 

45 return HTTPError('offset, count and role must be integer', 400) 

46 if course is not None: 

47 course = parse.unquote(course) 

48 

49 filter_query = drop_none({'courses': course, 'role': role}) 

50 user_list = engine.User.objects(**filter_query) 

51 if offset is not None: 

52 user_list = user_list[offset:] 

53 if count is not None: 

54 user_list = user_list[:count] 

55 

56 user_list = [User(u).info for u in user_list] 

57 return HTTPResponse(data=user_list) 

58 

59 

60@user_router.get('/summary') 

61def get_user_summary(): 

62 user_count = engine.User.objects.count() 

63 breakdown = [{ 

64 "role": role.name.lower(), 

65 "count": engine.User.objects(role=role.value).count() 

66 } for role in engine.User.Role] 

67 return HTTPResponse(data={"userCount": user_count, "breakdown": breakdown}) 

68 

69 

70@user_router.post('') 

71def add_user(body: AddUserBody): 

72 '''Directly add a user without activation required. Admin only.''' 

73 try: 

74 User.signup(body.username, body.password, body.email).activate() 

75 except ValidationError as ve: 

76 return HTTPError('Signup Failed', 400, data=ve.to_dict()) 

77 except NotUniqueError: 

78 return HTTPError('User Exists', 400) 

79 except ValueError: 

80 return HTTPError('Not Allowed Name', 400) 

81 return HTTPResponse() 

82 

83 

84@user_router.patch('/{username}') 

85def update_user( 

86 body: UpdateUserBody, 

87 user: User = Depends(login_required), 

88 target_user: User = get_doc('username', User), 

89): 

90 if body.password is not None: 

91 target_user.change_password(body.password) 

92 logger.info('admin changed user password ' 

93 f'[actor={user.username}, user={target_user.username}]') 

94 payload = drop_none({ 

95 'profile__displayed_name': body.displayed_name, 

96 'role': body.role, 

97 }) 

98 if len(payload): 

99 fields = [*payload.keys()] 

100 target_user.update(**payload) 

101 logger.info( 

102 'admin changed user info ' 

103 f'[actor={user.username}, user={target_user.username}, fields={fields}]' 

104 ) 

105 return HTTPResponse()