Coverage for test\test_middlewares.py: 100%

1from nexios import NexiosApp 

2from nexios.http import Request, Response 

3from nexios import get_application 

4from nexios.testing import Client 

5import pytest 

6 

7app: NexiosApp = get_application() 

8 

9 

10@pytest.fixture(autouse=True) 

11async def async_client(): 

12 async with Client(app, log_requests=True) as c: 

13 yield c 

14 

15 

16async def test_middleware_modifies_response(async_client: Client): 

17 

18 app.router.routes.clear() 

19 app.http_middlewares.clear() 

20 

21 async def header_middleware(request: Request, response: Response, call_next): 

22 await call_next() 

23 response.set_header("X-Middleware-Added", "True") 

24 

25 app.add_middleware(header_middleware) 

26 

27 @app.route("/header-test") 

28 async def test_route(request: Request, response: Response): 

29 return response.text("OK") 

30 

31 # Execute test 

32 response = await async_client.get("/header-test") 

33 assert response.status_code == 200 

34 assert "X-Middleware-Added" in response.headers 

35 assert response.headers["X-Middleware-Added"] == "True" 

36 

37 

38async def test_middleware_modifies_request(async_client: Client): 

39 

40 app.router.routes.clear() 

41 app.http_middlewares.clear() 

42 

43 async def modify_request(request: Request, response: Response, call_next): 

44 request.state.text = "modified" 

45 return await call_next() 

46 

47 app.add_middleware(modify_request) 

48 

49 @app.route("/request-test") 

50 async def request_test(request: Request, response: Response): 

51 return response.text(request.state.text) 

52 

53 response = await async_client.get("/request-test") 

54 assert response.text == "modified" 

55 

56 

57async def test_middleware_order(async_client: Client): 

58 

59 app.router.routes.clear() 

60 app.http_middlewares.clear() 

61 

62 async def first_middleware(equest: Request, response: Response, call_next): 

63 await call_next() 

64 response.set_header("X-Order-1", "First") 

65 return response 

66 

67 async def second_middleware(equest: Request, response: Response, call_next): 

68 await call_next() 

69 response.set_header("X-Order-2", "Second") 

70 return response 

71 

72 app.add_middleware(first_middleware) 

73 app.add_middleware(second_middleware) 

74 

75 @app.route("/order-test") 

76 async def order_test(request: Request, response: Response): 

77 return response.text("OK") 

78 

79 response = await async_client.get("/order-test") 

80 assert response.headers["X-Order-1"] == "First" 

81 assert response.headers["X-Order-2"] == "Second" 

82 

83 

84async def test_middleware_short_circuit(async_client: Client): 

85 

86 app.router.routes.clear() 

87 app.http_middlewares.clear() 

88 

89 async def blocking_middleware(request: Request, response: Response, call_next): 

90 if request.headers.get("Authorization") != "Valid": 

91 return response.text("Unauthorized", status_code=403) 

92 return await call_next() 

93 

94 app.add_middleware(blocking_middleware) 

95 

96 @app.route("/protected") 

97 async def protected_route(request: Request, response: Response): 

98 return response.text("Secret") 

99 

100 # Test blocked request 

101 response = await async_client.get("/protected") 

102 assert response.status_code == 403 

103 assert response.text == "Unauthorized" 

104 

105 # Test allowed request 

106 response = await async_client.get("/protected", headers={"Authorization": "Valid"}) 

107 assert response.status_code == 200 

108 assert response.text == "Secret" 

109 

110 

111async def test_middleware_exception_handling(async_client: Client): 

112 

113 app.router.routes.clear() 

114 app.http_middlewares.clear() 

115 

116 async def exception_middleware(request: Request, response: Response, call_next): 

117 try: 

118 return await call_next() 

119 except Exception: 

120 return response.text("Handled Error", status_code=500) 

121 

122 app.add_middleware(exception_middleware) 

123 

124 @app.route("/error-route") 

125 async def error_route(request: Request, response: Response): 

126 raise ValueError("Something went wrong") 

127 

128 response = await async_client.get("/error-route") 

129 assert response.status_code == 500 

130 assert response.text == "Handled Error"