Coverage for tests/model_tests/test_opportunities.py: 97%

1"""Test for the Opportunities model.""" 

2 

3import os 

4import sys 

5import uuid 

6from unittest.mock import MagicMock, patch 

7from io import BytesIO 

8from dotenv import load_dotenv 

9import pytest 

10from flask import session 

11import pandas as pd 

12 

13 

14# flake8: noqa: F811 

15 

16# Add the root directory to the Python path 

17sys.path.append( 

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

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 app.config["TESTING"] = True 

34 return app 

35 

36 

37@pytest.fixture() 

38def employer_model(): 

39 """Fixture to create an employer model.""" 

40 from ...employers.models import Employers # pylint: disable=import-outside-toplevel 

41 

42 return Employers() 

43 

44 

45@pytest.fixture() 

46def opportunity_model(): 

47 """Fixture to create a user model.""" 

48 from ...opportunities.models import ( 

49 Opportunity, 

50 ) # pylint: disable=import-outside-toplevel 

51 

52 return Opportunity() 

53 

54 

55@pytest.fixture() 

56def database(): 

57 """Fixture to create a database.""" 

58 

59 database = DatabaseMongoManager( 

60 shared.getenv("MONGO_URI"), 

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

62 ) 

63 current_opportunities = database.get_all("opportunities") 

64 database.delete_all("opportunities") 

65 yield database 

66 database.delete_all("opportunities") 

67 for opportunity in current_opportunities: 

68 database.insert("opportunities", opportunity) 

69 

70 database.connection.close() 

71 

72 

73@pytest.fixture() 

74def dummy_data(database): 

75 """Gives dummy data""" 

76 data = { 

77 "Title": ["Opportunity 1", "Opportunity 2"], 

78 "Description": ["Description 1", "Description 2"], 

79 "URL": ["http://example.com/1", "http://example.com/2"], 

80 "Spots_available": [5, 10], 

81 "Location": ["Location 1", "Location 2"], 

82 "Duration": ["1_week", "1_month"], 

83 "Employer_email": ["employer1@example.com", "employer2@example.com"], 

84 "Modules_required": ["THING1,THING2", "THING3"], 

85 "Courses_required": ["THING1", "THING2"], 

86 } 

87 module1 = { 

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

89 "module_id": "THING1", 

90 "module_name": "THING1", 

91 "module_description": "THING1", 

92 } 

93 module2 = { 

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

95 "module_id": "THING2", 

96 "module_name": "THING2", 

97 "module_description": "THING2", 

98 } 

99 module3 = { 

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

101 "module_id": "THING3", 

102 "module_name": "THING3", 

103 "module_description": "THING3", 

104 } 

105 course1 = { 

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

107 "course_id": "THING1", 

108 "course_name": "THING1", 

109 "course_description": "THING1", 

110 } 

111 course2 = { 

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

113 "course_id": "THING2", 

114 "course_name": "THING2", 

115 "course_description": "THING2", 

116 } 

117 employer1 = { 

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

119 "company_name": "Company1", 

120 "email": "employer1@example.com", 

121 } 

122 employer2 = { 

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

124 "company_name": "Company2", 

125 "email": "employer2@example.com", 

126 } 

127 

128 database.insert("modules", module1) 

129 database.insert("modules", module2) 

130 database.insert("modules", module3) 

131 database.insert("courses", course1) 

132 database.insert("courses", course2) 

133 database.insert("employers", employer1) 

134 database.insert("employers", employer2) 

135 

136 yield data 

137 

138 database.delete_by_id("modules", module1["_id"]) 

139 database.delete_by_id("modules", module2["_id"]) 

140 database.delete_by_id("modules", module3["_id"]) 

141 database.delete_by_id("courses", course1["_id"]) 

142 database.delete_by_id("courses", course2["_id"]) 

143 database.delete_by_id("employers", employer1["_id"]) 

144 database.delete_by_id("employers", employer2["_id"]) 

145 

146 

147def test_start_session(app, employer_model): 

148 """Test starting a session.""" 

149 with app.app_context(): 

150 with app.test_request_context(): 

151 response = employer_model.start_session() 

152 assert response.status_code == 200 

153 assert session["employer_logged_in"] is True 

154 

155 

156def test_add_update_opportunity_unauthorized(opportunity_model, database, app): 

157 """Test unauthorized access when updating an opportunity.""" 

158 database.delete_all_by_field("opportunities", "_id", "123") 

159 database.insert("opportunities", {"_id": "123", "employer_id": "456"}) 

160 with app.app_context(): # Set up Flask application context 

161 with app.test_request_context(): # Set up request context for session 

162 # Mock the session 

163 session["employer"] = {"_id": "123"} 

164 

165 # Call the method 

166 response = opportunity_model.add_update_opportunity( 

167 {"_id": "123", "employer_id": "456"} 

168 ) 

169 

170 # Verify the response 

171 assert response[1] == 401 

172 assert response[0].json == {"error": "Unauthorized Access."} 

173 

174 

175def test_rank_preferences_success(opportunity_model, database, app): 

176 """Test updating preferences for an existing opportunity.""" 

177 with app.app_context(): # Set up Flask application context 

178 # Mock the database manager 

179 with patch("app.DATABASE_MANAGER", database): 

180 # Mock the database responses 

181 database.get_one_by_id = MagicMock( 

182 return_value={"_id": "1", "title": "Job 1"} 

183 ) 

184 database.update_one_by_field = MagicMock(return_value=True) 

185 

186 # Call the method 

187 response = opportunity_model.rank_preferences("1", ["pref1", "pref2"]) 

188 

189 # Verify the database calls 

190 database.get_one_by_id.assert_called_once_with("opportunities", "1") 

191 database.update_one_by_field.assert_called_once_with( 

192 "opportunities", "_id", "1", {"preferences": ["pref1", "pref2"]} 

193 ) 

194 

195 # Verify the response 

196 assert response[1] == 200 

197 assert response[0].json == {"message": "Preferences updated"} 

198 

199 

200def test_rank_preferences_not_found(opportunity_model, database, app): 

201 """Test updating preferences for a non-existent opportunity.""" 

202 with app.app_context(): # Set up Flask application context 

203 # Mock the database manager 

204 with patch("app.DATABASE_MANAGER", database): 

205 # Mock the database responses 

206 database.get_one_by_id = MagicMock(return_value=None) 

207 database.update_one_by_field = MagicMock() # Properly mock this method 

208 

209 # Call the method 

210 response = opportunity_model.rank_preferences("1", ["pref1", "pref2"]) 

211 

212 # Verify the database calls 

213 database.get_one_by_id.assert_called_once_with("opportunities", "1") 

214 database.update_one_by_field.assert_not_called() # Now this will work 

215 

216 # Verify the response 

217 assert response[1] == 404 

218 assert response[0].json == {"error": "Opportunity not found"} 

219 

220 

221def test_search_opportunities_exception(opportunity_model, app): 

222 """Test searching opportunities with an exception.""" 

223 

224 with app.app_context(): # Set up Flask application context 

225 with app.test_request_context(): # Set up request context for session 

226 opportunities = opportunity_model.get_opportunities_for_search("dfasdf") 

227 assert opportunities == [] 

228 

229 

230def test_search_opportunities_by_company(opportunity_model, database, app): 

231 """Test searching opportunities by company name only.""" 

232 

233 # Clear any existing data 

234 database.delete_all_by_field("opportunities", "_id", "123") 

235 database.delete_all_by_field("employers", "_id", "456") 

236 

237 # Insert test data 

238 database.insert( 

239 "opportunities", {"_id": "123", "title": "SE", "employer_id": "456"} 

240 ) 

241 database.insert("employers", {"_id": "456", "company_name": "Company1"}) 

242 

243 with app.app_context(): # Set up Flask application context 

244 # Call the function with only the company name 

245 opportunities = opportunity_model.get_opportunities_for_search("456") 

246 

247 # Assert the results 

248 assert len(opportunities) == 1 

249 assert opportunities[0]["_id"] == "123" 

250 assert opportunities[0]["title"] == "SE" 

251 assert opportunities[0]["company_name"] == "Company1" 

252 

253 # Clean up 

254 database.delete_all_by_field("opportunities", "_id", "123") 

255 database.delete_all_by_field("employers", "_id", "456") 

256 

257 

258def test_get_opportunities_by_title_no_title(opportunity_model, app): 

259 """Test getting opportunities by title.""" 

260 

261 with app.app_context(): 

262 opportunities = opportunity_model.get_opportunities_by_title("") 

263 assert opportunities == [] 

264 

265 

266def test_get_opportunities_by_title_exception(opportunity_model, app, database): 

267 """Test getting opportunities by title.""" 

268 assert database.get_all("opportunities") == [] 

269 with app.app_context(): 

270 opportunities = opportunity_model.get_opportunities_by_title("SE") 

271 assert opportunities == [] 

272 

273 

274def test_get_opportunities_by_title(opportunity_model, database, app): 

275 """Test getting opportunities by title.""" 

276 

277 # Clear any existing data 

278 database.delete_all_by_field("opportunities", "_id", "123") 

279 database.delete_all_by_field("employers", "_id", "456") 

280 

281 # Insert test data 

282 database.insert( 

283 "opportunities", {"_id": "123", "title": "SE", "employer_id": "456"} 

284 ) 

285 database.insert("employers", {"_id": "456", "company_name": "Company1"}) 

286 

287 with app.app_context(): # Set up Flask application context 

288 # Call the function with only the title 

289 opportunities = opportunity_model.get_opportunities_by_title("SE") 

290 

291 # Assert the results 

292 assert len(opportunities) == 1 

293 assert opportunities[0]["_id"] == "123" 

294 assert opportunities[0]["title"] == "SE" 

295 assert opportunities[0]["employer_id"] == "456" 

296 

297 # Clean up 

298 database.delete_all_by_field("opportunities", "_id", "123") 

299 database.delete_all_by_field("employers", "_id", "456") 

300 

301 

302def test_get_opportunities_by_company_no_company(opportunity_model, app): 

303 """Test getting opportunities by company.""" 

304 

305 with app.app_context(): 

306 opportunities = opportunity_model.get_opportunities_by_company("") 

307 assert opportunities == [] 

308 

309 

310def test_get_opportunities_by_company_does_not_exist(opportunity_model, app): 

311 """Test getting opportunities by company.""" 

312 

313 with app.app_context(): 

314 opportunities = opportunity_model.get_opportunities_by_company("Company1") 

315 assert opportunities == [] 

316 

317 

318def test_get_opportunities_by_company(opportunity_model, database, app): 

319 """Test getting opportunities by company""" 

320 database.delete_all_by_field("opportunities", "_id", "123") 

321 database.delete_all_by_field("employers", "_id", "456") 

322 

323 database.insert( 

324 "opportunities", {"_id": "123", "title": "SE", "employer_id": "456"} 

325 ) 

326 database.insert("employers", {"_id": "456", "company_name": "Company1"}) 

327 

328 with app.app_context(): 

329 opportunities = opportunity_model.get_opportunities_by_company("Company1") 

330 

331 assert len(opportunities) == 1 

332 assert opportunities[0]["_id"] == "123" 

333 assert opportunities[0]["title"] == "SE" 

334 assert opportunities[0]["employer_id"] == "456" 

335 

336 database.delete_all_by_field("opportunities", "_id", "123") 

337 database.delete_all_by_field("employers", "_id", "456") 

338 

339 

340def test_get_opportunities_by_company_id(opportunity_model, database, app): 

341 """Test getting opportunities by company id""" 

342 database.delete_all_by_field("opportunities", "_id", "123") 

343 

344 database.insert( 

345 "opportunities", {"_id": "123", "title": "SE", "employer_id": "456"} 

346 ) 

347 

348 with app.app_context(): 

349 with app.test_request_context(): 

350 opportunities = opportunity_model.get_opportunity_by_company_id("456") 

351 assert len(opportunities) == 1 

352 assert opportunities[0]["_id"] == "123" 

353 

354 database.delete_all_by_field("opportunities", "_id", "123") 

355 

356 

357def test_get_opportunity_by_id(opportunity_model, database, app): 

358 """Test getting an opportunity by id""" 

359 opportunities = database.get_all("opportunities") 

360 database.delete_all("opportunities") 

361 

362 op = { 

363 "_id": "2", 

364 "name": "Opportunity 2", 

365 } 

366 database.insert("opportunities", op) 

367 with app.app_context(): 

368 with app.test_request_context(): 

369 # Test for an existing opportunity 

370 opportunity_model.get_opportunity_by_id("2") 

371 opportunity = opportunity_model.get_opportunity_by_id("2") 

372 assert opportunity is not None 

373 assert opportunity["_id"] == "2" 

374 assert opportunity["name"] == "Opportunity 2" 

375 

376 # Test for a non-existing opportunity 

377 opportunity = opportunity_model.get_opportunity_by_id("999") 

378 assert opportunity is None 

379 

380 database.delete_all("opportunities") 

381 for opportunity in opportunities: 

382 database.insert("opportunities", opportunity) 

383 

384 

385def test_get_employer_by_id(opportunity_model, database, app): 

386 """Test getting an employer by id""" 

387 database.delete_all_by_field("opportunities", "employer_id", "456") 

388 opportunity = {"_id": uuid.uuid4().hex, "employer_id": "456"} 

389 database.insert("opportunities", opportunity) 

390 database.insert("employers", {"_id": "456", "company_name": "Company1"}) 

391 

392 with app.app_context(): 

393 with app.test_request_context(): 

394 employer = opportunity_model.get_employer_by_id(opportunity["_id"]) 

395 assert employer == "456" 

396 

397 database.delete_all("opportunities") 

398 database.delete_all_by_field("employers", "_id", "456") 

399 

400 

401def test_get_employer_by_id_no_employer(opportunity_model, database, app): 

402 """Test getting an employer by id""" 

403 database.delete_all_by_field("opportunities", "_id", "123") 

404 

405 with app.app_context(): 

406 with app.test_request_context(): 

407 employer = opportunity_model.get_employer_by_id("123") 

408 assert employer == "" 

409 

410 

411def test_get_opportunities(opportunity_model, database, app): 

412 """Test getting all opportunities""" 

413 opportunity = database.get_all("opportunities") 

414 database.delete_all("opportunities") 

415 

416 database.insert("opportunities", {"_id": "123", "employer_id": "456"}) 

417 

418 with app.app_context(): 

419 with app.test_request_context(): 

420 opportunities = opportunity_model.get_opportunities() 

421 opportunities = opportunity_model.get_opportunities() 

422 

423 assert len(opportunities) == 1 

424 

425 database.delete_all_by_field("opportunities", "_id", "123") 

426 

427 for op in opportunity: 

428 database.insert("opportunities", op) 

429 

430 

431def test_get_opportunities_by_duration(opportunity_model, database, app): 

432 """Test getting opportunities by duration""" 

433 opportunity = database.get_all("opportunities") 

434 database.delete_all("opportunities") 

435 

436 database.insert( 

437 "opportunities", {"_id": "123", "employer_id": "456", "duration": "1_week"} 

438 ) 

439 

440 with app.app_context(): 

441 with app.test_request_context(): 

442 opportunities = opportunity_model.get_opportunities_by_duration( 

443 '["1_week", "1_day"]' 

444 ) 

445 

446 assert opportunities[1] == 200 

447 

448 for op in opportunity: 

449 database.insert("opportunities", op) 

450 

451 

452def test_delete_opportunity_by_id(opportunity_model, database, app): 

453 """Test deleting an opportunity by id""" 

454 opportunity = database.get_all("opportunities") 

455 students = database.get_all("students") 

456 

457 database.delete_all("opportunities") 

458 database.delete_all("students") 

459 

460 database.insert("opportunities", {"_id": "123", "employer_id": "456"}) 

461 database.insert("students", {"_id": "123", "preferences": ["123"]}) 

462 

463 with app.app_context(): 

464 with app.test_request_context(): 

465 opportunities = opportunity_model.delete_opportunity_by_id("123") 

466 

467 assert opportunities[1] == 200 

468 

469 database.delete_all_by_field("opportunities", "_id", "123") 

470 database.delete_all_by_field("students", "_id", "123") 

471 

472 for op in opportunity: 

473 database.insert("opportunities", op) 

474 

475 for student in students: 

476 database.insert("students", student) 

477 

478 

479def test_delete_opportunity_by_id_no_opportunity(opportunity_model, database, app): 

480 """Test deleting an opportunity by id""" 

481 opportunity = database.get_all("opportunities") 

482 

483 database.delete_all("opportunities") 

484 

485 with app.app_context(): 

486 with app.test_request_context(): 

487 opportunities = opportunity_model.delete_opportunity_by_id("123") 

488 assert opportunities[1] == 404 

489 assert opportunities[0].json == {"error": "Opportunity not found"} 

490 

491 for op in opportunity: 

492 database.insert("opportunities", op) 

493 

494 

495def test_get_valid_students(opportunity_model, database, app): 

496 """Test getting valid students""" 

497 opportunity = database.get_all("opportunities") 

498 students = database.get_all("students") 

499 

500 database.delete_all("opportunities") 

501 database.delete_all("students") 

502 

503 database.insert("opportunities", {"_id": "1234", "employer_id": "456"}) 

504 database.insert("students", {"_id": "123", "preferences": ["1234"]}) 

505 

506 with app.app_context(): 

507 with app.test_request_context(): 

508 result = opportunity_model.get_valid_students("1234") 

509 

510 assert result[0]["_id"] == "123" 

511 assert result[0]["preferences"] == ["1234"] 

512 

513 database.delete_all_by_field("opportunities", "_id", "1234") 

514 database.delete_all_by_field("students", "_id", "123") 

515 

516 for op in opportunity: 

517 database.insert("opportunities", op) 

518 

519 for student in students: 

520 database.insert("students", student) 

521 

522 

523def test_rank_preferences(opportunity_model, database, app): 

524 """Test ranking preferences""" 

525 opportunity = database.get_all("opportunities") 

526 

527 database.delete_all("opportunities") 

528 

529 database.insert( 

530 "opportunities", {"_id": "1234", "employer_id": "456", "preferences": ["123"]} 

531 ) 

532 

533 with app.app_context(): 

534 with app.test_request_context(): 

535 opportunities = opportunity_model.rank_preferences("1234", ["321"]) 

536 assert opportunities[1] == 200 

537 assert opportunities[0].json == {"message": "Preferences updated"} 

538 

539 updated = database.get_one_by_id("opportunities", "1234") 

540 assert updated["preferences"] == ["321"] 

541 

542 database.delete_all_by_field("opportunities", "_id", "1234") 

543 

544 for op in opportunity: 

545 database.insert("opportunities", op) 

546 

547 

548def test_rank_preferences_no_opportunity(opportunity_model, database, app): 

549 """Test ranking preferences""" 

550 opportunity = database.get_all("opportunities") 

551 

552 database.delete_all("opportunities") 

553 

554 with app.app_context(): 

555 with app.test_request_context(): 

556 opportunities = opportunity_model.rank_preferences("1234", ["321"]) 

557 assert opportunities[1] == 404 

558 assert opportunities[0].json == {"error": "Opportunity not found"} 

559 

560 for op in opportunity: 

561 database.insert("opportunities", op) 

562 

563 

564def test_delete_all_opportunity_admin(opportunity_model, database, app): 

565 """Test deleting all opportunities""" 

566 opportunity = database.get_all("opportunities") 

567 students = database.get_all("students") 

568 

569 database.delete_all("opportunities") 

570 database.delete_all("students") 

571 

572 database.insert("opportunities", {"_id": "1234", "employer_id": "456"}) 

573 database.insert("students", {"_id": "123", "preferences": ["1234"]}) 

574 

575 with app.app_context(): 

576 with app.test_request_context(): 

577 opportunities = opportunity_model.delete_all_opportunities(True) 

578 assert opportunities[1] == 200 

579 assert opportunities[0].json == {"message": "All opportunities deleted"} 

580 

581 database.delete_all_by_field("students", "_id", "123") 

582 database.delete_all_by_field("opportunities", "_id", "1234") 

583 

584 for op in opportunity: 

585 database.insert("opportunities", op) 

586 

587 for student in students: 

588 database.insert("students", student) 

589 

590 

591def test_delete_all_opportunity_employer( 

592 opportunity_model, employer_model, database, app 

593): 

594 """Test deleting all opportunities""" 

595 opportunity = database.get_all("opportunities") 

596 students = database.get_all("students") 

597 

598 database.delete_all("opportunities") 

599 database.delete_all("students") 

600 

601 database.insert("opportunities", {"_id": "1234", "employer_id": "456"}) 

602 database.insert("students", {"_id": "123", "preferences": ["1234"]}) 

603 

604 with app.app_context(): 

605 with app.test_request_context(): 

606 employer_model.start_session() 

607 session["employer"] = {"_id": "456"} 

608 opportunities = opportunity_model.delete_all_opportunities(False) 

609 assert opportunities[1] == 200 

610 assert opportunities[0].json == {"message": "All opportunities deleted"} 

611 

612 result = database.get_all("opportunities") 

613 assert len(result) == 0 

614 

615 database.delete_all_by_field("students", "_id", "123") 

616 database.delete_all_by_field("opportunities", "_id", "1234") 

617 

618 for op in opportunity: 

619 database.insert("opportunities", op) 

620 

621 for student in students: 

622 database.insert("students", student) 

623 

624 

625def test_download_opportunities_admin(opportunity_model, database, app): 

626 """Test downloading opportunities""" 

627 opportunity = database.get_all("opportunities") 

628 employers = database.get_all("employers") 

629 

630 database.delete_all("opportunities") 

631 database.delete_all("employers") 

632 

633 database.insert( 

634 "employers", 

635 {"_id": "456", "company_name": "Company1", "email": "dummy@dummy.com"}, 

636 ) 

637 database.insert( 

638 "opportunities", 

639 { 

640 "_id": "1234", 

641 "employer_id": "456", 

642 "title": "Job 1", 

643 "description": "Description 1", 

644 "duration": "1_week", 

645 "location": "Location 1", 

646 "modules_required": ["Module 1", "Module 2"], 

647 "url": "URL 1", 

648 "courses_required": ["Course 1", "Course 2"], 

649 "spots_available": 10, 

650 }, 

651 ) 

652 

653 with app.app_context(): 

654 with app.test_request_context(): 

655 opportunities = opportunity_model.download_opportunities(True) 

656 assert opportunities.status_code == 200 

657 

658 database.delete_all_by_field("opportunities", "_id", "1234") 

659 database.delete_all_by_field("employers", "_id", "456") 

660 

661 for op in opportunity: 

662 database.insert("opportunities", op) 

663 

664 for employer in employers: 

665 database.insert("employers", employer) 

666 

667 

668def test_download_opportunities_employer( 

669 opportunity_model, employer_model, database, app 

670): 

671 """Test downloading opportunities""" 

672 opportunity = database.get_all("opportunities") 

673 employers = database.get_all("employers") 

674 

675 database.delete_all("opportunities") 

676 database.delete_all("employers") 

677 

678 database.insert( 

679 "employers", 

680 {"_id": "456", "company_name": "Company1", "email": "dummy@dummy.com"}, 

681 ) 

682 database.insert( 

683 "opportunities", 

684 { 

685 "_id": "1234", 

686 "employer_id": "456", 

687 "title": "Job 1", 

688 "description": "Description 1", 

689 "duration": "1_week", 

690 "location": "Location 1", 

691 "modules_required": ["Module 1", "Module 2"], 

692 "url": "URL 1", 

693 "courses_required": ["Course 1", "Course 2"], 

694 "spots_available": 10, 

695 }, 

696 ) 

697 

698 with app.app_context(): 

699 with app.test_request_context(): 

700 employer_model.start_session() 

701 session["employer"] = {"_id": "456"} 

702 opportunities = opportunity_model.download_opportunities(False) 

703 assert opportunities.status_code == 200 

704 

705 database.delete_all_by_field("opportunities", "_id", "1234") 

706 database.delete_all_by_field("employers", "_id", "456") 

707 

708 for op in opportunity: 

709 database.insert("opportunities", op) 

710 

711 for employer in employers: 

712 database.insert("employers", employer) 

713 

714 

715def test_upload_opportunities(opportunity_model, database, app, dummy_data): 

716 """Test upload opportunities""" 

717 assert len(database.get_all("opportunities")) == 0 

718 df = pd.DataFrame(dummy_data) 

719 file = BytesIO() 

720 df.to_excel(file, index=False) 

721 file.seek(0) 

722 file.name = "opportunities.xlsx" 

723 

724 with app.app_context(): 

725 with app.test_request_context(): 

726 # Test successful upload (admin) 

727 response = opportunity_model.upload_opportunities(file, is_admin=True) 

728 assert response[0].json == { 

729 "message": "Opportunities uploaded successfully" 

730 } 

731 assert response[1] == 200 

732 

733 result = database.get_one_by_field( 

734 "opportunities", "title", "Opportunity 1" 

735 ) 

736 assert result["title"] == "Opportunity 1" 

737 result = database.get_one_by_field( 

738 "opportunities", "title", "Opportunity 2" 

739 ) 

740 assert result["title"] == "Opportunity 2" 

741 

742 

743def test_upload_opportunities_employer( 

744 opportunity_model, employer_model, database, app, dummy_data 

745): 

746 """Test upload opportunities by employer""" 

747 assert len(database.get_all("opportunities")) == 0 

748 

749 df = pd.DataFrame(dummy_data) 

750 file = BytesIO() 

751 df.to_excel(file, index=False) 

752 file.seek(0) 

753 file.name = "opportunities.xlsx" 

754 with app.app_context(): 

755 with app.test_request_context(): 

756 employer_model.start_session() 

757 session["employer"] = {"_id": "456"} 

758 response = opportunity_model.upload_opportunities(file, is_admin=False) 

759 assert response[0].json == { 

760 "message": "Opportunities uploaded successfully" 

761 } 

762 assert response[1] == 200 

763 

764 result = database.get_all_by_field("opportunities", "employer_id", "456") 

765 assert len(result) == 2 

766 assert result[0]["title"] == "Opportunity 1" 

767 assert result[1]["title"] == "Opportunity 2" 

768 

769 

770def test_upload_opportunities_wrong_spot_available( 

771 opportunity_model, database, app, dummy_data 

772): 

773 """Test upload opportunities with invalid spots available value""" 

774 assert len(database.get_all("opportunities")) == 0 

775 

776 dummy_data["Spots_available"] = ["invalid", 10] 

777 df = pd.DataFrame(dummy_data) 

778 file = BytesIO() 

779 df.to_excel(file, index=False) 

780 file.seek(0) 

781 file.name = "opportunities.xlsx" 

782 with app.app_context(): 

783 with app.test_request_context(): 

784 response = opportunity_model.upload_opportunities(file, is_admin=True) 

785 assert response[1] == 400 

786 assert ( 

787 "Failed to upload opportunities: invalid literal for int() with base 10: 'invalid'" 

788 in response[0].json["error"] 

789 ) 

790 

791 

792def test_upload_opportunities_wrong_duration( 

793 opportunity_model, database, app, dummy_data 

794): 

795 """Test upload opportunities with invalid duration value""" 

796 

797 assert len(database.get_all("opportunities")) == 0 

798 dummy_data["Spots_available"] = [5, 10] 

799 dummy_data["Duration"] = ["invalid", "1_month"] 

800 df = pd.DataFrame(dummy_data)