Coverage for core/models/user.py: 100.00%

1import pytz 

2 

3from django.contrib.auth.models import User as DjangoUser 

4from django.db import models, transaction 

5from phone_field import PhoneField 

6 

7# DO NOT IMPORT OTHER TRACKER APP MODELS HERE, IT WILL CAUSE A CIRCULAR IMPORT SINCE ALL/MOST MODELS IMPORT CORE.COREUSER and/or CORE.COREMODEL 

8# Import directly in helper functions instead to lazy-load it - See CoreUser.list_projects() for an example 

9from . import core as core_models 

10 

11 

12TIMEZONE_CHOICES = tuple((tz, tz) for tz in pytz.all_timezones) 

13 

14 

15class CoreUserData(core_models.CoreModel): 

16 """ 

17 Demographic information about a user. 

18 """ 

19 

20 name_prefix = models.CharField(max_length=255, blank=True, null=True, default="") 

21 first_name = models.CharField(max_length=255, blank=True, null=True, default="") 

22 middle_name = models.CharField(max_length=255, blank=True, null=True, default="") 

23 last_name = models.CharField(max_length=255, blank=True, null=True, default="") 

24 name_suffix = models.CharField(max_length=255, blank=True, null=True, default="") 

25 email = models.EmailField(max_length=255) 

26 secondary_email = models.EmailField(max_length=255, blank=True, null=True, default="") 

27 home_phone = PhoneField(blank=True, null=True) 

28 mobile_phone = PhoneField(blank=True, null=True) 

29 work_phone = PhoneField(blank=True, null=True) 

30 address_line_1 = models.CharField(max_length=255, blank=True, null=True, default="") 

31 address_line_2 = models.CharField(max_length=255, blank=True, null=True, default="") 

32 postal_code = models.CharField(max_length=255, blank=True, null=True, default="") 

33 city = models.CharField(max_length=255, blank=True, null=True, default="") 

34 state = models.CharField(max_length=255, blank=True, null=True, default="") 

35 country = models.CharField(max_length=255, blank=True, null=True, default="") 

36 timezone = models.CharField(max_length=255, default='UTC', choices=TIMEZONE_CHOICES) 

37 

38 

39class CoreUserActiveManager(models.Manager): 

40 """ 

41 Active CoreUsers are ones that are not deleted. 

42 """ 

43 

44 def get_queryset(self): 

45 return super().get_queryset().filter(deleted=None) 

46 

47 

48class CoreUserManager(core_models.CoreModelManager): 

49 """ 

50 General helper methods for managing CoreUsers, active or deleted. 

51 """ 

52 

53 def get_or_create_api_user(self): 

54 """ 

55 Creates an API user if it does not exist, primarily for Web tasks. 

56 

57 Returns: 

58 api_user (CoreUser): The API user. 

59 """ 

60 

61 try: 

62 api_user = CoreUser.objects.get(pk='75af4764-0f94-49f2-a6dc-3dbfe1b577f9') 

63 except CoreUser.DoesNotExist: 

64 with transaction.atomic(): 

65 api_user = CoreUser( 

66 id='75af4764-0f94-49f2-a6dc-3dbfe1b577f9', 

67 created_by_id='75af4764-0f94-49f2-a6dc-3dbfe1b577f9', 

68 ) 

69 api_user.save() 

70 api_user_data = CoreUserData( 

71 id='373f414f-9692-4e5c-92f2-5781dbad5c04', 

72 created_by_id='75af4764-0f94-49f2-a6dc-3dbfe1b577f9', 

73 first_name='API', 

74 last_name='USER', 

75 address_line_1='', 

76 address_line_2='', 

77 city='', 

78 state='', 

79 country='', 

80 postal_code=0, 

81 ) 

82 api_user_data.save() 

83 api_user.current = api_user_data 

84 api_user.save() 

85 

86 return api_user 

87 

88 def get_or_create_system_user(self): 

89 """ 

90 Creates a system user if it does not exist, primarily for setup and system tasks. 

91 

92 Returns: 

93 system_user (CoreUser): The system user. 

94 """ 

95 

96 try: 

97 system_user = CoreUser.objects.get(pk='45407f07-21e9-42ba-8c39-03b57767fe76') 

98 except CoreUser.DoesNotExist: 

99 with transaction.atomic(): 

