Coverage for test\test_auth.py: 96%

1import pytest 

2from nexios import get_application 

3from nexios.auth.backends.jwt import create_jwt, decode_jwt 

4from nexios.auth.decorator import auth 

5from nexios.config.base import MakeConfig 

6from nexios.http import Request, Response 

7from nexios.testing import Client 

8from nexios.auth.base import AuthenticationBackend, SimpleUser, UnauthenticatedUser 

9 

10 

11@pytest.fixture 

12async def test_client(): 

13 app = get_application(MakeConfig({"secret_key": "1234"})) 

14 async with Client(app) as client: 

15 yield client, app 

16 

17 

18@pytest.fixture 

19def mock_user(): 

20 return {"id": 1, "username": "testuser"} 

21 

22 

23@pytest.fixture 

24def valid_token(mock_user): 

25 return create_jwt(mock_user) 

26 

27 

28@pytest.fixture 

29def expired_token(mock_user): 

30 return create_jwt({"exp": 1, **mock_user}) 

31 

32 

33async def test_jwt_auth_success(test_client, mock_user, valid_token): 

34 client, app = test_client 

35 

36 async def mock_authenticate(**kwargs): 

37 return mock_user 

38 

39 from nexios.auth.backends import JWTAuthBackend 

40 from nexios.auth.middleware import AuthenticationMiddleware 

41 

42 app.add_middleware( 

43 AuthenticationMiddleware( 

44 backend=JWTAuthBackend(authenticate_func=mock_authenticate) 

45 ) 

46 ) 

47 

48 @app.get("/protected") 

49 @auth(["jwt"]) 

50 async def protected_route(req: Request, res: Response): 

51 return res.json({"user": req.user}) 

52 

53 response = await client.get( 

54 "/protected", headers={"Authorization": f"Bearer {valid_token}"} 

55 ) 

56 

57 assert response.status_code == 200 

58 assert response.json()["user"] == mock_user 

59 

60 

61async def test_jwt_auth_missing_header(test_client, mock_user): 

62 client, app = test_client 

63 

64 async def mock_authenticate(**kwargs): 

65 return mock_user 

66 

67 from nexios.auth.backends import JWTAuthBackend 

68 from nexios.auth.middleware import AuthenticationMiddleware 

69 

70 app.add_middleware( 

71 AuthenticationMiddleware( 

72 backend=JWTAuthBackend(authenticate_func=mock_authenticate) 

73 ) 

74 ) 

75 

76 @app.get("/protected") 

77 @auth(["jwt"]) 

78 async def protected_route(req: Request, res: Response): 

79 return res.json({"user": req.user}) 

80 

81 # Test without auth header 

82 response = await client.get("/protected") 

83 

84 assert response.status_code == 401 

85 

86 

87async def test_jwt_auth_invalid_token(test_client, mock_user): 

88 client, app = test_client 

89 

90 async def mock_authenticate(**kwargs): 

91 return mock_user 

92 

93 from nexios.auth.backends import JWTAuthBackend 

94 from nexios.auth.middleware import AuthenticationMiddleware 

95 

96 app.add_middleware( 

97 AuthenticationMiddleware( 

98 backend=JWTAuthBackend(authenticate_func=mock_authenticate) 

99 ) 

100 ) 

101 

102 @app.get("/protected") 

103 @auth(["jwt"]) 

104 async def protected_route(req: Request, res: Response): 

105 return res.json({"user": req.user}) 

106 

107 # Test with invalid token 

108 response = await client.get( 

109 "/protected", headers={"Authorization": "Bearer invalid_token"} 

110 ) 

111 

112 assert response.status_code == 401 

113 

114 

115async def test_jwt_auth_expired_token(test_client, mock_user, expired_token): 

116 client, app = test_client 

117 

118 async def mock_authenticate(**kwargs): 

119 return mock_user 

120 

121 from nexios.auth.backends import JWTAuthBackend 

122 from nexios.auth.middleware import AuthenticationMiddleware 

123 

124 app.add_middleware( 

125 AuthenticationMiddleware( 

126 backend=JWTAuthBackend(authenticate_func=mock_authenticate) 

127 ) 

128 ) 

129 

130 @app.get("/protected") 

131 @auth(["jwt"]) 

132 async def protected_route(req: Request, res: Response): 

133 return res.json({"user": req.user}) 

134 

135 # Test with expired token 

136 response = await client.get( 

137 "/protected", headers={"Authorization": f"Bearer {expired_token}"} 

138 ) 

139 

140 assert response.status_code == 401 

141 

142 

