Coverage for polar/kit/cors.py: 62%

1import dataclasses 1 ctx1a

2from collections.abc import Sequence 1 ctx1a

3from typing import Protocol 1 ctx1a

4 

5from starlette.datastructures import Headers 1 ctx1a

6from starlette.middleware.cors import CORSMiddleware 1 ctx1a

7from starlette.types import ASGIApp, Receive, Scope, Send 1 ctx1a

8 

9 

10class CORSMatcher(Protocol): 1 ctx1a

11 def __call__(self, origin: str, scope: Scope) -> bool: ... 11 ↛ exitline 11 didn't return from function '__call__' because 1 ctx1a

12 

13 

14@dataclasses.dataclass 1 ctx1a

15class CORSConfig: 1 ctx1a

16 matcher: CORSMatcher 1 ctx1a

17 allow_origins: Sequence[str] = () 1 ctx1a

18 allow_methods: Sequence[str] = ("GET",) 1 ctx1a

19 allow_headers: Sequence[str] = () 1 ctx1a

20 allow_credentials: bool = False 1 ctx1a

21 allow_origin_regex: str | None = None 1 ctx1a

22 expose_headers: Sequence[str] = () 1 ctx1a

23 max_age: int = 600 1 ctx1a

24 

25 def get_middleware(self, app: ASGIApp) -> CORSMiddleware: 1 ctx1a

26 return CORSMiddleware( (empty)

27 app=app, 

28 allow_origins=self.allow_origins, 

29 allow_methods=self.allow_methods, 

30 allow_headers=self.allow_headers, 

31 allow_credentials=self.allow_credentials, 

32 allow_origin_regex=self.allow_origin_regex, 

33 expose_headers=self.expose_headers, 

34 max_age=self.max_age, 

35 ) 

36 

37 

38class CORSMatcherMiddleware: 1 ctx1a

39 def __init__(self, app: ASGIApp, *, configs: Sequence[CORSConfig]) -> None: 1 ctx1a

40 self.app = app (empty)

41 self.config_middlewares = tuple( (empty)

42 (config, config.get_middleware(app)) for config in configs 

43 ) 

44 

45 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: 1 ctx1a

46 if scope["type"] != "http": # pragma: no cover 2 ctx1cb

47 await self.app(scope, receive, send) (empty)

48 return 

49 

50 method = scope["method"] 2 ctx1cb

51 headers = Headers(scope=scope) 2 ctx1cb

52 origin = headers.get("origin") 2 ctx1cb

53 

54 if origin is None: 54 ↛ 58line 54 didn't jump to line 58 because the condition on line 54 was always true2 ctx1cb

55 await self.app(scope, receive, send) 2 ctx1cb

56 return 2 ctx1cb

57 

58 middleware = self._get_config_middleware(origin, scope) 

59 if middleware is None: 

60 await self.app(scope, receive, send) 

61 return 

62 

63 if method == "OPTIONS" and "access-control-request-method" in headers: 

64 response = middleware.preflight_response(request_headers=headers) 

65 await response(scope, receive, send) 

66 return 

67 await middleware.simple_response(scope, receive, send, request_headers=headers) 

68 

69 def _get_config_middleware( 1 ctx1a

70 self, origin: str, scope: Scope 

71 ) -> CORSMiddleware | None: 

72 for config, middleware in self.config_middlewares: 

73 if config.matcher(origin, scope): 

74 return middleware 

75 return None 

76 

77 

78__all__ = ["CORSConfig", "CORSMatcherMiddleware", "Scope"] 1 ctx1a