100 system_user = CoreUser( 

101 id='45407f07-21e9-42ba-8c39-03b57767fe76', 

102 created_by_id='45407f07-21e9-42ba-8c39-03b57767fe76', 

103 ) 

104 system_user.save() 

105 system_user_data = CoreUserData( 

106 id='02e94188-5b8e-494a-922c-bc6ed2ffcfc4', 

107 created_by_id='45407f07-21e9-42ba-8c39-03b57767fe76', 

108 first_name='SYSTEM', 

109 last_name='USER', 

110 address_line_1='', 

111 address_line_2='', 

112 city='', 

113 state='', 

114 country='', 

115 postal_code=0, 

116 ) 

117 system_user_data.save() 

118 system_user.current = system_user_data 

119 system_user.save() 

120 

121 return system_user 

122 

123 def create_core_user_from_web(self, request_data: dict) -> 'CoreUser': 

124 """ 

125 Takes a flat dictionary and creates a CoreUser and CoreUserData object. 

126 

127 Args: 

128 request_data (dict): Probably a JSON payload from a POST request. 

129 

130 Returns: 

131 new_user (CoreUser): The new user that was created. 

132 """ 

133 

134 with transaction.atomic(): 

135 api_user = CoreUser.objects.get_or_create_api_user() 

136 

137 django_user = DjangoUser.objects.create_user( 

138 username=request_data.get('email'), 

139 email=request_data.get('email'), 

140 password=request_data.get('password') 

141 ) 

142 

143 core_user_data = CoreUserData( 

144 created_by_id=api_user.id, 

145 first_name=request_data.get('first_name', ''), 

146 last_name=request_data.get('last_name', ''), 

147 email=request_data.get('email'), 

148 secondary_email=request_data.get('secondary_email', ''), 

149 home_phone=request_data.get('home_phone', ''), 

150 mobile_phone=request_data.get('mobile_phone', ''), 

151 work_phone=request_data.get('work_phone', ''), 

152 address_line_1=request_data.get('address_line_1', ''), 

153 address_line_2=request_data.get('address_line_2', ''), 

154 postal_code=request_data.get('postal_code', ''), 

155 city=request_data.get('city', ''), 

156 state=request_data.get('state', ''), 

157 country=request_data.get('country', ''), 

158 timezone=request_data.get('timezone', ''), 

159 ) 

160 core_user_data.save() 

161 

162 new_user = CoreUser( 

163 created_by_id=api_user.id, 

164 current=core_user_data, 

165 user=django_user 

166 ) 

167 new_user.save() 

168 

169 return new_user 

170 

171 

172class CoreUser(core_models.CoreModel, core_models.CoreModelActiveManager, core_models.CoreModelManager): 

173 """ 

174 A user of the system. The _real_ inforamation about a user is stored in `current` as CoreUserData. 

175 Every time a user is updated, a new CoreUserData object is created and the `current` field is updated to reflect the new data. 

176 The `user` field is a Django User object that is used for authentication and authorization. 

177 The `current` field is a ForeignKey to CoreUserData, which contains the user's demographic information. 

178 """ 

179 

180 class Meta: 

181 ordering = ['current__last_name', 'current__first_name', 'current__email'] 

182 

183 active_objects = CoreUserActiveManager() 

184 objects = CoreUserManager() 

185 

186 current = models.ForeignKey(CoreUserData, on_delete=models.CASCADE, blank=True, null=True) 

187 user = models.OneToOneField(DjangoUser, on_delete=models.CASCADE, blank=True, null=True, related_name='django_user') 

188 

189 def deactivate_login(self) -> None: 

190 """ 

191 Deactivates a user's login and disaables login to the webapp/api. 

192 """ 

193 

194 self.user.is_active = False 

195 self.user.save() 

196 

197 def list_projects(self): 

198 """ 

199 Get all projects the user is a member of or owns. 

200 

201 Returns: 

202 projects (list): The projects the user is a member of or owns. 

203 """ 

204 

205 from project.models.project import Project 

206 

207 # Get projects from organization memberships 

208 project_ids = set(self.organizationmembers_set.values_list('projects', flat=True).exclude(projects__isnull=True)) 

209 # Get projects the user owns 

210 project_ids.update(self.project_set.values_list('id', flat=True)) 

211 # Always query active_objects to exclude deleted items 

212 projects = Project.active_objects.filter(id__in=project_ids) 

213 

214 return projects 

215 

