How to use portend - 10 common examples

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 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 check_port_availability(host, port):
    """
    Make sure the port is available for the server to start.
    """
    try:
        portend.free(host, port, timeout=PORT_AVAILABILITY_CHECK_TIMEOUT)
    except portend.Timeout:
        # Bypass check when socket activation is used
        # https://manpages.debian.org/testing/libsystemd-dev/sd_listen_fds.3.en.html#ENVIRONMENT
        if not os.environ.get("LISTEN_PID", None):
            # Port is occupied
            logger.error(
                "Port {} is occupied.\n"
                "Please check that you do not have other processes "
                "running on this port and try again.\n".format(port)
            )
            sys.exit(1)

def start(self):
        """Start the HTTP server."""
        if self.running:
            self.bus.log('Already serving on %s' % self.description)
            return

        self.interrupt = None
        if not self.httpserver:
            raise ValueError('No HTTP server has been created.')

        if not os.environ.get('LISTEN_PID', None):
            # Start the httpserver in a new thread.
            if isinstance(self.bind_addr, tuple):
                portend.free(*self.bind_addr, timeout=Timeouts.free)

        import threading
        t = threading.Thread(target=self._start_http_thread)
        t.setName('HTTPServer ' + t.getName())
        t.start()

        self.wait()
        self.running = True
        self.bus.log('Serving on %s' % self.description)
    start.priority = 75

def start(self):
        """Start the HTTP server."""
        if self.running:
            self.bus.log('Already serving on %s' % self.description)
            return

        self.interrupt = None
        if not self.httpserver:
            raise ValueError('No HTTP server has been created.')

        if not os.environ.get('LISTEN_PID', None):
            # Start the httpserver in a new thread.
            if isinstance(self.bind_addr, tuple):
                portend.free(*self.bind_addr, timeout=Timeouts.free)

        import threading
        t = threading.Thread(target=self._start_http_thread)
        t.setName('HTTPServer ' + t.getName())
        t.start()

        self.wait()
        self.running = True
        self.bus.log('Serving on %s' % self.description)
    start.priority = 75

@contextlib.contextmanager
def _safe_wait(host, port):
    """
    On systems where a loopback interface is not available and the
    server is bound to all interfaces, it's difficult to determine
    whether the server is in fact occupying the port. In this case,
    just issue a warning and move on. See issue #1100.
    """
    try:
        yield
    except portend.Timeout:
        if host == portend.client_host(host):
            raise
        msg = 'Unable to verify that the server is bound on %r' % port
        warnings.warn(msg)

def _safe_wait(host, port):
    """
    On systems where a loopback interface is not available and the
    server is bound to all interfaces, it's difficult to determine
    whether the server is in fact occupying the port. In this case,
    just issue a warning and move on. See issue #1100.
    """
    try:
        yield
    except portend.Timeout:
        if host == portend.client_host(host):
            raise
        msg = 'Unable to verify that the server is bound on %r' % port
        warnings.warn(msg)

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)

@contextlib.contextmanager
def _safe_wait(host, port):
    """
    On systems where a loopback interface is not available and the
    server is bound to all interfaces, it's difficult to determine
    whether the server is in fact occupying the port. In this case,
    just issue a warning and move on. See issue #1100.
    """
    try:
        yield
    except portend.Timeout:
        if host == portend.client_host(host):
            raise
        msg = 'Unable to verify that the server is bound on %r' % port
        warnings.warn(msg)