Coverage for test\test_dependencies.py: 98%

1from nexios import get_application 

2import pytest 

3from nexios.http import Request, Response 

4from nexios.testing import Client 

5from nexios import Depend 

6from typing import Optional 

7from pydantic import BaseModel 

8 

9 

10@pytest.fixture 

11async def di_client(): 

12 app = get_application() 

13 async with Client(app) as client: 

14 yield client, app 

15 

16 

17# Test basic dependency injection 

18async def test_basic_di(di_client): 

19 client, app = di_client 

20 

21 async def get_message(): 

22 return "Hello from DI" 

23 

24 @app.get("/di/basic") 

25 async def basic_di_route( 

26 req: Request, res: Response, msg: str = Depend(get_message) 

27 ): 

28 return res.text(msg) 

29 

30 response = await client.get("/di/basic") 

31 assert response.status_code == 200 

32 assert response.text == "Hello from DI" 

33 

34 

35# Test dependency with request access 

36async def test_request_dependency(di_client): 

37 client, app = di_client 

38 

39 async def get_user_agent(req: Request): 

40 return req.headers.get("user-agent", "unknown") 

41 

42 @app.get("/di/user-agent") 

43 async def user_agent_route( 

44 req: Request, res: Response, ua: str = Depend(get_user_agent) 

45 ): 

46 return res.text(ua) 

47 

48 response = await client.get("/di/user-agent", headers={"User-Agent": "test-agent"}) 

49 assert response.status_code == 200 

50 assert response.text == "test-agent" 

51 

52 

53# Test nested dependencies 

54async def test_nested_dependencies(di_client): 

55 client, app = di_client 

56 

57 async def get_config(): 

58 return {"env": "test"} 

59 

60 async def get_service(config: dict = Depend(get_config)): 

61 return f"Service in {config['env']} environment" 

62 

63 @app.get("/di/nested") 

64 async def nested_route( 

65 req: Request, res: Response, service: str = Depend(get_service) 

66 ): 

67 return res.text(service) 

68 

69 response = await client.get("/di/nested") 

70 assert response.status_code == 200 

71 assert response.text == "Service in test environment" 

72 

73 

74# Test dependency with pydantic model 

75async def test_pydantic_dependency(di_client): 

76 client, app = di_client 

77 

78 class QueryParams(BaseModel): 

79 page: int = 1 

80 limit: int = 10 

81 

82 async def get_params(req: Request) -> QueryParams: 

83 return QueryParams(**req.query_params) 

84 

85 @app.get("/di/pydantic") 

86 async def pydantic_route( 

87 req: Request, res: Response, params: QueryParams = Depend(get_params) 

88 ): 

89 return res.json({"page": params.page, "limit": params.limit}) 

90 

91 response = await client.get("/di/pydantic?page=2&limit=20") 

92 assert response.status_code == 200 

93 assert response.json() == {"page": 2, "limit": 20} 

94 

95 

96# Test optional dependencies 

97async def test_optional_dependency(di_client): 

98 client, app = di_client 

99 

100 async def optional_header(req: Request): 

101 if "x-optional" in req.headers: 

102 return req.headers["x-optional"] 

103 return None 

104 

105 @app.get("/di/optional") 

106 async def optional_route( 

107 req: Request, res: Response, header: Optional[str] = Depend(optional_header) 

108 ): 

109 return res.text(header or "not-provided") 

110 

111 # Test with header 

112 response = await client.get("/di/optional", headers={"x-optional": "provided"}) 

113 assert response.status_code == 200 

114 assert response.text == "provided" 

115 

116 # Test without header 

117 response = await client.get("/di/optional") 

118 assert response.status_code == 200 

119 assert response.text == "not-provided" 

120 

121 

122# Test dependency error handling 

123async def test_dependency_error(di_client): 

124 client, app = di_client 

125 

126 async def failing_dependency(): 

127 raise ValueError("Dependency failed") 

128 

129 @app.get("/di/error") 

130 async def error_route( 

131 req: Request, res: Response, dep: str = Depend(failing_dependency) 

132 ): 

133 return res.text("should-not-reach-here") 

134 

135 response = await client.get("/di/error") 

136 assert response.status_code == 500 

137 assert "Dependency failed" in response.text 

138 

139 

140# Test sync dependencies 

141async def test_sync_dependency(di_client): 

142 client, app = di_client 

143 

144 def sync_dependency(): 

145 return "sync-result" 

146 

147 @app.get("/di/sync") 

148 async def sync_route( 

149 req: Request, res: Response, result: str = Depend(sync_dependency) 

150 ): 

151 return res.text(result) 

152 

153 response = await client.get("/di/sync") 

154 assert response.status_code == 200 

155 assert response.text == "sync-result" 

156 

157 

158# Test dependency with route parameters 

159async def test_route_param_dependency(di_client): 

160 client, app = di_client 

161 

162 async def get_id_param(req: Request): 

163 return req.path_params["id"] 

164 

165 @app.get("/di/param/{id}") 

166 async def param_route(req: Request, res: Response, id: str = Depend(get_id_param)): 

167 return res.text(id) 

168 

169 response = await client.get("/di/param/123") 

170 assert response.status_code == 200 

171 assert response.text == "123"