Coverage for routers/vehicles.py: 100%

1from fastapi import APIRouter, Depends, HTTPException, Response 

2import duckdb 

3from typing import List 

4 

5from database import get_db 

6import schemas 

7 

8router = APIRouter( 

9 prefix="/api/v1", 

10 tags=["Vehicles"] 

11) 

12 

13@router.get("/vehicles/{vrm}", response_model=schemas.VehicleResponse) 

14def get_vehicle(vrm: str, response: Response, db: duckdb.DuckDBPyConnection = Depends(get_db)): 

15 # Setting Cache-Control headers for semi-static data (1 day cache) 

16 response.headers["Cache-Control"] = "public, max-age=86400, stale-while-revalidate=3600" 

17 

18 # Query using DuckDB 

19 query = """ 

20 SELECT vrm, make, model, year, color, mot_expiry, generation_id 

21 FROM vehicles 

22 WHERE vrm = ? 

23 LIMIT 1 

24 """ 

25 db.execute(query, [vrm.upper()]) 

26 result = db.fetchone() 

27 

28 if not result: 

29 raise HTTPException(status_code=404, detail="Vehicle not found") 

30 

31 columns = [desc[0] for desc in db.description] 

32 vehicle = dict(zip(columns, result)) 

33 

34 return vehicle 

35 

36@router.get("/vehicles/{vrm}/mot-history", response_model=schemas.VehicleMOTHistoryResponse) 

37def get_mot_history(vrm: str, response: Response, db: duckdb.DuckDBPyConnection = Depends(get_db)): 

38 # Setting Cache-Control headers 

39 response.headers["Cache-Control"] = "public, max-age=86400, stale-while-revalidate=3600" 

40 

41 query = """ 

42 SELECT  

43 t.test_id as test_number,  

44 t.test_date,  

45 t.result,  

46 t.odometer as odometer_reading, 

47 'mi' as odometer_unit, 

48 list( 

49 CASE WHEN i.type IS NOT NULL THEN 

50 {'item_type': i.type, 'description': d.description_text, 'pos_code': NULL} 

51 ELSE NULL END 

52 ) FILTER (WHERE i.type IS NOT NULL) as items 

53 FROM mot_tests t 

54 LEFT JOIN mot_items i ON t.test_id = i.test_id AND t.test_date = i.test_date 

55 LEFT JOIN defect_descriptions d ON i.description_id = d.description_id 

56 WHERE t.vrm = ? 

57 GROUP BY t.test_id, t.test_date, t.result, t.odometer 

58 ORDER BY t.test_date DESC 

59 """ 

60 db.execute(query, [vrm.upper()]) 

61 rows = db.fetchall() 

62 

63 if not rows: 

64 raise HTTPException(status_code=404, detail="MOT history not found for this vehicle") 

65 

66 columns = [desc[0] for desc in db.description] 

67 

68 tests = [] 

69 for row_tuple in rows: 

70 row = dict(zip(columns, row_tuple)) 

71 row['test_number'] = str(row['test_number']) 

72 tests.append(row) 

73 

74 return { 

75 "vrm": vrm.upper(), 

76 "tests": tests 

77 }