Mercurial > repos > bcclaywell > argo_navis

comparison venv/lib/python2.7/site-packages/setuptools/tests/server.py @ 0:d67268158946 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
planemo upload commit a3f181f5f126803c654b3a66dd4e83a48f7e203b
author bcclaywell
date Mon, 12 Oct 2015 17:43:33 -0400
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:d67268158946
1 """Basic http server for tests to simulate PyPI or custom indexes
2 """
3
4 import time
5 import threading
6 from setuptools.compat import BaseHTTPRequestHandler
7 from setuptools.compat import HTTPServer, SimpleHTTPRequestHandler
8
9 class IndexServer(HTTPServer):
10 """Basic single-threaded http server simulating a package index
11
12 You can use this server in unittest like this::
13 s = IndexServer()
14 s.start()
15 index_url = s.base_url() + 'mytestindex'
16 # do some test requests to the index
17 # The index files should be located in setuptools/tests/indexes
18 s.stop()
19 """
20 def __init__(self, server_address=('', 0),
21 RequestHandlerClass=SimpleHTTPRequestHandler):
22 HTTPServer.__init__(self, server_address, RequestHandlerClass)
23 self._run = True
24
25 def start(self):
26 self.thread = threading.Thread(target=self.serve_forever)
27 self.thread.start()
28
29 def stop(self):
30 "Stop the server"
31
32 # Let the server finish the last request and wait for a new one.
33 time.sleep(0.1)
34
35 self.shutdown()
36 self.thread.join()
37 self.socket.close()
38
39 def base_url(self):
40 port = self.server_port
41 return 'http://127.0.0.1:%s/setuptools/tests/indexes/' % port
42
43 class RequestRecorder(BaseHTTPRequestHandler):
44 def do_GET(self):
45 requests = vars(self.server).setdefault('requests', [])
46 requests.append(self)
47 self.send_response(200, 'OK')
48
49 class MockServer(HTTPServer, threading.Thread):
50 """
51 A simple HTTP Server that records the requests made to it.
52 """
53 def __init__(self, server_address=('', 0),
54 RequestHandlerClass=RequestRecorder):
55 HTTPServer.__init__(self, server_address, RequestHandlerClass)
56 threading.Thread.__init__(self)
57 self.setDaemon(True)
58 self.requests = []
59
60 def run(self):
61 self.serve_forever()
62
63 @property
64 def url(self):
65 return 'http://localhost:%(server_port)s/' % vars(self)