Coverage for tests/model_tests/test_courses.py: 99%

1"""Tests for the user routes.""" 

2 

3# pylint: disable=redefined-outer-name 

4# flake8: noqa: F811 

5 

6import os 

7import sys 

8import uuid 

9from io import BytesIO 

10import pytest 

11from dotenv import load_dotenv 

12import pandas as pd 

13 

14 

15# Add the root directory to the Python path 

16sys.path.append( 

17 os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) 

18) 

19 

20from core import shared 

21from core.database_mongo_manager import DatabaseMongoManager 

22 

23os.environ["IS_TEST"] = "True" 

24 

25load_dotenv() 

26 

27 

28@pytest.fixture() 

29def app(): 

30 """Fixture to create a test client.""" 

31 from ...app import app # pylint: disable=import-outside-toplevel 

32 

33 return app 

34 

35 

36@pytest.fixture() 

37def database(): 

38 """Fixture to create a test database.""" 

39 

40 database = DatabaseMongoManager( 

41 shared.getenv("MONGO_URI"), 

42 shared.getenv("MONGO_DB_TEST", "cs3528_testing"), 

43 ) 

44 

45 courses = database.get_all("courses") 

46 database.delete_all("courses") 

47 yield database 

48 

49 for course in courses: 

50 database.insert("courses", course) 

51 database.delete_all_by_field("users", "email", "dummy@dummy.com") 

52 database.delete_all_by_field("courses", "course_id", "CS101") 

53 # Cleanup code 

54 database.connection.close() 

55 

56 

57@pytest.fixture() 

58def course_model(): 

59 """Fixture to create a Course model instance.""" 

60 from courses.models import Course 

61 

62 return Course() 

63 

64 

65@pytest.fixture() 

66def sample_course(database): 

67 """Fixture to create a sample course.""" 

68 yield { 

69 "_id": uuid.uuid4().hex, 

70 "course_id": "CS101", 

71 "course_name": "Introduction to Computer Science", 

72 "course_description": "Basic concepts of computer science.", 

73 } 

74 database.delete_all_by_field("courses", "course_id", "CS101") 

75 

76 

77def test_add_course(database, course_model, sample_course, app): 

78 """Test adding a course.""" 

79 database.delete_all_by_field("courses", "course_id", "CS101") 

80 with app.app_context(): 

81 with app.test_request_context(): 

82 response = course_model.add_course(sample_course) 

83 assert response[1] == 200 

84 assert database.get_one_by_field("courses", "course_id", "CS101") is not None 

85 database.delete_all_by_field("courses", "course_id", "CS101") 

86 

87 

88def test_add_existing_course(database, course_model, sample_course, app): 

89 """Test adding an existing course.""" 

90 database.delete_all_by_field("courses", "course_id", "CS101") 

91 with app.app_context(): 

92 with app.test_request_context(): 

93 course_model.add_course(sample_course) 

94 response = course_model.add_course(sample_course) 

95 assert response[1] == 400 

96 assert response[0].json["error"] == "Course already in database" 

97 database.delete_all_by_field("courses", "course_id", "CS101") 

98 

99 

100def test_delete_course(database, course_model, sample_course, app): 

101 """Test deleting a course.""" 

102 database.delete_all_by_field("courses", "course_id", "CS101") 

103 database.delete_all_by_field( 

104 "students", "course", "CS101" 

105 ) # Ensure no students are linked 

106 with app.app_context(): 

107 with app.test_request_context(): 

108 course_model.add_course(sample_course) 

109 course_id = sample_course["_id"] 

110 

111 # Ensure the course exists before deletion 

112 assert database.get_one_by_field( 

113 "courses", "course_id", "CS101" 

114 ), "Course was not added" 

115 

116 response = course_model.delete_course_by_uuid(course_id) 

117 

118 # Ensure course is deleted 

119 assert response[1] == 200, f"Unexpected response: {response[0].json}" 

120 assert ( 

121 database.get_one_by_id("courses", course_id) is None 

122 ), "Course still exists after deletion" 

123 

124 

125def test_delete_nonexistent_course(course_model, app): 

126 """Test deleting a nonexistent course.""" 

127 with app.app_context(): 

128 with app.test_request_context(): 

129 response = course_model.delete_course_by_uuid("nonexistent_id") 

130 assert response[1] == 404 

131 assert response[0].json["error"] == "Course not found" 

132 

133 

134def test_get_course_by_id(database, course_model, sample_course, app): 

135 """Test getting a course by ID.""" 

136 database.delete_all_by_field("courses", "course_id", "CS101") 

137 with app.app_context(): 

138 with app.test_request_context(): 

139 course_model.add_course(sample_course) 

140 course = course_model.get_course_by_id("CS101") 

141 assert course is not None 

142 assert course["course_id"] == "CS101" 

143 database.delete_all_by_field("courses", "course_id", "CS101") 

144 

145 

