Coverage for app/tests/api/test_tasks.py: 99%

1from datetime import datetime, timedelta 

2from typing import Any 

3 

4from fastapi.testclient import TestClient 

5from sqlmodel import Session 

6 

7from app import crud 

8from app.core.config import settings 

9from app.models import Priority, UserCreate 

10from app.tests.utils.user import ( 

11 create_random_user, 

12 user_authentication_headers, 

13) 

14from app.tests.utils.utils import random_lower_string 

15 

16 

17def _get_priority(db: Session, pid: int = 3) -> Priority: 

18 """Get a seeded priority row (1=urgent, 2=high, 3=medium, 4=low).""" 

19 priority = db.get(Priority, pid) 

20 assert priority is not None, f"Priority {pid} not seeded — run init_db first" 

21 return priority 

22 

23 

24def _fresh_user_headers(client: TestClient, db: Session) -> dict[str, str]: 

25 """Create a brand-new user and return auth headers.""" 

26 password = random_lower_string() 

27 user_in = UserCreate(email=f"{random_lower_string()}@test.com", password=password) 

28 user = crud.create_user(session=db, user_create=user_in) 

29 return user_authentication_headers( 

30 client=client, email=user.email, password=password 

31 ) 

32 

33 

34def _create_task( 

35 client: TestClient, headers: dict[str, str], **fields: object 

36) -> dict[str, Any]: 

37 """Create a task via API and assert success.""" 

38 data: dict[str, object] = {"title": random_lower_string(), **fields} 

39 resp = client.post(f"{settings.API_V1_STR}/tasks/", headers=headers, json=data) 

40 assert resp.status_code == 200 

41 result: dict[str, Any] = resp.json() 

42 return result 

43 

44 

45def test_create_task( 

46 client: TestClient, superuser_token_headers: dict[str, str] 

47) -> None: 

48 title = random_lower_string() 

49 description = random_lower_string() 

50 data = {"title": title, "description": description} 

51 response = client.post( 

52 f"{settings.API_V1_STR}/tasks/", headers=superuser_token_headers, json=data 

53 ) 

54 assert response.status_code == 200 

55 content = response.json() 

56 assert content["title"] == title 

57 assert content["description"] == description 

58 assert "id" in content 

59 assert "owner_id" in content 

60 

61 

62def test_create_task_with_owner( 

63 client: TestClient, superuser_token_headers: dict[str, str], db: Session 

64) -> None: 

65 title = random_lower_string() 

66 description = random_lower_string() 

67 

68 # create a random user 

69 user = create_random_user(db) 

70 

71 # try to create a task with owner_id - this should be ignored 

72 data = {"title": title, "description": description, "owner_id": str(user.id)} 

73 response = client.post( 

74 f"{settings.API_V1_STR}/tasks/", headers=superuser_token_headers, json=data 

75 ) 

76 assert response.status_code == 200 

77 content = response.json() 

78 assert content["title"] == title 

79 assert content["description"] == description 

80 assert "id" in content 

81 assert "owner_id" in content 

82 # the owner_id should be the superuser's ID, not the random user's ID 

83 assert content["owner_id"] != str(user.id) 

84 

85 

86def test_read_task(client: TestClient, superuser_token_headers: dict[str, str]) -> None: 

87 # create a task 

88 title = random_lower_string() 

89 description = random_lower_string() 

90 data = {"title": title, "description": description} 

91 response = client.post( 

92 f"{settings.API_V1_STR}/tasks/", headers=superuser_token_headers, json=data 

93 ) 

94 assert response.status_code == 200 

95 content = response.json() 

96 task_id = content["id"] 

97 

98 # read the task 

99 response = client.get( 

100 f"{settings.API_V1_STR}/tasks/{task_id}", headers=superuser_token_headers 

101 ) 

102 assert response.status_code == 200 

103 content = response.json() 

104 assert content["title"] == title 

105 assert content["description"] == description 

106 assert content["id"] == task_id 

107 

108 

109def test_update_task( 

110 client: TestClient, superuser_token_headers: dict[str, str] 

111) -> None: 

112 # create a task 

113 title = random_lower_string() 

114 description = random_lower_string() 

115 data = {"title": title, "description": description} 

