Coverage for test\test_session.py: 93%

1from http import cookies 

2import os 

3import json 

4import pytest 

5from datetime import datetime, timedelta, timezone 

6from nexios import get_application, NexiosApp 

7from nexios.http import Request, Response 

8from nexios.session.base import BaseSessionInterface 

9from nexios.session.file import FileSessionManager 

10from nexios.session.signed_cookies import SignedSessionManager 

11from nexios.testing import Client 

12from nexios.config import MakeConfig, get_config, set_config 

13from typing import Tuple, Dict, Any 

14 

15 

16# Fixtures for different session configurations 

17@pytest.fixture 

18async def file_session_client(tmp_path) -> Tuple[Client, NexiosApp]: 

19 """Client with file-based session configuration""" 

20 app = get_application( 

21 MakeConfig( 

22 { 

23 "secret_key": "file_session_secret", 

24 "session": { 

25 "session_cookie_name": "file_session", 

26 "session_permanent": True, 

27 "session_expiration_time": 30, 

28 "manager": FileSessionManager, 

29 }, 

30 "session_file_name": str(tmp_path / "sessions"), 

31 "SESSION_FILE_STORAGE_PATH": str(tmp_path / "sessions"), 

32 } 

33 ) 

34 ) 

35 async with Client(app) as client: 

36 yield client, app 

37 

38 

39@pytest.fixture 

40async def signed_session_client() -> Tuple[Client, NexiosApp]: 

41 """Client with signed cookie session configuration""" 

42 app = get_application( 

43 MakeConfig( 

44 { 

45 "secret_key": "signed_session_secret", 

46 "session": { 

47 "session_cookie_name": "signed_session", 

48 "session_permanent": True, 

49 "session_expiration_time": 30, 

50 "manager": SignedSessionManager, 

51 }, 

52 } 

53 ) 

54 ) 

55 async with Client(app) as client: 

56 yield client, app 

57 

58 

59# # Test session middleware 

60async def test_session_middleware_initialization( 

61 file_session_client: Tuple[Client, NexiosApp], 

62): 

63 client, app = file_session_client 

64 

65 @app.get("/test-session") 

66 async def test_session(req: Request, res: Response): 

67 

68 req.session["key"] = "value" 

69 return res.text("OK") 

70 

71 response = await client.get("/test-session") 

72 assert response.status_code == 200 

73 assert response.text == "OK" 

74 

75 

76async def test_session_middleware_no_secret_key(): 

77 app = get_application(MakeConfig({"secret_key": None})) 

78 

79 @app.get("/test-session") 

80 async def test_session(req: Request, res: Response): 

81 try: 

82 req.session["key"] = "value" 

83 except AssertionError: 

84 return res.text("No session", status_code=512) 

85 return res.text("OK") 

86 

87 async with Client(app) as client: 

88 response = await client.get("/test-session") 

89 assert response.status_code == 512 

90 assert response.text == "No session" 

91 

92 

93# # Test file session manager 

94# async def test_file_session_operations( 

95# file_session_client: Tuple[Client, NexiosApp], tmp_path 

96# ): 

97# client, app = file_session_client 

98 

99# @app.get("/set-session") 

100# async def set_session(req: Request, res: Response): 

101# req.session["test_key"] = "test_value" 

102# req.session["user"] = {"id": 1, "name": "Test"} 

103# return res.text("Session set") 

104 

105# @app.get("/get-session") 

106# async def get_session(req: Request, res: Response): 

107# return res.json( 

108# {"test_key": req.session.get("test_key"), "user": req.session.get("user")} 

109# ) 

110 

111# @app.get("/delete-session") 

112# async def delete_session(req: Request, res: Response): 

113# del req.session["test_key"] 

114# return res.text("Session deleted") 

115 

116# @app.get("/clear-session") 

117# async def clear_session(req: Request, res: Response): 

118# req.session.clear() 

119# return res.text("Session cleared") 

120 

121# # # Set session 

122# response = await client.get("/set-session") 

123# assert response.status_code == 200 

124 

125# # Verify cookie was set 

126# assert "file_session" in response.cookies 

127 

128# session_id = response.cookies["file_session"] 

129 

130# # # Verify session file was created 

