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

1import dataclasses 

2from collections.abc import Sequence 

3from typing import Protocol 

4 

5from starlette.datastructures import Headers 

6from starlette.middleware.cors import CORSMiddleware 

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

8 

9 

10class CORSMatcher(Protocol): 

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

12 

13 

14@dataclasses.dataclass 

15class CORSConfig: 

16 matcher: CORSMatcher 

17 allow_origins: Sequence[str] = () 

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

19 allow_headers: Sequence[str] = () 

20 allow_credentials: bool = False 

21 allow_origin_regex: str | None = None 

22 expose_headers: Sequence[str] = () 

23 max_age: int = 600 

24 

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

26 return CORSMiddleware( 

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: 

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

40 self.app = app 

41 self.config_middlewares = tuple( 

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

43 ) 

44 

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

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

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

48 return 

49 

50 method = scope["method"] 

51 headers = Headers(scope=scope) 

52 origin = headers.get("origin") 

53 

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

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

56 return 

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( 

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"]