116 response = client.post( 

117 f"{settings.API_V1_STR}/tasks/", headers=superuser_token_headers, json=data 

118 ) 

119 assert response.status_code == 200 

120 content = response.json() 

121 task_id = content["id"] 

122 

123 # update the task 

124 new_title = random_lower_string() 

125 new_description = random_lower_string() 

126 data = {"title": new_title, "description": new_description} 

127 response = client.put( 

128 f"{settings.API_V1_STR}/tasks/{task_id}", 

129 headers=superuser_token_headers, 

130 json=data, 

131 ) 

132 assert response.status_code == 200 

133 content = response.json() 

134 assert content["title"] == new_title 

135 assert content["description"] == new_description 

136 assert content["id"] == task_id 

137 

138 

139def test_delete_task( 

140 client: TestClient, superuser_token_headers: dict[str, str] 

141) -> None: 

142 # create a task 

143 title = random_lower_string() 

144 description = random_lower_string() 

145 data = {"title": title, "description": description} 

146 response = client.post( 

147 f"{settings.API_V1_STR}/tasks/", headers=superuser_token_headers, json=data 

148 ) 

149 assert response.status_code == 200 

150 content = response.json() 

151 task_id = content["id"] 

152 

153 # delete the task 

154 response = client.delete( 

155 f"{settings.API_V1_STR}/tasks/{task_id}", headers=superuser_token_headers 

156 ) 

157 assert response.status_code == 200 

158 

159 # try to read the deleted task 

160 response = client.get( 

161 f"{settings.API_V1_STR}/tasks/{task_id}", headers=superuser_token_headers 

162 ) 

163 assert response.status_code == 404 

164 

165 

166def test_create_task_with_tags( 

167 client: TestClient, superuser_token_headers: dict[str, str] 

168) -> None: 

169 title = random_lower_string() 

170 data = {"title": title, "tags": ["work", "urgent"]} 

171 response = client.post( 

172 f"{settings.API_V1_STR}/tasks/", headers=superuser_token_headers, json=data 

173 ) 

174 assert response.status_code == 200 

175 content = response.json() 

176 assert content["title"] == title 

177 assert content["tags"] == ["work", "urgent"] 

178 

179 

180def test_create_task_with_priority( 

181 client: TestClient, superuser_token_headers: dict[str, str], db: Session 

182) -> None: 

183 priority = _get_priority(db) 

184 title = random_lower_string() 

185 data = {"title": title, "priority_id": priority.id} 

186 response = client.post( 

187 f"{settings.API_V1_STR}/tasks/", headers=superuser_token_headers, json=data 

188 ) 

189 assert response.status_code == 200 

190 content = response.json() 

191 assert content["title"] == title 

192 assert content["priority_id"] == priority.id 

193 

194 

195def test_create_task_with_due_date( 

196 client: TestClient, superuser_token_headers: dict[str, str] 

197) -> None: 

198 title = random_lower_string() 

199 due = "2026-04-01T09:00:00" 

200 data = {"title": title, "due": due} 

201 response = client.post( 

202 f"{settings.API_V1_STR}/tasks/", headers=superuser_token_headers, json=data 

203 ) 

204 assert response.status_code == 200 

205 content = response.json() 

206 assert content["title"] == title 

207 assert content["due"] is not None 

208 

209 

210def test_create_task_with_scheduled_start( 

211 client: TestClient, superuser_token_headers: dict[str, str] 

212) -> None: 

213 title = random_lower_string() 

214 scheduled_start = "2026-03-15T10:00:00" 

215 data = {"title": title, "scheduled_start": scheduled_start} 

216 response = client.post( 

217 f"{settings.API_V1_STR}/tasks/", headers=superuser_token_headers, json=data 

218 ) 

219 assert response.status_code == 200 

220 content = response.json() 

221 assert content["title"] == title 

222 assert content["scheduled_start"] is not None 

223 # The returned value should contain the datetime we sent 

224 assert scheduled_start in content["scheduled_start"] 

225 

226 

227def test_create_task_without_scheduled_start( 

228 client: TestClient, superuser_token_headers: dict[str, str] 

229) -> None: 

230 title = random_lower_string() 

231 data = {"title": title} 

