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

1from __future__ import annotations 

2 

3import typing as t 

4from contextlib import contextmanager 

5from contextlib import ExitStack 

6from copy import copy 

7from types import TracebackType 

8from urllib.parse import urlsplit 

9 

10import werkzeug.test 

11from click.testing import CliRunner 

12from werkzeug.test import Client 

13from werkzeug.wrappers import Request as BaseRequest 

14 

15from .cli import ScriptInfo 

16from .sessions import SessionMixin 

17 

18if t.TYPE_CHECKING: # pragma: no cover 

19 from werkzeug.test import TestResponse 

20 

21 from .app import Flask 

22 

23 

24class EnvironBuilder(werkzeug.test.EnvironBuilder): 

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

26 application. 

27 

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

29 :param path: URL path being requested. 

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

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

32 :data:`PREFERRED_URL_SCHEME`, ``subdomain``, 

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

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

35 :param url_scheme: Scheme to use instead of 

36 :data:`PREFERRED_URL_SCHEME`. 

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

38 ``data``. Also defaults ``content_type`` to 

39 ``application/json``. 

40 :param args: other positional arguments passed to 

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

42 :param kwargs: other keyword arguments passed to 

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

44 """ 

45 

46 def __init__( 

47 self, 

48 app: Flask, 

49 path: str = "/", 

50 base_url: str | None = None, 

51 subdomain: str | None = None, 

52 url_scheme: str | None = None, 

53 *args: t.Any, 

54 **kwargs: t.Any, 

55 ) -> None: 

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

57 base_url is not None 

58 ) != bool( 

59 subdomain or url_scheme 

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

61 

62 if base_url is None: 

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

64 app_root = app.config["APPLICATION_ROOT"] 

65 

66 if subdomain: 

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

68 

69 if url_scheme is None: 

70 url_scheme = app.config["PREFERRED_URL_SCHEME"] 

71 

72 url = urlsplit(path) 

73 base_url = ( 

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

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

76 ) 

77 path = url.path 

78 

79 if url.query: 

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

81 path += sep + url.query 

82 

83 self.app = app 

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

85 

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

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

88 

89 The serialization will be configured according to the config associated 

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

91 """ 

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

93 

94 

95class FlaskClient(Client): 

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

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

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

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

100 

101 .. versionchanged:: 0.12 

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

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

104 `client.environ_base`. 

105 

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

107 """ 

108 

109 application: Flask 

110 

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

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

113 self.preserve_context = False 

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

115 self._context_stack = ExitStack() 

116 self.environ_base = { 

117 "REMOTE_ADDR": "127.0.0.1", 

118 "HTTP_USER_AGENT": f"werkzeug/{werkzeug.__version__}", 

119 } 

120 

121 @contextmanager 

122 def session_transaction( 

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

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

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

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

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

128 stored back. 

129 

130 :: 

131 

132 with client.session_transaction() as session: 

133 session['value'] = 42 

134 

135 Internally this is implemented by going through a temporary test 

136 request context and since session handling could depend on 

137 request variables this function accepts the same arguments as 

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

139 passed through. 

140 """ 

141 if self._cookies is None: 

142 raise TypeError( 

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

144 ) 

145 

146 app = self.application 

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

148 self._add_cookies_to_wsgi(ctx.request.environ) 

149 

150 with ctx: 

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

152 

153 if sess is None: 

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

155 

156 yield sess 

157 resp = app.response_class() 

158 

159 if app.session_interface.is_null_session(sess): 

160 return 

161 

162 with ctx: 

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

164 

165 self._update_cookies_from_response( 

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

167 ctx.request.path, 

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

169 ) 

170 

171 def _copy_environ(self, other): 

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

173 

174 if self.preserve_context: 

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

176 

177 return out 

178 

179 def _request_from_builder_args(self, args, kwargs): 

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

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

182 

183 try: 

184 return builder.get_request() 

185 finally: 

186 builder.close() 

187 

188 def open( 

189 self, 

190 *args: t.Any, 

191 buffered: bool = False, 

192 follow_redirects: bool = False, 

193 **kwargs: t.Any, 

194 ) -> TestResponse: 

195 if args and isinstance( 

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

197 ): 

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

199 builder = copy(args[0]) 

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

201 request = builder.get_request() 

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

203 request = EnvironBuilder.from_environ( 

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

205 ).get_request() 

206 else: 

207 # isinstance(args[0], BaseRequest) 

208 request = copy(args[0]) 

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

210 else: 

211 # request is None 

212 request = self._request_from_builder_args(args, kwargs) 

213 

214 # Pop any previously preserved contexts. This prevents contexts 

215 # from being preserved across redirects or multiple requests 

216 # within a single block. 

217 self._context_stack.close() 

218 

219 response = super().open( 

220 request, 

221 buffered=buffered, 

222 follow_redirects=follow_redirects, 

223 ) 

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

225 

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

227 while self._new_contexts: 

228 cm = self._new_contexts.pop() 

229 self._context_stack.enter_context(cm) 

230 

231 return response 

232 

233 def __enter__(self) -> FlaskClient: 

234 if self.preserve_context: 

235 raise RuntimeError("Cannot nest client invocations") 

236 self.preserve_context = True 

237 return self 

238 

239 def __exit__( 

240 self, 

241 exc_type: type | None, 

242 exc_value: BaseException | None, 

243 tb: TracebackType | None, 

244 ) -> None: 

245 self.preserve_context = False 

246 self._context_stack.close() 

247 

248 

249class FlaskCliRunner(CliRunner): 

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

251 CLI commands. Typically created using 

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

253 """ 

254 

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

256 self.app = app 

257 super().__init__(**kwargs) 

258 

259 def invoke( # type: ignore 

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

261 ) -> t.Any: 

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

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

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

265 

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

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

268 app being tested. 

269 

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

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

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

273 

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

275 """ 

276 if cli is None: 

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

278 

279 if "obj" not in kwargs: 

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

281 

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