Coverage for core/services/user_service.py: 31.82%

1""" 

2用户服务层 

3""" 

4 

5from typing import List, Optional 

6 

7from core.models.user import (AccountCreate, User, UserCreate, UserResponse, 

8 UserUpdate) 

9from core.repositories.user_repository import UserRepository 

10from core.services.auth_service import AuthService 

11 

12 

13class UserService: 

14 """用户服务""" 

15 

16 def __init__(self): 

17 self.user_repository = UserRepository() 

18 self.auth_service = AuthService() 

19 

20 def create_user(self, user_data: UserCreate, current_user: User) -> Optional[User]: 

21 """创建用户（仅管理员可操作）""" 

22 if current_user.user_type != "admin": 

23 return None 

24 

25 # 检查用户名是否已存在 

26 existing_user = self.user_repository.get_user_by_username(user_data.username) 

27 if existing_user: 

28 return None 

29 

30 # 创建用户 

31 user = self.user_repository.create_user(user_data, self.auth_service) 

32 

33 # 创建默认账户 

34 account_data = AccountCreate( 

35 user_id=user.id, 

36 account_type="demo", 

37 balance=10000.0, 

38 available_balance=10000.0, 

39 currency="CNY", 

40 ) 

41 self.user_repository.create_account(account_data) 

42 

43 return user 

44 

45 def get_all_users(self, current_user: User) -> Optional[List[User]]: 

46 """获取所有用户（仅管理员可操作）""" 

47 if current_user.user_type != "admin": 

48 return None 

49 

50 return self.user_repository.get_all_users() 

51 

52 def get_user_by_id(self, user_id: str, current_user: User) -> Optional[User]: 

53 """获取用户信息""" 

54 # 用户只能查看自己的信息，管理员可以查看所有用户信息 

55 if current_user.user_type != "admin" and current_user.id != user_id: 

56 return None 

57 

58 return self.user_repository.get_user_by_id(user_id) 

59 

60 def update_user( 

61 self, user_id: str, user_data: UserUpdate, current_user: User 

62 ) -> Optional[User]: 

63 """更新用户信息""" 

64 # 用户只能更新自己的信息，管理员可以更新所有用户信息 

65 if current_user.user_type != "admin" and current_user.id != user_id: 

66 return None 

67 

68 # 普通用户不能修改用户类型 

69 if current_user.user_type != "admin" and user_data.user_type: 

70 user_data.user_type = None 

71 

72 # 如果密码为空或None，则不更新密码 

73 if not user_data.password or user_data.password.strip() == "": 

74 user_data.password = None 

75 

76 return self.user_repository.update_user(user_id, user_data, self.auth_service) 

77 

78 def delete_user(self, user_id: str, current_user: User) -> bool: 

79 """删除用户（仅管理员可操作）""" 

80 if current_user.user_type != "admin": 

81 return False 

82 

83 # 不能删除自己 

84 if current_user.id == user_id: 

85 return False 

86 

87 return self.user_repository.delete_user(user_id) 

88 

89 def get_user_accounts(self, user_id: str, current_user: User) -> Optional[List]: 

90 """获取用户账户信息""" 

91 # 用户只能查看自己的账户，管理员可以查看所有用户账户 

92 if current_user.user_type != "admin" and current_user.id != user_id: 

93 return None 

94 

95 return self.user_repository.get_user_accounts(user_id)