232 response = client.post( 

233 f"{settings.API_V1_STR}/tasks/", headers=superuser_token_headers, json=data 

234 ) 

235 assert response.status_code == 200 

236 content = response.json() 

237 assert content["title"] == title 

238 assert content["scheduled_start"] is None 

239 

240 

241def test_update_task_scheduled_start( 

242 client: TestClient, superuser_token_headers: dict[str, str] 

243) -> None: 

244 # create a task without scheduled_start 

245 title = random_lower_string() 

246 data = {"title": title} 

247 response = client.post( 

248 f"{settings.API_V1_STR}/tasks/", headers=superuser_token_headers, json=data 

249 ) 

250 assert response.status_code == 200 

251 task_id = response.json()["id"] 

252 

253 # update the task to set scheduled_start 

254 scheduled_start = "2026-03-16T14:30:00" 

255 response = client.put( 

256 f"{settings.API_V1_STR}/tasks/{task_id}", 

257 headers=superuser_token_headers, 

258 json={"scheduled_start": scheduled_start}, 

259 ) 

260 assert response.status_code == 200 

261 content = response.json() 

262 assert content["scheduled_start"] is not None 

263 assert scheduled_start in content["scheduled_start"] 

264 

265 # verify via GET 

266 response = client.get( 

267 f"{settings.API_V1_STR}/tasks/{task_id}", headers=superuser_token_headers 

268 ) 

269 assert response.status_code == 200 

270 content = response.json() 

271 assert content["scheduled_start"] is not None 

272 assert scheduled_start in content["scheduled_start"] 

273 

274 

275def test_clear_task_scheduled_start( 

276 client: TestClient, superuser_token_headers: dict[str, str] 

277) -> None: 

278 # create a task with scheduled_start 

279 title = random_lower_string() 

280 scheduled_start = "2026-03-17T09:00:00" 

281 data = {"title": title, "scheduled_start": scheduled_start} 

282 response = client.post( 

283 f"{settings.API_V1_STR}/tasks/", headers=superuser_token_headers, json=data 

284 ) 

285 assert response.status_code == 200 

286 content = response.json() 

287 task_id = content["id"] 

288 assert content["scheduled_start"] is not None 

289 

290 # clear scheduled_start by setting it to null 

291 response = client.put( 

292 f"{settings.API_V1_STR}/tasks/{task_id}", 

293 headers=superuser_token_headers, 

294 json={"scheduled_start": None}, 

295 ) 

296 assert response.status_code == 200 

297 content = response.json() 

298 assert content["scheduled_start"] is None 

299 

300 # verify via GET 

301 response = client.get( 

302 f"{settings.API_V1_STR}/tasks/{task_id}", headers=superuser_token_headers 

303 ) 

304 assert response.status_code == 200 

305 content = response.json() 

306 assert content["scheduled_start"] is None 

307 

308 

309def test_toggle_task_completed( 

310 client: TestClient, superuser_token_headers: dict[str, str] 

311) -> None: 

312 # create a task 

313 title = random_lower_string() 

314 description = random_lower_string() 

315 data = {"title": title, "description": description} 

316 response = client.post( 

317 f"{settings.API_V1_STR}/tasks/", headers=superuser_token_headers, json=data 

318 ) 

319 assert response.status_code == 200 

320 content = response.json() 

321 task_id = content["id"] 

322 initial_completed = content["completed"] 

323 

324 # toggle completed status 

325 response = client.put( 

326 f"{settings.API_V1_STR}/tasks/{task_id}/toggle-completed", 

327 headers=superuser_token_headers, 

328 ) 

329 assert response.status_code == 200 

330 content = response.json() 

331 assert content["completed"] != initial_completed 

332 

333 # toggle again 

334 response = client.put( 

335 f"{settings.API_V1_STR}/tasks/{task_id}/toggle-completed", 

336 headers=superuser_token_headers, 

337 ) 

338 assert response.status_code == 200 

339 content = response.json() 

340 assert content["completed"] == initial_completed 

341 

342 

343# --------------------------------------------------------------------------- 

344# List tasks 

345# --------------------------------------------------------------------------- 

346 

347 

348def test_read_tasks( 

349 client: TestClient, superuser_token_headers: dict[str, str] 

350) -> None: 

