Coverage for tests/model_tests/test_employers.py: 99%
134 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-05-05 14:02 +0000
« prev ^ index » next coverage.py v7.8.0, created at 2025-05-05 14:02 +0000
1"""This module contains tests for the Employers model."""
3import os
4import uuid
5import sys
6from io import BytesIO
7import pytest
8import pandas as pd
9from dotenv import load_dotenv
10from core import shared
11from core.database_mongo_manager import DatabaseMongoManager
13sys.path.append(
14 os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
15)
17os.environ["IS_TEST"] = "True"
18load_dotenv()
21@pytest.fixture()
22def app():
23 """Fixture to create a test client."""
24 from ...app import app # pylint: disable=import-outside-toplevel
26 return app
29@pytest.fixture()
30def database():
31 """Fixture to create a test database."""
33 database = DatabaseMongoManager(
34 shared.getenv("MONGO_URI"),
35 shared.getenv("MONGO_DB_TEST", "cs3528_testing"),
36 )
37 employers = database.get_all("employers")
38 database.delete_all("employers")
39 yield database
41 database.delete_all("employers")
43 for employer in employers:
44 database.insert("employers", employer)
45 database.delete_all_by_field("_id", "company_name", "email")
46 # Cleanup code
47 database.connection.close()
50@pytest.fixture()
51def employer_model():
52 """Fixture to create a Employer model instance."""
53 from employers.models import Employers # pylint: disable=import-outside-toplevel
55 return Employers()
58@pytest.fixture()
59def sample_employer(database):
60 """Fixture to create a sample employer."""
61 yield {
62 "_id": uuid.uuid4().hex,
63 "company_name": "TechCorp",
64 "email": "contact@techcorp.com",
65 }
66 database.delete_all_by_field("employers", "company_name", "TechCorp")
69def test_register_employer(database, employer_model, app):
70 """Test registering an employer."""
71 employer = {
72 "_id": uuid.uuid4().hex,
73 "company_name": "TechCorp",
74 "email": "contact@techcorp.com",
75 }
76 database.delete_all_by_field("employers", "email", "contact@techcorp.com")
77 with app.app_context():
78 response = employer_model.register_employer(employer)
79 assert response[1] == 200
80 assert (
81 database.get_one_by_field("employers", "email", "contact@techcorp.com")
82 is not None
83 )
86def test_register_existing_email(database, employer_model, app):
87 """Test registering an employer with an existing email."""
88 employer = {
89 "_id": uuid.uuid4().hex,
90 "company_name": "TechCorp",
91 "email": "contact@techcorp.com",
92 }
93 database.insert("employers", employer)
94 with app.app_context():
95 response = employer_model.register_employer(employer)
96 assert response[1] == 400
97 assert response[0].json["error"] == "Email already in use"
100def test_get_employer_by_id(database, employer_model, app):
101 """Test retrieving an employer by ID."""
102 employer = {
103 "_id": uuid.uuid4().hex,
104 "company_name": "TechCorp",
105 "email": "contact@techcorp.com",
106 }
107 database.insert("employers", employer)
108 with app.app_context():
109 retrieved_employer = employer_model.get_employer_by_id(employer["_id"])
110 assert retrieved_employer is not None
111 assert retrieved_employer["company_name"] == "TechCorp"
114def test_delete_employer(database, employer_model, app):
115 """Test deleting an employer."""
116 employer = {
117 "_id": uuid.uuid4().hex,
118 "company_name": "TechCorp",
119 "email": "contact@techcorp.com",
120 }
121 database.insert("employers", employer)
122 with app.app_context():
123 response = employer_model.delete_employer_by_id(employer["_id"])
124 assert response[1] == 200
125 assert database.get_one_by_id("employers", employer["_id"]) is None
128def test_download_all_employers(database, employer_model, app):
129 """Test downloading all employers."""
130 employer = {
131 "_id": uuid.uuid4().hex,
132 "company_name": "TechCorp",
133 "email": "contact@techcorp.com",
134 }
135 database.insert("employers", employer)
137 with app.app_context():
138 with app.test_request_context(): # Ensure proper request context
139 response = employer_model.download_all_employers()
140 assert response.status_code == 200
141 assert (
142 response.mimetype
143 == "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
144 )
147def test_upload_employers(database, employer_model, app):
148 """Test uploading employers from an Excel file."""
149 df = pd.DataFrame([{"Company_name": "TechCorp", "Email": "contact@techcorp.com"}])
150 excel_buffer = BytesIO()
151 df.to_excel(excel_buffer, index=False)
152 excel_buffer.seek(0)
153 excel_buffer.name = "employers.xlsx"
154 with app.app_context():
155 response = employer_model.upload_employers(excel_buffer)
156 assert response[1] == 200
157 assert (
158 database.get_one_by_field("employers", "email", "contact@techcorp.com")
159 is not None
160 )
163def test_get_company_name(database, employer_model, app, sample_employer):
164 """Test getting company name by ID."""
165 database.insert("employers", sample_employer)
166 with app.app_context():
167 company_name = employer_model.get_company_name(sample_employer["_id"])
168 assert company_name == "TechCorp"
171def test_employer_login(database, employer_model, app, sample_employer):
172 """Test employer login."""
173 database.insert("employers", sample_employer)
174 with app.app_context():
175 with app.test_request_context():
176 response = employer_model.employer_login(sample_employer["email"])
177 assert response[1] == 200
178 assert response[0].json["message"] == "OTP sent"
181def test_employer_login_nonexistant(database, employer_model, app, sample_employer):
182 """Test employer login."""
183 with app.app_context():
184 with app.test_request_context():
185 response = employer_model.employer_login(sample_employer["email"])
186 assert response[1] == 404
187 assert response[0].json["error"] == "Email not found"
190def test_get_employers(database, employer_model, app, sample_employer):
191 """Test getting all employers."""
192 database.insert("employers", sample_employer)
193 with app.app_context():
194 employers = employer_model.get_employers()
195 assert len(employers) > 0
196 assert employers[0]["company_name"] == "TechCorp"
199def test_update_employer(database, employer_model, app, sample_employer):
200 """Test updating an employer."""
201 database.insert("employers", sample_employer)
202 update_data = {"company_name": "TechCorp Updated"}
203 with app.app_context():
204 response = employer_model.update_employer(sample_employer["_id"], update_data)
205 assert response[1] == 200
206 updated_employer = database.get_one_by_id("employers", sample_employer["_id"])
207 assert updated_employer["company_name"] == "TechCorp Updated"
210def test_rank_preferences(database, employer_model, app):
211 """Test ranking preferences for an opportunity."""
212 opportunity = {
213 "_id": uuid.uuid4().hex,
214 "employer_id": "E101",
215 "title": "Internship",
216 "preferences": [],
217 }
218 database.insert("opportunities", opportunity)
219 preferences = ["Preference1", "Preference2"]
220 with app.app_context():
221 response = employer_model.rank_preferences(opportunity["_id"], preferences)
222 assert response[1] == 200
223 updated_opportunity = database.get_one_by_id(
224 "opportunities", opportunity["_id"]
225 )
226 assert updated_opportunity["preferences"] == preferences
229def test_get_company_email_by_id(database, employer_model, app, sample_employer):
230 """Test getting company email by ID."""
231 database.insert("employers", sample_employer)
232 with app.app_context():
233 email = employer_model.get_company_email_by_id(sample_employer["_id"])
234 assert email == "contact@techcorp.com"
237def test_delete_all_employers(database, employer_model, app):
238 """Test deleting all employers."""
239 employer = {
240 "_id": uuid.uuid4().hex,
241 "company_name": "TechCorp",
242 "email": "contact@techcorp.com",
243 }
244 database.insert("employers", employer)
245 with app.app_context():
246 response = employer_model.delete_all_employers()
247 assert response[1] == 200
248 assert database.get_all("employers") == []