131# session_file = tmp_path / "sessions" / f"{session_id}.json" 

132# assert session_file.exists() 

133 

134# response = await client.get("/get-session") 

135# assert response.status_code == 200 

136# # assert response.json() == { 

137# # "test_key": "test_value", 

138# # "user": {"id": 1, "name": "Test"}, 

139# # } 

140 

141# # Delete item from session 

142# response = await client.get("/delete-session") 

143# # assert response.status_code == 200 

144 

145# # Verify deletion 

146# response = await client.get("/get-session") 

147# # assert response.json()["test_key"] is None 

148 

149# # Clear session 

150# response = await client.get("/clear-session") 

151# # assert response.status_code == 200 

152# # assert not session_file.exists() 

153 

154 

155# # Test signed cookie session manager 

156# async def test_signed_session_operations(signed_session_client: Tuple[Client, NexiosApp]): 

157# client, app = signed_session_client 

158 

159# @app.get("/set-session") 

160# async def set_session(req: Request, res: Response): 

161# req.session["test_key"] = "test_value" 

162# req.session["user"] = {"id": 1, "name": "Test"} 

163# return res.text("Session set") 

164 

165# @app.get("/get-session") 

166# async def get_session(req: Request, res: Response): 

167# return res.json({ 

168# "test_key": req.session.get("test_key"), 

169# "user": req.session.get("user") 

170# }) 

171 

172# # Set session 

173# response = await client.get("/set-session") 

174# assert response.status_code == 200 

175 

176# # Verify cookie was set 

177# assert "signed_session" in response.cookies 

178# session_cookie = response.cookies["signed_session"] 

179 

180# # Get session 

181# response = await client.get("/get-session") 

182# assert response.status_code == 200 

183# assert response.json() == { 

184# "test_key": "test_value", 

185# "user": {"id": 1, "name": "Test"} 

186# } 

187 

188# # Test with invalid cookie 

189# client.cookies["signed_session"] = "invalid.token" 

190# response = await client.get("/get-session") 

191# assert response.status_code == 200 

192# assert response.json() == { 

193# "test_key": None, 

194# "user": None 

195# } 

196 

197# # Test session expiration 

198# async def test_session_expiration(file_session_client: Tuple[Client, NexiosApp]): 

199# client, app = file_session_client 

200 

201# @app.get("/set-expiring-session") 

202# async def set_expiring_session(req: Request, res: Response): 

203# req.session["temp"] = "data" 

204# # Set expiration to 1 second from now 

205# req.session._session_cache["__expires"] = ( 

206# datetime.now(timezone.utc) + timedelta(seconds=1) 

207# ).isoformat() 

208# return res.text("Session set") 

209 

210# @app.get("/check-expired") 

211# async def check_expired(req: Request, res: Response): 

212# return res.json({"expired": req.session.has_expired()}) 

213 

214# # Set session 

215# response = await client.get("/set-expiring-session") 

216# assert response.status_code == 200 

217 

218# # Check immediately - should not be expired 

219# response = await client.get("/check-expired") 

220# assert response.status_code == 200 

221# assert response.json()["expired"] is False 

222 

223# # Wait for expiration 

224# import time 

225# time.sleep(2) 

226 

227# # Check again - should be expired 

228# response = await client.get("/check-expired") 

229# assert response.status_code == 200 

230# assert response.json()["expired"] is True 

231 

232# # Test session cookie settings 

233# async def test_session_cookie_settings(file_session_client: Tuple[Client, NexiosApp]): 

234# client, app = file_session_client 

235 

236# # Update cookie settings 

237# app.config.session.session_cookie_httponly = True 

238# app.config.session.session_cookie_secure = True 

239# app.config.session.session_cookie_samesite = "lax" 

240# app.config.session.session_cookie_path = "/test" 

241# app.config.session.session_cookie_domain = "example.com" 

242 

243# @app.get("/set-cookie-settings") 

244# async def set_cookie_settings(req: Request, res: Response): 

245# req.session["test"] = "value" 

246# return res.text("OK") 

247 

248# response = await client.get("/set-cookie-settings") 

249# assert response.status_code == 200 

250 

251 

252# cookie = response.cookies["file_session"] 