143async def test_jwt_auth_validation_failure(test_client, valid_token): 

144 client, app = test_client 

145 

146 # Mock authenticate function to return None (invalid user) 

147 async def mock_authenticate(**kwargs): 

148 return SimpleUser(username="nexios-dev") 

149 

150 from nexios.auth.backends import JWTAuthBackend 

151 from nexios.auth.middleware import AuthenticationMiddleware 

152 

153 app.add_middleware( 

154 AuthenticationMiddleware( 

155 backend=JWTAuthBackend(authenticate_func=mock_authenticate) 

156 ) 

157 ) 

158 

159 @app.get("/protected") 

160 @auth(["jwt"]) 

161 async def protected_route(req: Request, res: Response): 

162 return res.json({"user": req.user}) 

163 

164 # Test with valid token but invalid user 

165 response = await client.get( 

166 "/protected", headers={"Authorization": f"Bearer {valid_token}"} 

167 ) 

168 

169 assert response.status_code == 200 

170 

171 

172async def test_jwt_auth_with_auth_decorator(test_client, mock_user, valid_token): 

173 client, app = test_client 

174 

175 async def mock_authenticate(**kwargs): 

176 return mock_user 

177 

178 from nexios.auth.backends import JWTAuthBackend 

179 from nexios.auth.middleware import AuthenticationMiddleware 

180 from nexios.auth.decorator import auth 

181 

182 app.add_middleware( 

183 AuthenticationMiddleware( 

184 backend=JWTAuthBackend(authenticate_func=mock_authenticate) 

185 ) 

186 ) 

187 

188 @app.get("/protected-decorator") 

189 @auth(["jwt"]) 

190 async def protected_route(req: Request, res: Response): 

191 return res.json({"user": req.user}) 

192 

193 # Test with valid token 

194 response = await client.get( 

195 "/protected-decorator", headers={"Authorization": f"Bearer {valid_token}"} 

196 ) 

197 assert response.status_code == 200 

198 assert response.json()["user"] == mock_user 

199 

200 # Test without token (should be unauthorized) 

201 response = await client.get("/protected-decorator") 

202 assert response.status_code == 401 

203 

204 

205def test_create_jwt(): 

206 from jwt import decode as jwt_decode 

207 from jwt import PyJWT 

208 

209 payload = {"user_id": 1, "username": "test"} 

210 token = create_jwt(payload, "test_secret") 

211 

212 decoded = jwt_decode(token, "test_secret", algorithms=["HS256"]) 

213 assert decoded["user_id"] == 1 

214 assert decoded["username"] == "test" 

215 

216 

217def test_decode_jwt_valid(): 

218 payload = {"user_id": 1, "username": "test"} 

219 token = create_jwt(payload, "test_secret", algorithm="HS256") 

220 

221 decoded = decode_jwt(token, "test_secret", ["HS256"]) 

222 assert decoded["user_id"] == 1 

223 assert decoded["username"] == "test" 

224 

225 

226def test_decode_jwt_expired(): 

227 payload = {"user_id": 1, "username": "test", "exp": 1} # Expired in 1970 

228 token = create_jwt(payload, "test_secret", algorithm="HS256") 

229 

230 with pytest.raises(ValueError, match="Token has expired"): 

231 decode_jwt(token, "test_secret", ["HS256"]) 

232 

233 

234def test_decode_jwt_invalid(): 

235 with pytest.raises(ValueError, match="Invalid token"): 

236 decode_jwt("invalid.token", "test_secret", ["HS256"]) 

237 

238 

239async def test_custom_auth_backend(test_client): 

240 client, app = test_client 

241 

242 class CustomAuthBackend(AuthenticationBackend): 

243 async def authenticate(self, request: Request, response: Response): 

244 if request.headers.get("X-Custom-Auth") == "valid": 

245 return {"id": 1, "username": "custom_user"}, "X-auth" 

246 return None 

247 

248 from nexios.auth.middleware import AuthenticationMiddleware 

249 

250 app.add_middleware(AuthenticationMiddleware(backend=CustomAuthBackend())) 

251 

252 @app.get("/custom-protected") 

253 @auth(["X-auth"]) 

254 async def custom_protected(req: Request, res: Response): 

255 return res.json({"user": req.user}) 

256 

257 # Test with valid custom auth 

258 response = await client.get("/custom-protected", headers={"X-Custom-Auth": "valid"}) 

259 assert response.status_code == 200 

260 assert response.json()["user"] == {"id": 1, "username": "custom_user"} 

261 

262 response = await client.get("/custom-protected") 

263 assert response.status_code == 401