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
|
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See LICENSE in the project root
# for license information.
"""Test helpers for networking.
"""
import os
import re
import requests
import socket
import threading
import time
from debugpy.common import log, util
from tests.patterns import some
used_ports = set()
def get_test_server_port(max_retries=10):
"""Returns a server port number that can be safely used for listening without
clashing with another test worker process, when running with pytest-xdist.
If multiple test workers invoke this function with the same min value, each of
them will receive a different number that is not lower than start (but may be
higher). If the resulting value is >=stop, it is a fatal error.
Note that if multiple test workers invoke this function with different ranges
that overlap, conflicts are possible!
Args:
max_retries: Number of times to retry finding an available port
"""
try:
worker_id = util.force_ascii(os.environ["PYTEST_XDIST_WORKER"])
except KeyError:
n = 0
else:
assert worker_id == some.bytes.matching(
rb"gw(\d+)"
), "Unrecognized PYTEST_XDIST_WORKER format"
n = int(worker_id[2:])
# Try multiple times to find an available port, with retry logic
for attempt in range(max_retries):
port = 5678 + (n * 300) + attempt
while port in used_ports:
port += 1
# Verify the port is actually available by trying to bind to it
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
sock.bind(("127.0.0.1", port))
sock.close()
used_ports.add(port)
log.info("Allocated port {0} for worker {1}", port, n)
return port
except OSError as e:
log.warning("Port {0} unavailable (attempt {1}/{2}): {3}", port, attempt + 1, max_retries, e)
sock.close()
time.sleep(0.1 * (attempt + 1)) # Exponential backoff
# Fall back to original behavior if all retries fail
port = 5678 + (n * 300)
while port in used_ports:
port += 1
used_ports.add(port)
log.warning("Using fallback port {0} after {1} retries", port, max_retries)
return port
def find_http_url(text):
match = re.search(r"https?://[-.0-9A-Za-z]+(:\d+)/?", text)
return match.group() if match else None
def wait_until_port_is_listening(port, interval=1, max_attempts=1000):
"""Blocks until the specified TCP port on localhost is listening, and can be
connected to.
Tries to connect to the port periodically, and repeats until connection succeeds.
Connection is immediately closed before returning.
"""
for i in range(1, max_attempts + 1):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
log.info("Probing localhost:{0} (attempt {1})...", port, i)
sock.connect(("localhost", port))
except socket.error as exc:
# The first attempt will almost always fail, because the port isn't
# open yet. But if it keeps failing after that, we want to know why.
if i > 1:
log.warning("Failed to connect to localhost:{0}:\n{1}", port, exc)
time.sleep(interval)
else:
log.info("localhost:{0} is listening - server is up!", port)
return
finally:
sock.close()
class WebRequest(object):
"""An async wrapper around requests."""
@staticmethod
def get(*args, **kwargs):
return WebRequest("get", *args, **kwargs)
@staticmethod
def post(*args, **kwargs):
return WebRequest("post", *args, **kwargs)
def __init__(self, method, url, *args, **kwargs):
"""Invokes requests.method(url, *args, **kwargs) on a background thread,
and immediately returns.
If method() raises an exception, it is logged, unless log_errors=False.
"""
self.method = method
self.url = url
self.log_errors = kwargs.pop("log_errors", True)
self.request = None
"""The underlying requests.Request object.
Not set until wait_for_response() returns.
"""
self.exception = None
"""Exception that occurred while performing the request, if any.
Not set until wait_for_response() returns.
"""
log.info("{0}", self)
func = getattr(requests, method)
self._worker_thread = threading.Thread(
target=lambda: self._worker(func, *args, **kwargs),
name=f"WebRequest({self})",
)
self._worker_thread.daemon = True
self._worker_thread.start()
def __str__(self):
return f"HTTP {self.method.upper()} {self.url}"
def _worker(self, func, *args, **kwargs):
try:
self.request = func(self.url, *args, **kwargs)
except Exception as exc:
if self.log_errors:
log.swallow_exception("{0} failed:", self)
self.exception = exc
else:
log.info(
"{0} --> {1} {2}", self, self.request.status_code, self.request.reason
)
def wait_for_response(self, timeout=None):
"""Blocks until the request completes, and returns self.request."""
if self._worker_thread.is_alive():
log.info("Waiting for response to {0} ...", self)
self._worker_thread.join(timeout)
if self.exception is not None:
raise self.exception
return self.request
def response_text(self):
"""Blocks until the request completes, and returns the response body."""
return self.wait_for_response().text
class WebServer(object):
"""Interacts with a web server listening on localhost on the specified port."""
def __init__(self, port):
self.port = port
self.url = f"http://localhost:{port}"
def __enter__(self):
"""Blocks until the server starts listening on self.port."""
log.info("Web server expected on {0}", self.url)
wait_until_port_is_listening(self.port, interval=3)
return self
def __exit__(self, exc_type, exc_value, exc_tb):
"""Sends an HTTP /exit GET request to the server.
The server is expected to terminate its process while handling that request.
"""
self.get("/exit", log_errors=False)
def get(self, path, *args, **kwargs):
return WebRequest.get(self.url + path, *args, **kwargs)
def post(self, path, *args, **kwargs):
return WebRequest.post(self.url + path, *args, **kwargs)
|
