Coverage for app/utils/check_token.py: 83%

1import requests 

2from app.config.config import POSTGRES_HOSTNAME 

3 

4from app.utils.response_format import generate_error_response 

5from app.utils.typing import JSONObject, Optional 

6from fastapi.responses import JSONResponse 

7 

8constellation_check_error: dict[int, JSONObject] = { 

9 401: { 

10 "description": "Unauthorized", 

11 "content": { 

12 "application/json": { 

13 "example": { 

14 "success": False, 

15 "data": None, 

16 "message": "Could not validate credentials", 

17 "error": { 

18 "code": "INVALID_CREDENTIALS", 

19 "message": "Could not validate credentials" 

20 } 

21 } 

22 } 

23 } 

24 }, 

25 403: { 

26 "description": "Forbidden: Insufficient permissions", 

27 "content": { 

28 "application/json": { 

29 "example": { 

30 "success": False, 

31 "data": None, 

32 "message": "Forbidden: Insufficient permissions", 

33 "error": { 

34 "code": "INSUFFICIENT_PERMISSIONS", 

35 "message": "Forbidden: Insufficient permissions" 

36 } 

37 } 

38 } 

39 } 

40 }, 

41 404: { 

42 "description": "Not Found", 

43 "content": { 

44 "application/json": { 

45 "example": { 

46 "success": False, 

47 "data": None, 

48 "message": "... not found", 

49 "error": { 

50 "code": "NOT_FOUND", 

51 "message": "... not found" 

52 } 

53 } 

54 } 

55 } 

56 } 

57} 

58 

59data_permissions = [ 

60 "READ", 

61 "COMMENT", 

62 "WRITE", 

63 "ADMIN", 

64 "OWNER" 

65] 

66 

67def check_constellation_access(token: str, constellation_uuid: str, asked_permission: str = "READ") -> Optional[JSONResponse]: 

68 try: 

69 response = requests.get(f"{POSTGRES_HOSTNAME}/me/constellations/{constellation_uuid}/access", headers={"Authorization": f"Bearer {token}"}) 

70 if response.status_code != 200: 

71 return generate_error_response( 

72 status_code=response.status_code, 

73 error_code="INVALID_CREDENTIALS", 

74 error_message="Could not validate credentials", 

75 message="Could not validate credentials" 

76 ) 

77 data = response.json() 

78 if data_permissions.index(data["access"]) < data_permissions.index(asked_permission): 

79 return generate_error_response( 

80 status_code=403, 

81 error_code="INSUFFICIENT_PERMISSIONS", 

82 error_message="Forbidden: Insufficient permissions", 

83 message="Forbidden: Insufficient permissions" 

84 ) 

85 except requests.exceptions.RequestException as e: 

86 return generate_error_response( 

87 status_code=500, 

88 error_code="INTERNAL_SERVER_ERROR", 

89 error_message="Error while checking token", 

90 message=f"Error while checking token: {e}" 

91 ) 

92 return None