351 _create_task(client, superuser_token_headers) 

352 response = client.get( 

353 f"{settings.API_V1_STR}/tasks/", headers=superuser_token_headers 

354 ) 

355 assert response.status_code == 200 

356 content = response.json() 

357 assert content["count"] >= 1 

358 assert len(content["data"]) >= 1 

359 

360 

361def test_read_tasks_normal_user_sees_only_own(client: TestClient, db: Session) -> None: 

362 headers_a = _fresh_user_headers(client, db) 

363 headers_b = _fresh_user_headers(client, db) 

364 

365 task_a = _create_task(client, headers_a, title="Task A") 

366 _create_task(client, headers_b, title="Task B") 

367 

368 resp = client.get(f"{settings.API_V1_STR}/tasks/?limit=500", headers=headers_a) 

369 assert resp.status_code == 200 

370 ids = [t["id"] for t in resp.json()["data"]] 

371 assert task_a["id"] in ids 

372 # user A should not see user B's tasks 

373 assert all(t["owner_id"] == task_a["owner_id"] for t in resp.json()["data"]) 

374 

375 

376# --------------------------------------------------------------------------- 

377# Reorder tasks 

378# --------------------------------------------------------------------------- 

379 

380 

381def test_reorder_tasks(client: TestClient, db: Session) -> None: 

382 headers = _fresh_user_headers(client, db) 

383 t1 = _create_task(client, headers) 

384 t2 = _create_task(client, headers) 

385 t3 = _create_task(client, headers) 

386 

387 # Reorder: t3, t1, t2 

388 resp = client.post( 

389 f"{settings.API_V1_STR}/tasks/reorder", 

390 headers=headers, 

391 json=[t3["id"], t1["id"], t2["id"]], 

392 ) 

393 assert resp.status_code == 200 

394 

395 # Verify positions via list (ordered by position) 

396 resp = client.get(f"{settings.API_V1_STR}/tasks/", headers=headers) 

397 ids = [t["id"] for t in resp.json()["data"]] 

398 assert ids == [t3["id"], t1["id"], t2["id"]] 

399 

400 

401# --------------------------------------------------------------------------- 

402# Error paths 

403# --------------------------------------------------------------------------- 

404 

405 

406def test_read_task_not_found( 

407 client: TestClient, superuser_token_headers: dict[str, str] 

408) -> None: 

409 import uuid 

410 

411 fake_id = str(uuid.uuid4()) 

412 resp = client.get( 

413 f"{settings.API_V1_STR}/tasks/{fake_id}", headers=superuser_token_headers 

414 ) 

415 assert resp.status_code == 404 

416 

417 

418def test_update_task_not_found( 

419 client: TestClient, superuser_token_headers: dict[str, str] 

420) -> None: 

421 import uuid 

422 

423 fake_id = str(uuid.uuid4()) 

424 resp = client.put( 

425 f"{settings.API_V1_STR}/tasks/{fake_id}", 

426 headers=superuser_token_headers, 

427 json={"title": "nope"}, 

428 ) 

429 assert resp.status_code == 404 

430 

431 

432def test_delete_task_not_found( 

433 client: TestClient, superuser_token_headers: dict[str, str] 

434) -> None: 

435 import uuid 

436 

437 fake_id = str(uuid.uuid4()) 

438 resp = client.delete( 

439 f"{settings.API_V1_STR}/tasks/{fake_id}", headers=superuser_token_headers 

440 ) 

441 assert resp.status_code == 404 

442 

443 

444def test_normal_user_cannot_read_other_users_task( 

445 client: TestClient, db: Session 

446) -> None: 

447 headers_a = _fresh_user_headers(client, db) 

448 headers_b = _fresh_user_headers(client, db) 

449 

450 task_a = _create_task(client, headers_a) 

451 resp = client.get(f"{settings.API_V1_STR}/tasks/{task_a['id']}", headers=headers_b) 

452 assert resp.status_code == 400 

453 

454 

455def test_normal_user_cannot_delete_other_users_task( 

456 client: TestClient, db: Session 

457) -> None: 

458 headers_a = _fresh_user_headers(client, db) 

459 headers_b = _fresh_user_headers(client, db) 

460 

461 task_a = _create_task(client, headers_a) 

