Coverage for src/lexigram/auth/authn/oauth2_session.py: 52%

1"""HTTP session adapters bridging authlib to ``HTTPClientProtocol``. 

2 

3Collaborators of :mod:`lexigram.auth.authn.oauth2`: they normalise requests 

4and responses from any HTTP library (aiohttp, httpx, …) into the interface 

5authlib expects, keeping the OAuth2 layer transport-agnostic. 

6""" 

7 

8from __future__ import annotations 

9 

10from typing import TYPE_CHECKING, Any, Self, cast 

11 

12if TYPE_CHECKING: 

13 from lexigram.contracts.web import HTTPClientProtocol 

14 

15 

16class LexigramConnectSession: 

17 """Session adapter for authlib that delegates to ``HTTPClientProtocol``. 

18 

19 Bridges authlib's internal session interface to Lexigram's 

20 ``HTTPClientProtocol`` contract so the OAuth2 layer remains independent 

21 of the underlying HTTP library (aiohttp, httpx, …). 

22 """ 

23 

24 def __init__(self, http_client: HTTPClientProtocol | None): 

25 self._http_client = http_client 

26 

27 async def request( 

28 self, 

29 method: str, 

30 url: str, 

31 **kwargs: Any, 

32 ) -> LexigramConnectResponse: 

33 """Make a request using the injected ``HTTPClientProtocol``.""" 

34 if self._http_client is None: 

35 raise RuntimeError("HTTP client not configured for LexigramConnectSession") 

36 

37 # Map authlib-style kwargs to the generic HTTPClientProtocol interface 

38 client_kwargs: dict[str, Any] = {} 

39 

40 if "headers" in kwargs: 

41 client_kwargs["headers"] = kwargs["headers"] 

42 if "data" in kwargs: 

43 client_kwargs["data"] = kwargs["data"] 

44 if "json" in kwargs: 

45 client_kwargs["json"] = kwargs["json"] 

46 if "params" in kwargs: 

47 client_kwargs["params"] = kwargs["params"] 

48 if "content" in kwargs: 

49 client_kwargs["data"] = kwargs["content"] 

50 

51 response = await self._http_client.request( 

52 method.upper(), 

53 url, 

54 **client_kwargs, 

55 ) 

56 return LexigramConnectResponse(response) 

57 

58 async def get(self, url: str, **kwargs: Any) -> LexigramConnectResponse: 

59 if self._http_client is None: 

60 raise RuntimeError("HTTP client not configured for LexigramConnectSession") 

61 response = await self._http_client.get(url, **kwargs) 

62 return LexigramConnectResponse(response) 

63 

64 async def post(self, url: str, **kwargs: Any) -> LexigramConnectResponse: 

65 if self._http_client is None: 

66 raise RuntimeError("HTTP client not configured for LexigramConnectSession") 

67 response = await self._http_client.post(url, **kwargs) 

68 return LexigramConnectResponse(response) 

69 

70 async def put(self, url: str, **kwargs: Any) -> LexigramConnectResponse: 

71 if self._http_client is None: 

72 raise RuntimeError("HTTP client not configured for LexigramConnectSession") 

73 response = await self._http_client.put(url, **kwargs) 

74 return LexigramConnectResponse(response) 

75 

76 async def delete(self, url: str, **kwargs: Any) -> LexigramConnectResponse: 

77 if self._http_client is None: 

78 raise RuntimeError("HTTP client not configured for LexigramConnectSession") 

79 response = await self._http_client.delete(url, **kwargs) 

80 return LexigramConnectResponse(response) 

81 

82 async def patch(self, url: str, **kwargs: Any) -> LexigramConnectResponse: 

83 if self._http_client is None: 

84 raise RuntimeError("HTTP client not configured for LexigramConnectSession") 

85 response = await self._http_client.patch(url, **kwargs) 

86 return LexigramConnectResponse(response) 

87 

88 async def head(self, url: str, **kwargs: Any) -> LexigramConnectResponse: 

89 if self._http_client is None: 

90 raise RuntimeError("HTTP client not configured for LexigramConnectSession") 

91 response = await self._http_client.head(url, **kwargs) 

92 return LexigramConnectResponse(response) 

93 

94 

95class LexigramConnectResponse: 

96 """Response adapter that makes any HTTP response compatible with authlib. 

97 

98 Normalises response objects from different HTTP libraries 

99 (aiohttp, httpx, …) into the interface expected by authlib. 

100 """ 

101 

102 def __init__(self, response: Any): 

103 self._response = response 

104 

105 @property 

106 def status_code(self) -> int: 

107 # httpx exposes `.status_code`; aiohttp exposes `.status`. 

108 # Check both, preferring `.status_code`, and accept only genuine ints 

109 # so that MagicMock auto-attributes (not ints) are skipped correctly. 

110 for attr in ("status_code", "status"): 

111 code = getattr(self._response, attr, None) 

112 if isinstance(code, int): 

113 return code 

114 raise AttributeError( 

115 f"Response object {type(self._response).__name__!r} has no " 

116 f"integer 'status_code' or 'status' attribute" 

117 ) 

118 

119 @property 

120 def headers(self) -> dict[str, str]: 

121 return {str(k): str(v) for k, v in dict(self._response.headers).items()} 

122 

123 async def json(self) -> dict[str, Any]: 

124 return cast("dict[str, Any]", await self._response.json()) 

125 

126 async def text(self) -> str: 

127 return str(await self._response.text()) 

128 

129 def raise_for_status(self) -> None: 

130 self._response.raise_for_status() 

131 

132 async def close(self) -> None: 

133 """Close the underlying response if the client supports it.""" 

134 close = getattr(self._response, "close", None) or getattr( 

135 self._response, "release", None 

136 ) 

137 if callable(close): 

138 await close() 

139 

140 async def __aenter__(self) -> Self: 

141 return self 

142 

143 async def __aexit__( 

144 self, 

145 exc_type: type[BaseException] | None, 

146 exc_val: BaseException | None, 

147 exc_tb: object, 

148 ) -> None: 

149 await self.close() 

150 

151 

152__all__ = [ 

153 "LexigramConnectResponse", 

154 "LexigramConnectSession", 

155]