Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def start(self, port, *args, **config):
config.update(self.config)
config['server.socket_port'] = port
portend.free('localhost', port)
server = self[port] = subprocess.Popen((sys.executable, '-m', self.module, '-c', json.dumps(config)) + args)
portend.occupied('localhost', port)
server.started = time.time()
assert not server.poll()
return clients.Resource('http://localhost:{}'.format(port))
def wait(self):
"""Wait until the HTTP server is ready to receive requests."""
while not getattr(self.httpserver, 'ready', False):
if self.interrupt:
raise self.interrupt
time.sleep(.1)
# Wait for port to be occupied
if not os.environ.get('LISTEN_PID', None):
# Wait for port to be occupied if not running via socket-activation
# (for socket-activation the port will be managed by systemd )
if isinstance(self.bind_addr, tuple):
with _safe_wait(*self.bound_addr):
portend.occupied(*self.bound_addr, timeout=Timeouts.occupied)
if self.interrupt:
raise self.interrupt
time.sleep(.1)
# bypass check when LISTEN_PID is set
if os.environ.get('LISTEN_PID', None):
return
# bypass check when running via socket-activation
# (for socket-activation the port will be managed by systemd)
if not isinstance(self.bind_addr, tuple):
return
# wait for port to be occupied
with _safe_wait(*self.bound_addr):
portend.occupied(*self.bound_addr, timeout=Timeouts.occupied)