Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/flask/testing.py: 61%

1from __future__ import annotations 

2 

3import importlib.metadata 

4import typing as t 

5from contextlib import contextmanager 

6from contextlib import ExitStack 

7from copy import copy 

8from types import TracebackType 

9from urllib.parse import urlsplit 

10 

11import werkzeug.test 

12from click.testing import CliRunner 

13from werkzeug.test import Client 

14from werkzeug.wrappers import Request as BaseRequest 

15 

16from .cli import ScriptInfo 

17from .sessions import SessionMixin 

18 

19if t.TYPE_CHECKING: # pragma: no cover 

20 from werkzeug.test import TestResponse 

21 

22 from .app import Flask 

23 

24 

25class EnvironBuilder(werkzeug.test.EnvironBuilder): 

26 """An :class:`~werkzeug.test.EnvironBuilder`, that takes defaults from the 

27 application. 

28 

29 :param app: The Flask application to configure the environment from. 

30 :param path: URL path being requested. 

31 :param base_url: Base URL where the app is being served, which 

32 ``path`` is relative to. If not given, built from 

33 :data:`PREFERRED_URL_SCHEME`, ``subdomain``, 

34 :data:`SERVER_NAME`, and :data:`APPLICATION_ROOT`. 

35 :param subdomain: Subdomain name to append to :data:`SERVER_NAME`. 

36 :param url_scheme: Scheme to use instead of 

37 :data:`PREFERRED_URL_SCHEME`. 

38 :param json: If given, this is serialized as JSON and passed as 

39 ``data``. Also defaults ``content_type`` to 

40 ``application/json``. 

41 :param args: other positional arguments passed to 

42 :class:`~werkzeug.test.EnvironBuilder`. 

43 :param kwargs: other keyword arguments passed to 

44 :class:`~werkzeug.test.EnvironBuilder`. 

45 """ 

46 

47 def __init__( 

48 self, 

49 app: Flask, 

50 path: str = "/", 

51 base_url: str | None = None, 

52 subdomain: str | None = None, 

53 url_scheme: str | None = None, 

54 *args: t.Any, 

55 **kwargs: t.Any, 

56 ) -> None: 

57 assert not (base_url or subdomain or url_scheme) or ( 

58 base_url is not None 

59 ) != bool( 

60 subdomain or url_scheme 

61 ), 'Cannot pass "subdomain" or "url_scheme" with "base_url".' 

62 

63 if base_url is None: 

64 http_host = app.config.get("SERVER_NAME") or "localhost" 

65 app_root = app.config["APPLICATION_ROOT"] 

66 

67 if subdomain: 

68 http_host = f"{subdomain}.{http_host}" 

69 

70 if url_scheme is None: 

71 url_scheme = app.config["PREFERRED_URL_SCHEME"] 

72 

73 url = urlsplit(path) 

74 base_url = ( 

75 f"{url.scheme or url_scheme}://{url.netloc or http_host}" 

76 f"/{app_root.lstrip('/')}" 

77 ) 

78 path = url.path 

79 

80 if url.query: 

81 sep = b"?" if isinstance(url.query, bytes) else "?" 

82 path += sep + url.query 

83 

84 self.app = app 

85 super().__init__(path, base_url, *args, **kwargs) 

86 

87 def json_dumps(self, obj: t.Any, **kwargs: t.Any) -> str: # type: ignore 

88 """Serialize ``obj`` to a JSON-formatted string. 

89 

90 The serialization will be configured according to the config associated 

91 with this EnvironBuilder's ``app``. 

92 """ 

93 return self.app.json.dumps(obj, **kwargs) 

94 

95 

96_werkzeug_version = "" 

97 

98 

99def _get_werkzeug_version() -> str: 

100 global _werkzeug_version 

101 

102 if not _werkzeug_version: 

103 _werkzeug_version = importlib.metadata.version("werkzeug") 

104 

105 return _werkzeug_version 

106 

107 

108class FlaskClient(Client): 

109 """Works like a regular Werkzeug test client but has knowledge about 

110 Flask's contexts to defer the cleanup of the request context until 

111 the end of a ``with`` block. For general information about how to 

112 use this class refer to :class:`werkzeug.test.Client`. 

113 

114 .. versionchanged:: 0.12 

115 `app.test_client()` includes preset default environment, which can be 

116 set after instantiation of the `app.test_client()` object in 

117 `client.environ_base`. 

118 

119 Basic usage is outlined in the :doc:`/testing` chapter. 

120 """ 

121 

122 application: Flask 

123 

124 def __init__(self, *args: t.Any, **kwargs: t.Any) -> None: 

125 super().__init__(*args, **kwargs) 

126 self.preserve_context = False 

127 self._new_contexts: list[t.ContextManager[t.Any]] = [] 

128 self._context_stack = ExitStack() 

129 self.environ_base = { 

130 "REMOTE_ADDR": "127.0.0.1", 

131 "HTTP_USER_AGENT": f"Werkzeug/{_get_werkzeug_version()}", 

132 } 

133 

134 @contextmanager 

135 def session_transaction( 

136 self, *args: t.Any, **kwargs: t.Any 

137 ) -> t.Generator[SessionMixin, None, None]: 

