Coverage for or_dependencies.py: 82%

1import json 

2import logging 

3from typing import Callable 

4 

5from fastapi import Depends, Security 

6from fastapi.exceptions import RequestValidationError 

7from fastapi.routing import APIRoute 

8from fastapi.security import SecurityScopes 

9from starlette import status 

10from starlette.exceptions import HTTPException 

11from starlette.requests import Request 

12from starlette.responses import Response, JSONResponse 

13 

14import schemas 

15from chalicelib.utils import helper 

16from chalicelib.utils.log import sanitize 

17 

18logger = logging.getLogger(__name__) 

19 

20 

21async def OR_context(request: Request) -> schemas.CurrentContext: 

22 if hasattr(request.state, "currentContext"): 22 ↛ 25line 22 didn't jump to line 25 because the condition on line 22 was always true

23 return request.state.currentContext 

24 else: 

25 raise Exception("currentContext not found") 

26 

27 

28class ORRoute(APIRoute): 

29 def get_route_handler(self) -> Callable: 

30 original_route_handler = super().get_route_handler() 

31 

32 async def custom_route_handler(request: Request) -> Response: 

33 try: 

34 response: Response = await original_route_handler(request) 

35 except RequestValidationError as exc: 

36 # 422 validation exception 

37 logger.warning( 

38 f"!!! 422 exception when calling: {sanitize(request.method, max_length=16)} {sanitize(str(request.url))}") 

39 logger.warning(exc.errors()) 

40 for e in exc.errors(): 

41 if e.get("msg", "").endswith("must be alphanumeric"): 

42 return JSONResponse(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, 

43 content={"errors": [e["msg"][18:]], "detail": str(exc)}) 

44 raise exc 

45 except HTTPException as e: 

46 if e.status_code // 100 == 4: 46 ↛ 49line 46 didn't jump to line 49 because the condition on line 46 was always true

47 return JSONResponse(content={"errors": e.detail if isinstance(e.detail, list) else [e.detail]}, 

48 status_code=e.status_code) 

49 elif e.status_code // 100 == 5: 

50 logger.error(f"!!! status code:{e.status_code}") 

51 logger.exception(e) 

52 return JSONResponse(content={"errors": ["Internal server error."]}, 

53 status_code=e.status_code) 

54 else: 

55 raise e 

56 except Exception as e: 

57 logger.exception(e) 

58 return JSONResponse(content={"errors": ["Internal server error."]}, 

59 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR) 

60 

61 if isinstance(response, JSONResponse): 61 ↛ 74line 61 didn't jump to line 74 because the condition on line 61 was always true

62 response: JSONResponse = response 

63 body = json.loads(response.body.decode('utf8')) 

64 response.body = response.render(helper.cast_session_id_to_string(body)) 

65 response.headers["Content-Length"] = str(len(response.body)) 

66 if response.status_code == 200 \ 

67 and body is not None and isinstance(body, dict) \ 

68 and body.get("errors") is not None: 

69 if "not found" in body["errors"][0]: 

70 response.status_code = status.HTTP_404_NOT_FOUND 

71 else: 

72 response.status_code = status.HTTP_400_BAD_REQUEST 

73 

74 return response 

75 

76 return custom_route_handler 

77 

78 

79def __check_role(required_roles: SecurityScopes, context: schemas.CurrentContext = Depends(OR_context)): 

80 if len(required_roles.scopes) > 0: 80 ↛ exitline 80 didn't return from function '__check_role' because the condition on line 80 was always true

81 if context.role not in required_roles.scopes: 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true

82 raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, 

83 detail="You need a different role to access this resource") 

84 

85 

86def OR_role(*required_roles): 

87 return Security(__check_role, scopes=list(required_roles))