146def test_get_course_by_uuid(database, course_model, sample_course, app): 

147 """Test getting a course by UUID.""" 

148 database.delete_all_by_field("courses", "course_id", "CS101") 

149 with app.app_context(): 

150 with app.test_request_context(): 

151 course_model.add_course(sample_course) 

152 course = course_model.get_course_by_uuid(sample_course["_id"]) 

153 assert course is not None 

154 assert course["_id"] == sample_course["_id"] 

155 database.delete_all_by_field("courses", "course_id", "CS101") 

156 

157 

158def test_get_course_name_by_id(database, course_model, sample_course, app): 

159 """Test getting a course name by ID.""" 

160 database.delete_all_by_field("courses", "course_id", "CS101") 

161 with app.app_context(): 

162 with app.test_request_context(): 

163 course_model.add_course(sample_course) 

164 course_name = course_model.get_course_name_by_id("CS101") 

165 assert course_name == "Introduction to Computer Science" 

166 database.delete_all_by_field("courses", "course_id", "CS101") 

167 

168 

169def test_get_courses(database, course_model, sample_course, app): 

170 """Test getting all courses.""" 

171 database.delete_all_by_field("courses", "course_id", "CS101") 

172 with app.app_context(): 

173 with app.test_request_context(): 

174 course_model.add_course(sample_course) 

175 courses = course_model.get_courses() 

176 assert len(courses) > 0 

177 database.delete_all_by_field("courses", "course_id", "CS101") 

178 

179 

180def test_get_courses_map(database, course_model, sample_course, app): 

181 """Test getting courses map.""" 

182 database.delete_all_by_field("courses", "course_id", "CS101") 

183 with app.app_context(): 

184 with app.test_request_context(): 

185 course_model.add_course(sample_course) 

186 courses_map = course_model.get_courses_map() 

187 assert "CS101" in courses_map 

188 database.delete_all_by_field("courses", "course_id", "CS101") 

189 

190 

191def test_update_course(database, course_model, sample_course, app): 

192 """Test updating a course.""" 

193 database.delete_all_by_field("courses", "course_id", "CS101") 

194 with app.app_context(): 

195 with app.test_request_context(): 

196 course_model.add_course(sample_course) 

197 updated_course = { 

198 "course_id": "CS101", 

199 "course_name": "Intro to CS", 

200 "course_description": "Updated description.", 

201 } 

202 response = course_model.update_course(sample_course["_id"], updated_course) 

203 assert response[1] == 200 

204 course = course_model.get_course_by_uuid(sample_course["_id"]) 

205 assert course["course_name"] == "Intro to CS" 

206 assert course["course_description"] == "Updated description." 

207 

208 

209def test_reset_cache(database, course_model, sample_course, app): 

210 """Test resetting the courses cache.""" 

211 database.delete_all_by_field("courses", "course_id", "CS101") 

212 with app.app_context(): 

213 with app.test_request_context(): 

214 course_model.add_course(sample_course) 

215 result = course_model.reset_cache() 

216 assert len(result) > 0 

217 database.delete_all_by_field("courses", "course_id", "CS101") 

218 

219 

220def test_get_course_by_invalid_id(database, course_model, app): 

221 """Test getting a course by invalid ID.""" 

222 database.delete_all_by_field("courses", "course_id", "CS101") 

223 with app.app_context(): 

224 with app.test_request_context(): 

225 course = course_model.get_course_by_id("INVALID_ID") 

226 assert course is None 

227 database.delete_all_by_field("courses", "course_id", "CS101") 

228 

229 

230def test_get_course_by_invalid_uuid(database, course_model, app): 

231 """Test getting a course by invalid UUID.""" 

232 database.delete_all_by_field("courses", "course_id", "CS101") 

233 with app.app_context(): 

234 with app.test_request_context(): 

235 course = course_model.get_course_by_uuid("INVALID_UUID") 

236 assert course is None 

237 database.delete_all_by_field("courses", "course_id", "CS101") 

238 

239 

240def test_get_course_name_by_invalid_id(database, course_model, app): 

241 """Test getting a course name by invalid ID.""" 

242 database.delete_all_by_field("courses", "course_id", "CS101") 

243 with app.app_context(): 

244 with app.test_request_context(): 

245 course_name = course_model.get_course_name_by_id("INVALID_ID") 

246 assert course_name is None 

247 database.delete_all_by_field("courses", "course_id", "CS101") 

248 

249 

250def test_update_course_not_updated(database, course_model, sample_course, app): 

251 """Test updating a course that does not exist.""" 

252 database.delete_all_by_field("courses", "course_id", "CS101") 

253 with app.app_context(): 

254 with app.test_request_context(): 

255 response = course_model.update_course("nonexistent_uuid", sample_course) 

256 assert response[1] == 404 

257 assert response[0].json["error"] == "Course not found" 

258 database.delete_all_by_field("courses", "course_id", "CS101") 