138 """When used in combination with a ``with`` statement this opens a 

139 session transaction. This can be used to modify the session that 

140 the test client uses. Once the ``with`` block is left the session is 

141 stored back. 

142 

143 :: 

144 

145 with client.session_transaction() as session: 

146 session['value'] = 42 

147 

148 Internally this is implemented by going through a temporary test 

149 request context and since session handling could depend on 

150 request variables this function accepts the same arguments as 

151 :meth:`~flask.Flask.test_request_context` which are directly 

152 passed through. 

153 """ 

154 if self._cookies is None: 

155 raise TypeError( 

156 "Cookies are disabled. Create a client with 'use_cookies=True'." 

157 ) 

158 

159 app = self.application 

160 ctx = app.test_request_context(*args, **kwargs) 

161 self._add_cookies_to_wsgi(ctx.request.environ) 

162 

163 with ctx: 

164 sess = app.session_interface.open_session(app, ctx.request) 

165 

166 if sess is None: 

167 raise RuntimeError("Session backend did not open a session.") 

168 

169 yield sess 

170 resp = app.response_class() 

171 

172 if app.session_interface.is_null_session(sess): 

173 return 

174 

175 with ctx: 

176 app.session_interface.save_session(app, sess, resp) 

177 

178 self._update_cookies_from_response( 

179 ctx.request.host.partition(":")[0], 

180 ctx.request.path, 

181 resp.headers.getlist("Set-Cookie"), 

182 ) 

183 

184 def _copy_environ(self, other): 

185 out = {**self.environ_base, **other} 

186 

187 if self.preserve_context: 

188 out["werkzeug.debug.preserve_context"] = self._new_contexts.append 

189 

190 return out 

191 

192 def _request_from_builder_args(self, args, kwargs): 

193 kwargs["environ_base"] = self._copy_environ(kwargs.get("environ_base", {})) 

194 builder = EnvironBuilder(self.application, *args, **kwargs) 

195 

196 try: 

197 return builder.get_request() 

198 finally: 

199 builder.close() 

200 

201 def open( 

202 self, 

203 *args: t.Any, 

204 buffered: bool = False, 

205 follow_redirects: bool = False, 

206 **kwargs: t.Any, 

207 ) -> TestResponse: 

208 if args and isinstance( 

209 args[0], (werkzeug.test.EnvironBuilder, dict, BaseRequest) 

210 ): 

211 if isinstance(args[0], werkzeug.test.EnvironBuilder): 

212 builder = copy(args[0]) 

213 builder.environ_base = self._copy_environ(builder.environ_base or {}) 

214 request = builder.get_request() 

215 elif isinstance(args[0], dict): 

216 request = EnvironBuilder.from_environ( 

217 args[0], app=self.application, environ_base=self._copy_environ({}) 

218 ).get_request() 

219 else: 

220 # isinstance(args[0], BaseRequest) 

221 request = copy(args[0]) 

222 request.environ = self._copy_environ(request.environ) 

223 else: 

224 # request is None 

225 request = self._request_from_builder_args(args, kwargs) 

226 

227 # Pop any previously preserved contexts. This prevents contexts 

228 # from being preserved across redirects or multiple requests 

229 # within a single block. 

230 self._context_stack.close() 

231 

232 response = super().open( 

233 request, 

234 buffered=buffered, 

235 follow_redirects=follow_redirects, 

236 ) 

237 response.json_module = self.application.json # type: ignore[assignment] 

238 

239 # Re-push contexts that were preserved during the request. 

240 while self._new_contexts: 

241 cm = self._new_contexts.pop() 

242 self._context_stack.enter_context(cm) 

243 

244 return response 

245 

246 def __enter__(self) -> FlaskClient: 

247 if self.preserve_context: 

248 raise RuntimeError("Cannot nest client invocations") 

249 self.preserve_context = True 

250 return self 

251 

252 def __exit__( 

253 self, 

254 exc_type: type | None, 

255 exc_value: BaseException | None, 

256 tb: TracebackType | None, 

257 ) -> None: 

258 self.preserve_context = False 

259 self._context_stack.close() 

260 

261 

262class FlaskCliRunner(CliRunner): 

263 """A :class:`~click.testing.CliRunner` for testing a Flask app's 

264 CLI commands. Typically created using 

265 :meth:`~flask.Flask.test_cli_runner`. See :ref:`testing-cli`. 

266 """ 

267 

268 def __init__(self, app: Flask, **kwargs: t.Any) -> None: 

269 self.app = app 

270 super().__init__(**kwargs) 

271 

272 def invoke( # type: ignore 

273 self, cli: t.Any = None, args: t.Any = None, **kwargs: t.Any 

274 ) -> t.Any: 

275 """Invokes a CLI command in an isolated environment. See 

276 :meth:`CliRunner.invoke <click.testing.CliRunner.invoke>` for 

277 full method documentation. See :ref:`testing-cli` for examples. 

278 

279 If the ``obj`` argument is not given, passes an instance of 

280 :class:`~flask.cli.ScriptInfo` that knows how to load the Flask 

281 app being tested. 

282 

283 :param cli: Command object to invoke. Default is the app's 

284 :attr:`~flask.app.Flask.cli` group. 

285 :param args: List of strings to invoke the command with. 

286 

287 :return: a :class:`~click.testing.Result` object. 

288 """ 

289 if cli is None: 

290 cli = self.app.cli # type: ignore 

291 

292 if "obj" not in kwargs: 

293 kwargs["obj"] = ScriptInfo(create_app=lambda: self.app) 

294 

295 return super().invoke(cli, args, **kwargs)