Coverage for backend/tests/utils/create_data.py: 94%
109 statements
« prev ^ index » next coverage.py v7.10.7, created at 2025-09-23 17:56 +0000
1"""Functions for creating test data in the database"""
3import copy
4import random
6import app.eis.models as eis_models
7from app import models, utils
8from tests.utils.table_data import (
9 USER_DATA,
10 SETTINGS_DATA,
11 COMPANY_DATA,
12 LOCATION_DATA,
13 AGGREGATOR_DATA,
14 KEYWORD_DATA,
15 PERSON_DATA,
16 JOB_DATA,
17 add_mappings,
18 JOB_KEYWORD_MAPPINGS,
19 JOB_CONTACT_MAPPINGS,
20 FILE_DATA,
21 INTERVIEW_DATA,
22 INTERVIEW_INTERVIEWER_MAPPINGS,
23 JOB_ALERT_EMAIL_DATA,
24 JOB_SCRAPED_DATA,
25 EMAIL_SCRAPEDJOB_MAPPINGS,
26 SERVICE_LOG_DATA,
27 JOB_APPLICATION_UPDATE_DATA,
28)
31def create_settings(db) -> list[models.Setting]:
32 """Create sample settings"""
34 print("Creating settings...")
35 # noinspection PyArgumentList
36 settings = [models.Setting(**data) for data in SETTINGS_DATA]
37 db.add_all(settings)
38 db.commit()
39 return db.query(models.Setting).all()
42def create_users(db, user_data: list[dict] = None) -> list[models.User]:
43 """Create sample users and return them with non-hashed passwords but with database IDs"""
45 print("Creating users...")
46 users_hash = []
47 original_passwords = []
48 if not user_data:
49 user_data = USER_DATA
51 # Store the original password and hash it for database storage
52 for user in user_data:
53 user_dict = user.copy()
54 original_passwords.append(user_dict["password"]) # Store original password
55 user_dict["password"] = utils.hash_password(user_dict["password"])
56 # noinspection PyArgumentList
57 users_hash.append(models.User(**user_dict))
59 users_hash = add_to_db(db, users_hash)
61 # Create new user objects that won't become detached with the original passwords and database attributes
62 result_users = []
63 for i, user in enumerate(users_hash):
64 user_dict = {key: value for key, value in user.__dict__.items() if key != "_sa_instance_state"}
65 user_dict["password"] = original_passwords[i]
66 # noinspection PyArgumentList
67 new_user = models.User(**user_dict)
68 result_users.append(new_user)
70 return result_users
73def delete_user(db, user_email: str) -> None:
74 """Delete a user by email and return the deleted user
75 :param db: database session
76 :param user_email: user email address"""
78 user = db.query(models.User).filter(models.User.email == user_email).first()
79 if user:
80 db.delete(user)
81 db.commit()
84def override_entries_properties(data: list[dict], *args) -> list[dict]:
85 """Override the owner_id in a list of dictionaries
86 :param data: list of model entries to override
87 :param args: tuples of (key to override, list of models to get new IDs from)"""
89 data = copy.deepcopy(data)
90 for entry in data:
91 for arg in args:
92 key, values = arg
93 if entry.get(key, None) is not None:
94 try:
95 current_id = entry[key] - 1
96 new_id = values[current_id].id
97 entry[key] = new_id
98 except IndexError:
99 data.remove(entry)
100 break
101 return data
104def add_to_db(db, items: list[models.CommonBase]) -> list:
105 """Add a list of items to the database and commit"""
107 db.add_all(items)
108 db.commit()
109 for item in items:
110 db.refresh(item)
111 return items
114def create_keywords(db, users: list[models.User]) -> list[models.Keyword]:
115 """Create sample keywords"""
117 print("Creating keywords...")
118 # noinspection PyArgumentList
119 keywords = [
120 models.Keyword(**keyword)
121 for keyword in override_entries_properties(
122 KEYWORD_DATA,
123 ("owner_id", users),
124 )
125 ]
126 return add_to_db(db, keywords)
129def create_aggregators(db, users: list[models.User]) -> list[models.Aggregator]:
130 """Create sample aggregators"""
132 print("Creating aggregators...")
133 # noinspection PyArgumentList
134 aggregators = [
135 models.Aggregator(**aggregator)
136 for aggregator in override_entries_properties(
137 AGGREGATOR_DATA,
138 ("owner_id", users),
139 )
140 ]
141 return add_to_db(db, aggregators)
144def create_companies(db, users: list[models.User]) -> list[models.Company]:
145 """Create sample companies"""
147 print("Creating companies...")
148 # noinspection PyArgumentList
149 companies = [
150 models.Company(**company)
151 for company in override_entries_properties(
152 COMPANY_DATA,
153 ("owner_id", users),
154 )
155 ]
156 return add_to_db(db, companies)
159def create_locations(db, users: list[models.User]) -> list[models.Location]:
160 """Create sample locations"""
162 print("Creating locations...")
163 # noinspection PyArgumentList
164 locations = [
165 models.Location(**location)
166 for location in override_entries_properties(
167 LOCATION_DATA,
168 ("owner_id", users),
169 )
170 ]
171 return add_to_db(db, locations)
174def create_people(db, users: list[models.User], companies: list[models.Company]) -> list[models.Person]:
175 """Create sample people"""
177 print("Creating people...")
178 # noinspection PyArgumentList
179 persons = [
180 models.Person(**person)
181 for person in override_entries_properties(
182 PERSON_DATA,
183 ("owner_id", users),
184 ("company_id", companies),
185 )
186 ]
187 return add_to_db(db, persons)
190def create_jobs(
191 db,
192 keywords,
193 persons,
194 users: list[models.User],
195 companies: list[models.Company],
196 locations: list[models.Location],
197 aggregators: list[models.Aggregator],
198 files: list[models.File],
199) -> list[models.Job]:
200 """Create sample jobs"""
202 print("Creating jobs...")
203 # noinspection PyArgumentList
204 jobs = [
205 models.Job(**job)
206 for job in override_entries_properties(
207 JOB_DATA,
208 ("owner_id", users),
209 ("company_id", companies),
210 ("location_id", locations),
211 ("source_id", aggregators),
212 ("application_aggregator_id", aggregators),
213 ("cv_id", files),
214 ("cover_letter_id", files),
215 )
216 ]
218 # Add keywords to jobs
219 add_mappings(
220 primary_data=jobs,
221 secondary_data=keywords,
222 mapping_data=JOB_KEYWORD_MAPPINGS,
223 primary_key="job_id",
224 secondary_key="keyword_ids",
225 relationship_attr="keywords",
226 )
228 # Add contacts to jobs
229 add_mappings(
230 primary_data=jobs,
231 secondary_data=persons,
232 mapping_data=JOB_CONTACT_MAPPINGS,
233 primary_key="job_id",
234 secondary_key="person_ids",
235 relationship_attr="contacts",
236 )
238 return add_to_db(db, jobs)
241def create_files(db, users: list[models.User]) -> list[models.File]:
242 """Create sample files (CVs and cover letters)"""
244 print("Creating files...")
245 # noinspection PyArgumentList
246 files = [models.File(**file) for file in override_entries_properties(FILE_DATA, ("owner_id", users))]
247 return add_to_db(db, files)
250def create_interviews(
251 db,
252 persons,
253 users: list[models.User],
254 locations: list[models.Location],
255 jobs: list[models.Job],
256) -> list[models.Interview]:
257 """Create sample interviews"""
259 print("Creating interviews...")
260 # noinspection PyArgumentList
261 interviews = [
262 models.Interview(**interview)
263 for interview in override_entries_properties(
264 INTERVIEW_DATA,
265 ("owner_id", users),
266 ("location_id", locations),
267 ("job_id", jobs),
268 )
269 ]
271 # Add interviewers to interviews
272 add_mappings(
273 primary_data=interviews,
274 secondary_data=persons,
275 mapping_data=INTERVIEW_INTERVIEWER_MAPPINGS,
276 primary_key="interview_id",
277 secondary_key="person_ids",
278 relationship_attr="interviewers",
279 )
281 return add_to_db(db, interviews)
284def create_job_application_updates(
285 db,
286 users: list[models.User],
287 jobs: list[models.Job],
288) -> list[models.JobApplicationUpdate]:
289 """Create sample job application updates"""
291 print("Creating job application updates...")
292 # noinspection PyArgumentList
293 updates = [
294 models.JobApplicationUpdate(**update)
295 for update in override_entries_properties(
296 JOB_APPLICATION_UPDATE_DATA,
297 ("owner_id", users),
298 ("job_id", jobs),
299 )
300 ]
302 return add_to_db(db, updates)
305def create_job_alert_emails(
306 db, users: list[models.User], service_logs: list[eis_models.EisServiceLog]
307) -> list[eis_models.JobAlertEmail]:
308 """Create sample job alert emails"""
310 print("Creating job alert emails...")
311 # noinspection PyArgumentList
312 emails = [
313 eis_models.JobAlertEmail(**email)
314 for email in override_entries_properties(
315 JOB_ALERT_EMAIL_DATA,
316 ("owner_id", users),
317 ("service_log_id", service_logs),
318 )
319 ]
320 for email in emails:
321 email.external_email_id += str(random.random()) # Ensure uniqueness
323 return add_to_db(db, emails)
326def create_scraped_jobs(db, emails, users: list[models.User]) -> list[eis_models.ScrapedJob]:
327 """Create sample scraped jobs - some with scraped data, some without"""
329 print("Creating scraped jobs...")
330 # noinspection PyArgumentList
331 scraped_jobs = [
332 eis_models.ScrapedJob(**job_data)
333 for job_data in override_entries_properties(
334 JOB_SCRAPED_DATA,
335 ("owner_id", users),
336 )
337 ]
339 # Add email mappings to scraped jobs
340 add_mappings(
341 primary_data=emails,
342 secondary_data=scraped_jobs,
343 mapping_data=EMAIL_SCRAPEDJOB_MAPPINGS,
344 primary_key="email_id",
345 secondary_key="scraped_job_ids",
346 relationship_attr="jobs",
347 )
349 return add_to_db(db, scraped_jobs)
352def create_service_logs(db) -> list[eis_models.EisServiceLog]:
353 """Create sample service logs"""
355 print("Creating service logs...")
356 # noinspection PyArgumentList
357 logs = [eis_models.EisServiceLog(**log) for log in SERVICE_LOG_DATA]
359 return add_to_db(db, logs)