216 def list_git_repositories(self): 

217 """ 

218 A helper method to get all git repositories the user can see or owns. 

219 

220 Returns: 

221 repositories (list): All git repositories the user can see. 

222 """ 

223 

224 from project.models.project import Project 

225 from project.models.git_repository import GitRepository 

226 

227 # Get repositories from organizations 

228 repository_ids = set(self.organizationmembers_set.values_list('git_repositories', flat=True).exclude(git_repositories__isnull=True)) 

229 # Get repositories from personal projects 

230 projects = self.project_set.values_list('id', flat=True) 

231 # Always query active_objects to exclude deleted items 

232 repository_ids.update(Project.active_objects.filter(id__in=projects).values_list('git_repositories', flat=True)) 

233 # Get any personal repositories not attached to projects or organizations 

234 repository_ids.update(GitRepository.active_objects.filter(created_by=self).values_list('id', flat=True)) 

235 repositories = GitRepository.active_objects.filter(id__in=repository_ids) 

236 

237 return repositories 

238 

239 def list_issues(self): 

240 """ 

241 Get all issues the user is watching, assigned to, etc. 

242 

243 Returns: 

244 issues (list): The issues the user is watching, assigned to, etc. 

245 """ 

246 

247 from project.models.issue import Issue 

248 

249 # TODO: Add issues watching, assigned to, etc. 

250 issue_ids = set(self.issue_created_by.values_list('id', flat=True)) 

251 user_projects = self.list_projects() 

252 for project in user_projects: 

253 issue_ids.update(project.list_issues().values_list('id', flat=True)) 

254 # Always query active_objects to exclude deleted items 

255 issues = Issue.active_objects.filter(id__in=issue_ids) 

256 return issues 

257 

258 def list_organizations(self): 

259 """ 

260 A helper method to get all organiations the user is a member of, etc. 

261 

262 Returns: 

263 organizations (list): The organizations the user is a member of, etc. 

264 """ 

265 

266 return self.organizationmembers_set.all() 

267 

268 def list_users(self): 

269 """ 

270 Lists the users a user can see in the system, whether from other projects or organizations. 

271 

272 Returns: 

273 users (list): The users the user can see to add to projects. 

274 """ 

275 

276 users_ids_set = set() 

277 

278 for project in self.list_projects(): 

279 users_ids_set.update(project.users.all().values_list('id', flat=True)) 

280 

281 for organization in self.list_organizations(): 

282 users_ids_set.update(organization.members.all().values_list('id', flat=True)) 

283 

284 # Always query active_objects to exclude deleted items 

285 return CoreUser.active_objects.filter(id__in=users_ids_set) 

286 

287 

288 def __str__(self): 

289 potential_names = [] 

290 if self.current.first_name: 

291 potential_names.append(self.current.first_name) 

292 if self.current.last_name: 

293 potential_names.append(self.current.last_name) 

294 potential_names.append(f"({self.current.email})") 

295 return ' '.join(potential_names) 

296 

297 

298class UserLogin(core_models.CoreModel): 

299 """ 

300 A record of a user's login to the webapp or API. 

301 """ 

302 

303 class Meta: 

304 ordering = ['-login_time'] 

305 

306 user = models.ForeignKey(CoreUser, on_delete=models.CASCADE) 

307 login_time = models.DateTimeField(auto_now_add=True) 

308 x_forwarded_for = models.GenericIPAddressField(blank=True, null=True) 

309 remote_addr = models.GenericIPAddressField(blank=True, null=True) 

310 user_agent = models.CharField(max_length=255, blank=True, null=True, default="") 

311 session_key = models.CharField(max_length=255, blank=True, null=True, default="") 

312 

313 

314class UserLogout(core_models.CoreModel): 

315 """ 

316 A record of a user's logout from the webapp or API. 

317 """ 

318 

319 class Meta: 

320 ordering = ['-logout_time'] 

321 

322 user = models.ForeignKey(CoreUser, on_delete=models.CASCADE) 

323 logout_time = models.DateTimeField(auto_now_add=True) 

324 x_forwarded_for = models.GenericIPAddressField(blank=True, null=True) 

325 remote_addr = models.GenericIPAddressField(blank=True, null=True) 

326 user_agent = models.CharField(max_length=255, blank=True, null=True, default="") 

327 session_key = models.CharField(max_length=255, blank=True, null=True, default="")