462 resp = client.delete( 

463 f"{settings.API_V1_STR}/tasks/{task_a['id']}", headers=headers_b 

464 ) 

465 assert resp.status_code == 400 

466 

467 

468# --------------------------------------------------------------------------- 

469# Schedule endpoint tests 

470# --------------------------------------------------------------------------- 

471 

472 

473def _tomorrow_noon() -> str: 

474 """Return tomorrow at noon as an ISO string (no timezone).""" 

475 tomorrow = datetime.utcnow() + timedelta(days=1) 

476 return tomorrow.replace(hour=12, minute=0, second=0, microsecond=0).isoformat() 

477 

478 

479def test_schedule_assigns_scheduled_start(client: TestClient, db: Session) -> None: 

480 """Two tasks with a due date should both get a scheduled_start after scheduling.""" 

481 headers = _fresh_user_headers(client, db) 

482 due = _tomorrow_noon() 

483 task_ids = [] 

484 for _ in range(2): 

485 resp = client.post( 

486 f"{settings.API_V1_STR}/tasks/", 

487 headers=headers, 

488 json={"title": random_lower_string(), "due": due, "duration": 60}, 

489 ) 

490 assert resp.status_code == 200 

491 task_ids.append(resp.json()["id"]) 

492 

493 resp = client.post( 

494 f"{settings.API_V1_STR}/tasks/schedule", 

495 headers=headers, 

496 ) 

497 assert resp.status_code == 200 

498 data = resp.json()["data"] 

499 

500 # Find the two tasks by id 

501 scheduled = {t["id"]: t for t in data if t["id"] in task_ids} 

502 assert len(scheduled) == 2 

503 

504 starts = [] 

505 for tid in task_ids: 

506 t = scheduled[tid] 

507 assert t["scheduled_start"] is not None, f"Task {tid} has no scheduled_start" 

508 starts.append(datetime.fromisoformat(t["scheduled_start"])) 

509 

510 # Verify they don't overlap (each has duration 60 min) 

511 starts.sort() 

512 assert starts[1] >= starts[0] + timedelta(minutes=60), "Tasks overlap" 

513 

514 

515def test_schedule_respects_priority_order(client: TestClient, db: Session) -> None: 

516 """A higher-priority (lower id) task should receive an earlier or equal slot.""" 

517 headers = _fresh_user_headers(client, db) 

518 

519 # Ensure priority rows 1 and 3 exist 

520 for pid, pname in [(1, "urgent"), (3, "low")]: 

521 existing = db.get(Priority, pid) 

522 if not existing: 

523 p = Priority(id=pid, name=pname, value=pid) 

524 db.add(p) 

525 db.commit() 

526 

527 due = _tomorrow_noon() 

528 

529 resp_p1 = client.post( 

530 f"{settings.API_V1_STR}/tasks/", 

531 headers=headers, 

532 json={ 

533 "title": random_lower_string(), 

534 "due": due, 

535 "duration": 60, 

536 "priority_id": 1, 

537 }, 

538 ) 

539 assert resp_p1.status_code == 200 

540 p1_id = resp_p1.json()["id"] 

541 

542 resp_p3 = client.post( 

543 f"{settings.API_V1_STR}/tasks/", 

544 headers=headers, 

545 json={ 

546 "title": random_lower_string(), 

547 "due": due, 

548 "duration": 60, 

549 "priority_id": 3, 

550 }, 

551 ) 

552 assert resp_p3.status_code == 200 

553 p3_id = resp_p3.json()["id"] 

554 

555 resp = client.post( 

556 f"{settings.API_V1_STR}/tasks/schedule", 

557 headers=headers, 

558 ) 

559 assert resp.status_code == 200 

560 data = {t["id"]: t for t in resp.json()["data"]} 

561 

562 p1_start = data[p1_id]["scheduled_start"] 

563 p3_start = data[p3_id]["scheduled_start"] 

564 assert p1_start is not None 

565 assert p3_start is not None 

566 

567 assert datetime.fromisoformat(p1_start) <= datetime.fromisoformat( 

568 p3_start 

569 ), "p1 task should be scheduled at or before p3 task" 

570 

571 

572def test_schedule_respects_working_hours(client: TestClient, db: Session) -> None: 