253# assert cookie["httponly"] is True 

254# assert cookie["secure"] is True 

255# assert cookie["samesite"] == "lax" 

256# assert cookie["path"] == "/test" 

257# assert cookie["domain"] == "example.com" 

258# async def test_signed_session_operations( 

259# signed_session_client: Tuple[Client, NexiosApp], 

260# ): 

261# client, app = signed_session_client 

262 

263# @app.get("/set-session") 

264# async def set_session(req: Request, res: Response): 

265# req.session["test_key"] = "test_value" 

266# req.session["user"] = {"id": 1, "name": "Test"} 

267# return res.text("Session set") 

268 

269# @app.get("/get-session") 

270# async def get_session(req: Request, res: Response): 

271# return res.json( 

272# {"test_key": req.session.get("test_key"), "user": req.session.get("user")} 

273# ) 

274 

275# # Set session 

276# response = await client.get("/set-session") 

277# assert response.status_code == 200 

278 

279# # Verify cookie was set 

280# assert "signed_session" in response.cookies 

281# session_cookie = response.cookies["signed_session"] 

282 

283# # Get session 

284# response = await client.get("/get-session") 

285# assert response.status_code == 200 

286# assert response.json() == { 

287# "test_key": "test_value", 

288# "user": {"id": 1, "name": "Test"}, 

289# } 

290 

291# # Test with invalid cookie 

292# client.cookies["signed_session"] = "invalid.token" 

293# response = await client.get("/get-session") 

294# assert response.status_code == 200 

295# assert response.json() == {"test_key": None, "user": None} 

296 

297 

298# Test session cookie settings 

299async def test_session_cookie_settings(file_session_client: Tuple[Client, NexiosApp]): 

300 client, app = file_session_client 

301 

302 # Update cookie settings 

303 app.config.session.session_cookie_httponly = True 

304 app.config.session.session_permanent = False 

305 

306 @app.get("/set-cookie-settings") 

307 async def set_cookie_settings(req: Request, res: Response): 

308 req.session["test"] = "value" 

309 return res.text("OK") 

310 

311 response = await client.get("/set-cookie-settings") 

312 assert response.status_code == 200 

313 

314 cookie = response.cookies["file_session"] 

315 # assert cookie["httponly"] is True 

316 # assert cookie["secure"] is True 

317 # assert cookie["samesite"] == "lax" 

318 # assert cookie["path"] == "/test" 

319 # assert cookie["domain"] == "example.com" 

320 

321 

322# # Test session middleware with custom manager 

323# async def test_custom_session_manager(file_session_client: Tuple[Client, NexiosApp]): 

324# # Define a simple in-memory session manager for testing 

325# class MemorySessionManager(BaseSessionInterface): 

326# _store: Dict[str, Dict[str, Any]] = {} 

327 

328# async def load(self): 

329# self._session_cache = self._store.get(self.session_key, {}) 

330 

331# async def save(self): 

332# self._store[self.session_key] = self._session_cache 

333 

334# # app = get_application(MakeConfig({ 

335# # "secret_key": "custom_session_secret", 

336# # "session": { 

337# # "manager": MemorySessionManager, 

338# # "session_cookie_name": "custom_session" 

339# # } 

340# # })) 

341# client, app = file_session_client 

342# app.config.session.manager = MemorySessionManager 

343# app.config.session.session_cookie_name = "custom" 

344 

345# @app.get("/test-custom-manager") 

346# async def test_custom_manager(req: Request, res: Response): 

347 

348# if "count" not in req.session: 

349# req.session["count"] = 1 

350# else: 

351 

352# req.session["count"] += 1 

353# return res.json({"count": req.session["count"]}) 

354 

355# async with Client(app) as client: 

356# # First request 

357# response = await client.get("/test-custom-manager") 

358# assert response.status_code == 200 

359# assert response.json()["count"] == 1 

360 

361# # Second request 

362# response = await client.get("/test-custom-manager") 

363# assert response.status_code == 200 

364# assert response.json()["count"] == 2 

365 

366# # New client should start fresh 

367# async with Client(app) as new_client: 

368# response = await new_client.get("/test-custom-manager") 

369# assert response.status_code == 200 

370# assert response.json()["count"] == 1