1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
|
# -*- coding: utf-8 -*-
"""
tests.conftest
~~~~~~~~~~~~~~
:copyright: 2007 Pallets
:license: BSD-3-Clause
"""
from __future__ import print_function
import logging
import os
import platform
import signal
import subprocess
import sys
import textwrap
import time
from itertools import count
import pytest
from werkzeug import serving
from werkzeug._compat import to_bytes
from werkzeug.urls import url_quote
from werkzeug.utils import cached_property
try:
__import__("pytest_xprocess")
except ImportError:
@pytest.fixture(scope="session")
def xprocess():
pytest.skip("pytest-xprocess not installed.")
port_generator = count(13220)
def _patch_reloader_loop():
def f(x):
print("reloader loop finished")
# Need to flush for some reason even though xprocess opens the
# subprocess' stdout in unbuffered mode.
# flush=True makes the test fail on py2, so flush manually
sys.stdout.flush()
return time.sleep(x)
import werkzeug._reloader
werkzeug._reloader.ReloaderLoop._sleep = staticmethod(f)
pid_logger = logging.getLogger("get_pid_middleware")
pid_logger.setLevel(logging.INFO)
pid_handler = logging.StreamHandler(sys.stdout)
pid_logger.addHandler(pid_handler)
def _get_pid_middleware(f):
def inner(environ, start_response):
if environ["PATH_INFO"] == "/_getpid":
start_response("200 OK", [("Content-Type", "text/plain")])
pid_logger.info("pid=%s", os.getpid())
return [to_bytes(str(os.getpid()))]
return f(environ, start_response)
return inner
def _dev_server():
_patch_reloader_loop()
sys.path.insert(0, sys.argv[1])
import testsuite_app
app = _get_pid_middleware(testsuite_app.app)
serving.run_simple(application=app, **testsuite_app.kwargs)
class _ServerInfo(object):
xprocess = None
addr = None
url = None
port = None
last_pid = None
def __init__(self, xprocess, addr, url, port):
self.xprocess = xprocess
self.addr = addr
self.url = url
self.port = port
@cached_property
def logfile(self):
return self.xprocess.getinfo("dev_server").logpath.open()
def request_pid(self):
if self.url.startswith("http+unix://"):
from requests_unixsocket import get as rget
else:
from requests import get as rget
for i in range(10):
time.sleep(0.1 * i)
try:
response = rget(self.url + "/_getpid", verify=False)
self.last_pid = int(response.text)
return self.last_pid
except Exception as e: # urllib also raises socketerrors
print(self.url)
print(e)
def wait_for_reloader(self):
old_pid = self.last_pid
for i in range(20):
time.sleep(0.1 * i)
new_pid = self.request_pid()
if not new_pid:
raise RuntimeError("Server is down.")
if new_pid != old_pid:
return
raise RuntimeError("Server did not reload.")
def wait_for_reloader_loop(self):
for i in range(20):
time.sleep(0.1 * i)
line = self.logfile.readline()
if "reloader loop finished" in line:
return
@pytest.fixture
def dev_server(tmpdir, xprocess, request, monkeypatch):
"""Run werkzeug.serving.run_simple in its own process.
:param application: String for the module that will be created. The module
must have a global ``app`` object, a ``kwargs`` dict is also available
whose values will be passed to ``run_simple``.
"""
def run_dev_server(application):
app_pkg = tmpdir.mkdir("testsuite_app")
appfile = app_pkg.join("__init__.py")
port = next(port_generator)
appfile.write(
"\n\n".join(
(
"kwargs = {{'hostname': 'localhost', 'port': {port:d}}}".format(
port=port
),
textwrap.dedent(application),
)
)
)
monkeypatch.delitem(sys.modules, "testsuite_app", raising=False)
monkeypatch.syspath_prepend(str(tmpdir))
import testsuite_app
hostname = testsuite_app.kwargs["hostname"]
port = testsuite_app.kwargs["port"]
addr = "{}:{}".format(hostname, port)
if hostname.startswith("unix://"):
addr = hostname.split("unix://", 1)[1]
requests_url = "http+unix://" + url_quote(addr, safe="")
elif testsuite_app.kwargs.get("ssl_context", None):
requests_url = "https://localhost:{0}".format(port)
else:
requests_url = "http://localhost:{0}".format(port)
info = _ServerInfo(xprocess, addr, requests_url, port)
from xprocess import ProcessStarter
class Starter(ProcessStarter):
args = [sys.executable, __file__, str(tmpdir)]
@property
def pattern(self):
return "pid=%s" % info.request_pid()
xprocess.ensure("dev_server", Starter, restart=True)
@request.addfinalizer
def teardown():
# Killing the process group that runs the server, not just the
# parent process attached. xprocess is confused about Werkzeug's
# reloader and won't help here.
pid = info.request_pid()
if not pid:
return
if platform.system() == "Windows":
subprocess.call(["taskkill", "/F", "/T", "/PID", str(pid)])
else:
os.killpg(os.getpgid(pid), signal.SIGTERM)
return info
return run_dev_server
if __name__ == "__main__":
_dev_server()
|
