server.py 2.0 KB

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