live_server_helper.py 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112
  1. import sys
  2. class LiveServer(object):
  3. """The liveserver fixture
  4. This is the object which is returned to the actual user when they
  5. request the ``live_server`` fixture. The fixture handles creation
  6. and stopping however.
  7. """
  8. def __init__(self, addr):
  9. import django
  10. from django.db import connections
  11. from django.test.testcases import LiveServerThread
  12. from django.test.utils import modify_settings
  13. connections_override = {}
  14. for conn in connections.all():
  15. # If using in-memory sqlite databases, pass the connections to
  16. # the server thread.
  17. if (conn.settings_dict['ENGINE'] == 'django.db.backends.sqlite3' and
  18. conn.settings_dict['NAME'] == ':memory:'):
  19. # Explicitly enable thread-shareability for this connection
  20. conn.allow_thread_sharing = True
  21. connections_override[conn.alias] = conn
  22. liveserver_kwargs = {'connections_override': connections_override}
  23. from django.conf import settings
  24. if 'django.contrib.staticfiles' in settings.INSTALLED_APPS:
  25. from django.contrib.staticfiles.handlers import StaticFilesHandler
  26. liveserver_kwargs['static_handler'] = StaticFilesHandler
  27. else:
  28. from django.test.testcases import _StaticFilesHandler
  29. liveserver_kwargs['static_handler'] = _StaticFilesHandler
  30. if django.VERSION < (1, 11):
  31. host, possible_ports = parse_addr(addr)
  32. self.thread = LiveServerThread(host, possible_ports,
  33. **liveserver_kwargs)
  34. else:
  35. host = addr
  36. self.thread = LiveServerThread(host, **liveserver_kwargs)
  37. self._live_server_modified_settings = modify_settings(
  38. ALLOWED_HOSTS={'append': host},
  39. )
  40. self._live_server_modified_settings.enable()
  41. self.thread.daemon = True
  42. self.thread.start()
  43. self.thread.is_ready.wait()
  44. if self.thread.error:
  45. raise self.thread.error
  46. def stop(self):
  47. """Stop the server"""
  48. # .terminate() was added in Django 1.7
  49. terminate = getattr(self.thread, 'terminate', lambda: None)
  50. terminate()
  51. self.thread.join()
  52. self._live_server_modified_settings.disable()
  53. @property
  54. def url(self):
  55. return 'http://%s:%s' % (self.thread.host, self.thread.port)
  56. if sys.version_info < (3, 0):
  57. def __unicode__(self):
  58. return self.url
  59. def __add__(self, other):
  60. return unicode(self) + other # noqa: pyflakes on python3
  61. else:
  62. def __str__(self):
  63. return self.url
  64. def __add__(self, other):
  65. return str(self) + other
  66. def __repr__(self):
  67. return '<LiveServer listening at %s>' % self.url
  68. def parse_addr(specified_address):
  69. """Parse the --liveserver argument into a host/IP address and port range"""
  70. # This code is based on
  71. # django.test.testcases.LiveServerTestCase.setUpClass
  72. # The specified ports may be of the form '8000-8010,8080,9200-9300'
  73. # i.e. a comma-separated list of ports or ranges of ports, so we break
  74. # it down into a detailed list of all possible ports.
  75. possible_ports = []
  76. try:
  77. host, port_ranges = specified_address.split(':')
  78. for port_range in port_ranges.split(','):
  79. # A port range can be of either form: '8000' or '8000-8010'.
  80. extremes = list(map(int, port_range.split('-')))
  81. assert len(extremes) in (1, 2)
  82. if len(extremes) == 1:
  83. # Port range of the form '8000'
  84. possible_ports.append(extremes[0])
  85. else:
  86. # Port range of the form '8000-8010'
  87. for port in range(extremes[0], extremes[1] + 1):
  88. possible_ports.append(port)
  89. except Exception:
  90. raise Exception(
  91. 'Invalid address ("%s") for live server.' % specified_address)
  92. return host, possible_ports