259 

260 

261def test_update_course_students(database, course_model, sample_course, app): 

262 """Test updating a course and ensuring students are updated.""" 

263 database.delete_all_by_field("courses", "course_id", "CS101") 

264 database.delete_all_by_field("courses", "course_id", "CS102") 

265 database.delete_all_by_field("students", "email", "student@dummy.com") 

266 with app.app_context(): 

267 with app.test_request_context(): 

268 course_model.add_course(sample_course) 

269 student = { 

270 "_id": uuid.uuid4().hex, 

271 "email": "student@dummy.com", 

272 "course": "CS101", 

273 } 

274 database.insert("students", student) 

275 updated_course = { 

276 "course_id": "CS102", 

277 "course_name": "Intro to CS", 

278 "course_description": "Updated description.", 

279 } 

280 response = course_model.update_course(sample_course["_id"], updated_course) 

281 assert response[1] == 200 

282 updated_student = database.get_one_by_field( 

283 "students", "email", "student@dummy.com" 

284 ) 

285 assert updated_student["course"] == "CS102" 

286 database.delete_all_by_field("courses", "course_id", "CS101") 

287 database.delete_all_by_field("courses", "course_id", "CS102") 

288 database.delete_all_by_field("students", "email", "student@dummy.com") 

289 

290 

291def test_delete_all_courses(database, course_model, sample_course, app): 

292 """Test deleting all courses.""" 

293 if database.get_one_by_field("courses", "course_id", sample_course["course_id"]): 

294 database.delete_all("courses") 

295 database.insert("courses", sample_course) 

296 assert ( 

297 database.get_one_by_field("courses", "course_id", sample_course["course_id"]) 

298 is not None 

299 ) 

300 database.delete_all("students") 

301 

302 with app.app_context(): 

303 response = course_model.delete_all_courses() 

304 assert response[1] == 200 

305 assert database.get_all("courses") == [] 

306 

307 

308def test_download_all_courses(database, course_model, sample_course, app): 

309 """Test downloading all courses.""" 

310 database.insert("courses", sample_course) 

311 with app.app_context(): 

312 with app.test_request_context(): 

313 # Ensure the directory exists 

314 if os.name == "nt": 

315 os.makedirs("temp", exist_ok=True) 

316 else: 

317 os.makedirs("/tmp", exist_ok=True) 

318 response = course_model.download_all_courses() 

319 assert response.status_code == 200 

320 assert ( 

321 response.mimetype 

322 == "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" 

323 ) 

324 

325 

326def test_upload_course_data(database, course_model, app): 

327 """Test uploading course data from an Excel file.""" 

328 database.delete_all_by_field("courses", "course_id", "CS102") 

329 df = pd.DataFrame( 

330 [ 

331 { 

332 "UCAS_code": "CS102", 

333 "Course_name": "Data Science", 

334 "Qualification": "BSc", 

335 "Course_description": "Intro to Data Science", 

336 } 

337 ] 

338 ) 

339 excel_buffer = BytesIO() 

340 df.to_excel(excel_buffer, index=False) 

341 excel_buffer.seek(0) 

342 excel_buffer.name = "courses.xlsx" 

343 with app.app_context(): 

344 response = course_model.upload_course_data(excel_buffer) 

345 assert response[1] == 200 

346 assert database.get_one_by_field("courses", "course_id", "CS102") is not None 

347 

348 database.delete_all_by_field("courses", "course_id", "CS102") 

349 

350 

351def test_update_course_duplicate_id(database, course_model, sample_course, app): 

352 """Test updating a course to an existing course ID.""" 

353 database.delete_all_by_field("courses", "course_id", "CS102") 

354 database.insert("courses", sample_course) 

355 second_course = { 

356 "_id": uuid.uuid4().hex, 

357 "course_id": "CS102", 

358 "course_name": "Advanced Programming", 

359 "course_description": "Advanced topics in programming", 

360 } 

361 database.insert("courses", second_course) 

362 updated_course = {"course_id": "CS102", "course_name": "New Name"} 

363 with app.app_context(): 

364 response = course_model.update_course(sample_course["_id"], updated_course) 

365 assert response[1] == 400 

366 assert response[0].json["error"] == "Course ID already exists" 

367 

368 database.delete_all_by_field("courses", "course_id", "CS102") 

369 

370 

371def test_delete_course_with_students(database, course_model, sample_course, app): 

372 """Test deleting a course with enrolled students.""" 

373 database.insert("courses", sample_course) 

374 student = { 

375 "_id": uuid.uuid4().hex, 

376 "email": "student@example.com", 

377 "course": "CS101", 

378 } 

379 database.insert("students", student) 

380 with app.app_context(): 

381 response = course_model.delete_course_by_uuid(sample_course["_id"]) 

382 assert response[1] == 400 

383 assert response[0].json["error"] == "Course has students enrolled"