Coverage for modules\team\services.py: 36%

1from django.db import transaction 

2from django.utils import timezone 

3from core.exceptions import NotFoundException, BadRequestException 

4from .models import Department, TeamMember, Invitation, AttendanceLog 

5from modules.authentication.models import User, Membership 

6 

7class TeamService: 

8 @staticmethod 

9 def get_or_create_today_attendance(tenant_id, user): 

10 now = timezone.now() 

11 record = AttendanceLog.objects.filter( 

12 tenant_id=tenant_id, 

13 user=user, 

14 check_in__date=now.date(), 

15 status='CLOCKED_IN', 

16 check_out__isnull=True, 

17 ).order_by('-check_in').first() 

18 

19 if record: 

20 return record 

21 

22 return AttendanceLog.objects.create( 

23 tenant_id=tenant_id, 

24 user=user, 

25 check_in=now, 

26 status='CLOCKED_IN', 

27 ) 

28 

29 @staticmethod 

30 def clock_in(tenant_id, user, notes=None): 

31 today_open = AttendanceLog.objects.filter( 

32 tenant_id=tenant_id, 

33 user=user, 

34 check_out__isnull=True, 

35 status='CLOCKED_IN', 

36 ).order_by('-check_in').first() 

37 if today_open: 

38 raise BadRequestException('User is already clocked in for this organization.') 

39 

40 return AttendanceLog.objects.create( 

41 tenant_id=tenant_id, 

42 user=user, 

43 check_in=timezone.now(), 

44 status='CLOCKED_IN', 

45 notes=notes, 

46 ) 

47 

48 @staticmethod 

49 def clock_out(tenant_id, user, notes=None): 

50 record = AttendanceLog.objects.filter( 

51 tenant_id=tenant_id, 

52 user=user, 

53 check_out__isnull=True, 

54 status='CLOCKED_IN', 

55 ).order_by('-check_in').first() 

56 if not record: 

57 raise BadRequestException('User is not currently clocked in.') 

58 

59 record.check_out = timezone.now() 

60 record.status = 'CLOCKED_OUT' 

61 if notes: 

62 record.notes = notes 

63 record.save(update_fields=['check_out', 'status', 'notes', 'updated_at']) 

64 return record 

65 

66 @staticmethod 

67 def list_attendance(tenant_id, user=None): 

68 qs = AttendanceLog.objects.filter(tenant_id=tenant_id) 

69 if user is not None: 

70 qs = qs.filter(user=user) 

71 return qs.select_related('user') 

72 

73 @staticmethod 

74 def list_visible_attendance(tenant_id, user): 

75 membership = Membership.objects.filter( 

76 organization_id=tenant_id, 

77 user=user, 

78 ).first() 

79 role = 'ADMIN' if user.is_superuser else (membership.role if membership else 'MEMBER') 

80 

81 if role in ('OWNER', 'ADMIN', 'HR'): 

82 return AttendanceLog.objects.filter(tenant_id=tenant_id).select_related('user') 

83 

84 if role == 'MANAGER': 

85 department_id = TeamMember.objects.filter( 

86 tenant_id=tenant_id, 

87 user=user, 

88 ).values_list('department_id', flat=True).first() 

89 if department_id: 

90 department_user_ids = TeamMember.objects.filter( 

91 tenant_id=tenant_id, 

92 department_id=department_id, 

93 ).values_list('user_id', flat=True) 

94 return AttendanceLog.objects.filter( 

95 tenant_id=tenant_id, 

96 user_id__in=department_user_ids, 

97 ).select_related('user') 

98 

99 return AttendanceLog.objects.filter( 

100 tenant_id=tenant_id, 

101 user=user, 

102 ).select_related('user') 

103 @staticmethod 

104 def list_departments(tenant_id): 

105 return Department.objects.filter(tenant_id=tenant_id).select_related('parent', 'manager') 

106 

107 @staticmethod 

108 def create_department(tenant_id, data): 

109 if Department.objects.filter(tenant_id=tenant_id, code=data.code).exists(): 

110 raise BadRequestException(f"Department code '{data.code}' already exists.") 

111 

112 manager = None 

113 if data.manager_id: 

114 manager = User.objects.filter(id=data.manager_id).first() 

115 

116 parent = None 

117 if data.parent_id: 

118 parent = Department.objects.filter(id=data.parent_id, tenant_id=tenant_id).first() 

119 

120 return Department.objects.create( 

121 tenant_id=tenant_id, 

122 name=data.name, 

123 code=data.code, 

124 description=data.description, 

125 parent=parent, 

126 manager=manager 

127 ) 

128 

129 @staticmethod 

130 def list_members(tenant_id): 

131 return TeamMember.objects.filter(tenant_id=tenant_id).select_related('user', 'department') 

132 

133 @staticmethod 

134 @transaction.atomic 

135 def create_local_user(tenant_id, data): 

136 if User.objects.filter(email__iexact=data.local_email).exists(): 

137 raise BadRequestException('A user with this local email already exists.') 

138 

139 department = None 

140 if data.department_id: 

141 department = Department.objects.filter(id=data.department_id, tenant_id=tenant_id).first() 

142 if not department: 

143 raise BadRequestException('Department not found.') 

144 

145 user = User.objects.create_user( 

146 email=data.local_email, 

147 password=data.password or None, 

148 first_name=data.first_name, 

149 last_name=data.last_name, 

150 external_email=data.external_email, 

151 ) 

152 Membership.objects.create( 

153 user=user, 

154 organization_id=tenant_id, 

155 role=data.role, 

156 is_default=True, 

157 ) 

158 return TeamMember.objects.create( 

159 tenant_id=tenant_id, 

160 user=user, 

161 department=department, 

162 job_title=data.job_title, 

163 role=data.role, 

164 ) 

165 

166 @staticmethod 

167 @transaction.atomic 

168 def update_member(tenant_id, member_id, data): 

169 member = TeamMember.objects.filter(id=member_id, tenant_id=tenant_id).first() 

170 if not member: 

171 raise NotFoundException("Team member not found.") 

172 

173 if data.department_id is not None: 

174 dept = Department.objects.filter(id=data.department_id, tenant_id=tenant_id).first() if data.department_id else None 

175 member.department = dept 

176 

177 if data.job_title is not None: 

178 member.job_title = data.job_title 

179 

180 if data.role is not None: 

181 member.role = data.role 

182 # Sync membership 

183 Membership.objects.filter(user=member.user, organization_id=tenant_id).update(role=data.role) 

184 

185 if data.status is not None: 

186 member.status = data.status 

187 

188 member.save() 

189 return member 

190 

191 @staticmethod 

192 def create_invitation(tenant_id, invited_by, data): 

193 return Invitation.objects.create( 

194 tenant_id=tenant_id, 

195 email=data.email, 

196 role=data.role, 

197 department_id=data.department_id, 

198 invited_by=invited_by 

199 )