573 """All scheduled tasks must fall within default working hours (9-18).""" 

574 headers = _fresh_user_headers(client, db) 

575 due = _tomorrow_noon() 

576 client.post( 

577 f"{settings.API_V1_STR}/tasks/", 

578 headers=headers, 

579 json={"title": random_lower_string(), "due": due, "duration": 30}, 

580 ) 

581 

582 resp = client.post( 

583 f"{settings.API_V1_STR}/tasks/schedule", 

584 headers=headers, 

585 ) 

586 assert resp.status_code == 200 

587 

588 for task in resp.json()["data"]: 

589 if task["scheduled_start"] is not None: 

590 hour = datetime.fromisoformat(task["scheduled_start"]).hour 

591 assert hour >= 9, f"Task starts before work_start: hour={hour}" 

592 assert hour < 18, f"Task starts at or after work_end: hour={hour}" 

593 

594 

595def test_schedule_skips_completed_tasks(client: TestClient, db: Session) -> None: 

596 """A completed task must not receive a scheduled_start from the scheduler.""" 

597 headers = _fresh_user_headers(client, db) 

598 due = _tomorrow_noon() 

599 

600 # Create a task then mark it complete 

601 resp = client.post( 

602 f"{settings.API_V1_STR}/tasks/", 

603 headers=headers, 

604 json={"title": random_lower_string(), "due": due, "duration": 30}, 

605 ) 

606 assert resp.status_code == 200 

607 task_id = resp.json()["id"] 

608 

609 resp = client.put( 

610 f"{settings.API_V1_STR}/tasks/{task_id}/toggle-completed", 

611 headers=headers, 

612 ) 

613 assert resp.status_code == 200 

614 assert resp.json()["completed"] is True 

615 

616 resp = client.post( 

617 f"{settings.API_V1_STR}/tasks/schedule", 

618 headers=headers, 

619 ) 

620 assert resp.status_code == 200 

621 data = {t["id"]: t for t in resp.json()["data"]} 

622 

623 assert ( 

624 data[task_id]["scheduled_start"] is None 

625 ), "Completed task should not have scheduled_start set" 

626 

627 

628def test_schedule_only_affects_own_tasks(client: TestClient, db: Session) -> None: 

629 """Scheduling for user A must not modify user B's tasks.""" 

630 headers_a = _fresh_user_headers(client, db) 

631 headers_b = _fresh_user_headers(client, db) 

632 due = _tomorrow_noon() 

633 

634 # User B creates a task 

635 task_b = _create_task(client, headers_b, due=due, duration=30) 

636 

637 # User A triggers schedule 

638 client.post(f"{settings.API_V1_STR}/tasks/schedule", headers=headers_a) 

639 

640 # User B's task should still have no scheduled_start 

641 resp = client.get(f"{settings.API_V1_STR}/tasks/{task_b['id']}", headers=headers_b) 

642 assert resp.status_code == 200 

643 assert resp.json()["scheduled_start"] is None 

644 

645 

646def test_schedule_reschedules_all_incomplete_tasks( 

647 client: TestClient, db: Session 

648) -> None: 

649 """Running schedule twice should reschedule all incomplete tasks (not just unscheduled ones).""" 

650 headers = _fresh_user_headers(client, db) 

651 due = _tomorrow_noon() 

652 

653 _create_task(client, headers, due=due, duration=30) 

654 _create_task(client, headers, due=due, duration=30) 

655 

656 # First schedule 

657 resp = client.post(f"{settings.API_V1_STR}/tasks/schedule", headers=headers) 

658 assert resp.status_code == 200 

659 first_starts = { 

660 t["id"]: t["scheduled_start"] 

661 for t in resp.json()["data"] 

662 if t["scheduled_start"] is not None 

663 } 

664 assert len(first_starts) == 2 

665 

666 # Second schedule — should still set scheduled_start on all incomplete tasks 

667 resp = client.post(f"{settings.API_V1_STR}/tasks/schedule", headers=headers) 

668 assert resp.status_code == 200 

669 second_starts = { 

670 t["id"]: t["scheduled_start"] 

671 for t in resp.json()["data"] 

672 if t["scheduled_start"] is not None 

673 } 

674 assert len(second_starts) == 2