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
|
import http.client
import json
import os
import socket
import ssl
import sys
from pathlib import Path
import ephemeral_port_reserve
import pytest
from xprocess import ProcessStarter
from werkzeug.utils import cached_property
run_path = str(Path(__file__).parent / "live_apps" / "run.py")
class UnixSocketHTTPConnection(http.client.HTTPConnection):
def connect(self):
self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
self.sock.connect(self.host)
class DevServerClient:
def __init__(self, kwargs):
host = kwargs.get("hostname", "127.0.0.1")
if not host.startswith("unix"):
port = kwargs.get("port")
if port is None:
kwargs["port"] = port = ephemeral_port_reserve.reserve(host)
scheme = "https" if "ssl_context" in kwargs else "http"
self.addr = f"{host}:{port}"
self.url = f"{scheme}://{self.addr}"
else:
self.addr = host[7:] # strip "unix://"
self.url = host
self.log = None
def tail_log(self, path):
self.log = open(path)
self.log.read()
def connect(self, **kwargs):
protocol = self.url.partition(":")[0]
if protocol == "https":
if "context" not in kwargs:
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
kwargs["context"] = context
return http.client.HTTPSConnection(self.addr, **kwargs)
if protocol == "unix":
return UnixSocketHTTPConnection(self.addr, **kwargs)
return http.client.HTTPConnection(self.addr, **kwargs)
def request(self, path="", **kwargs):
kwargs.setdefault("method", "GET")
kwargs.setdefault("url", path)
conn = self.connect()
conn.request(**kwargs)
with conn.getresponse() as response:
response.data = response.read()
conn.close()
if response.headers.get("Content-Type", "").startswith("application/json"):
response.json = json.loads(response.data)
else:
response.json = None
return response
def wait_for_log(self, start):
while True:
for line in self.log:
if line.startswith(start):
return
def wait_for_reload(self):
self.wait_for_log(" * Restarting with ")
@pytest.fixture()
def dev_server(xprocess, request, tmp_path):
"""A function that will start a dev server in an external process
and return a client for interacting with the server.
"""
def start_dev_server(name="standard", **kwargs):
client = DevServerClient(kwargs)
class Starter(ProcessStarter):
args = [sys.executable, run_path, name, json.dumps(kwargs)]
# Extend the existing env, otherwise Windows and CI fails.
# Modules will be imported from tmp_path for the reloader
# but any existing PYTHONPATH is preserved.
# Unbuffered output so the logs update immediately.
original_python_path = os.getenv("PYTHONPATH")
if original_python_path:
new_python_path = os.pathsep.join((original_python_path, str(tmp_path)))
else:
new_python_path = str(tmp_path)
env = {**os.environ, "PYTHONPATH": new_python_path, "PYTHONUNBUFFERED": "1"}
@cached_property
def pattern(self):
client.request("/ensure")
return "GET /ensure"
# Each test that uses the fixture will have a different log.
xp_name = f"dev_server-{request.node.name}"
_, log_path = xprocess.ensure(xp_name, Starter, restart=True)
client.tail_log(log_path)
@request.addfinalizer
def close():
xprocess.getinfo(xp_name).terminate()
client.log.close()
return client
return start_dev_server
@pytest.fixture()
def standard_app(dev_server):
"""Equivalent to ``dev_server("standard")``."""
return dev_server()
|
