diff options
author | Jason R. Coombs <jaraco@jaraco.com> | 2021-11-12 19:37:53 -0500 |
---|---|---|
committer | Jason R. Coombs <jaraco@jaraco.com> | 2021-11-12 19:37:53 -0500 |
commit | 26d7411464115c6170ed666e72e388d53d4b7fd2 (patch) | |
tree | 40a141e309c856e1af1df3dc2ec7699288d3d384 /setuptools/tests/server.py | |
parent | 342e02e7a4dbedf0e3a04c4d2d213b5340d56010 (diff) | |
parent | 77678abf97b4a8ee5e6e67b14cb21f543cd6bfd9 (diff) | |
download | python-setuptools-git-feature/local-schemes.tar.gz |
Merge branch 'main' into feature/local-schemesfeature/local-schemes
Diffstat (limited to 'setuptools/tests/server.py')
-rw-r--r-- | setuptools/tests/server.py | 90 |
1 files changed, 90 insertions, 0 deletions
diff --git a/setuptools/tests/server.py b/setuptools/tests/server.py new file mode 100644 index 00000000..6717c053 --- /dev/null +++ b/setuptools/tests/server.py @@ -0,0 +1,90 @@ +"""Basic http server for tests to simulate PyPI or custom indexes +""" + +import os +import time +import threading +import http.server +import urllib.parse +import urllib.request + + +class IndexServer(http.server.HTTPServer): + """Basic single-threaded http server simulating a package index + + You can use this server in unittest like this:: + s = IndexServer() + s.start() + index_url = s.base_url() + 'mytestindex' + # do some test requests to the index + # The index files should be located in setuptools/tests/indexes + s.stop() + """ + + def __init__( + self, server_address=('', 0), + RequestHandlerClass=http.server.SimpleHTTPRequestHandler): + http.server.HTTPServer.__init__( + self, server_address, RequestHandlerClass) + self._run = True + + def start(self): + self.thread = threading.Thread(target=self.serve_forever) + self.thread.start() + + def stop(self): + "Stop the server" + + # Let the server finish the last request and wait for a new one. + time.sleep(0.1) + + self.shutdown() + self.thread.join() + self.socket.close() + + def base_url(self): + port = self.server_port + return 'http://127.0.0.1:%s/setuptools/tests/indexes/' % port + + +class RequestRecorder(http.server.BaseHTTPRequestHandler): + def do_GET(self): + requests = vars(self.server).setdefault('requests', []) + requests.append(self) + self.send_response(200, 'OK') + + +class MockServer(http.server.HTTPServer, threading.Thread): + """ + A simple HTTP Server that records the requests made to it. + """ + + def __init__( + self, server_address=('', 0), + RequestHandlerClass=RequestRecorder): + http.server.HTTPServer.__init__( + self, server_address, RequestHandlerClass) + threading.Thread.__init__(self) + self.daemon = True + self.requests = [] + + def run(self): + self.serve_forever() + + @property + def netloc(self): + return 'localhost:%s' % self.server_port + + @property + def url(self): + return 'http://%s/' % self.netloc + + +def path_to_url(path, authority=None): + """ Convert a path to a file: URL. """ + path = os.path.normpath(os.path.abspath(path)) + base = 'file:' + if authority is not None: + base += '//' + authority + url = urllib.parse.urljoin(base, urllib.request.pathname2url(path)) + return url |