server.py 2.4 KB

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