def test_invalid_login(self):
    """Tests if postgres server responds correctly to a invalid login attempt."""

    def postgresql_login():
      try:
        psycopg2.connect("postgres://scott:tiger@0.0.0.0:2504/")
      except psycopg2.OperationalError as e:
        return e
      return None

    options = {'enabled': 'True', 'port': 2504}
    postgresql_cap = postgresql.PostgreSQL(options, self.loop)

    server_coro = asyncio.start_server(
        postgresql_cap.handle_session, '0.0.0.0', 2504, loop=self.loop)
    self.server = self.loop.run_until_complete(server_coro)

    postgresql_task = self.loop.run_in_executor(None, postgresql_login)
    login_exception = self.loop.run_until_complete(postgresql_task)

    self.assertIsInstance(login_exception, psycopg2.OperationalError)
    self.assertEqual(
        str(login_exception),
        'FATAL:  password authentication failed for user "scott"\n') 

def test_connection(self):
    """ Tests if the capability is up, and sending
            HTTP 401 (Unauthorized) headers.
        """

    def http_request():
      client = httpclient.HTTPConnection('127.0.0.1', 8888)
      client.request('GET', '/')
      response = client.getresponse()
      self.assertEqual(response.status, 401)

    options = {'enabled': 'True', 'port': 8888, 'users': {'test': 'test'}}
    http_cap = http.Http(options, self.loop)

    server_coro = asyncio.start_server(
        http_cap.handle_session, '0.0.0.0', 8888, loop=self.loop)
    self.server = self.loop.run_until_complete(server_coro)

    http_task = self.loop.run_in_executor(None, http_request)
    self.loop.run_until_complete(http_task) 

def start_command_server():
    if not ENABLED:
        return

    stop_command_server()

    print(f'starting command server (view host / port in config file)')
    try:
        # noinspection PyTypeChecker
        add_task(COMMAND_SERVER_TASK_ID, start_server(handle_client, HOST, PORT))
    except Exception as e:
        print(f"\n------COMMAND SERVER------\nfailed to bind/create command server\n"
              f"this does not affect the bot, but it does mean that the command console will not work/be usable\n"
              f"if this error happens a lot, command server can be disabled in the config.json in the bot's configs folder\n"
              f'\nERROR INFO: {e}\n'
              f'EXTENDED INFO: \n{format_exc()}\n\n'
              f'------COMMAND SERVER------\n') 

def main():
    if config.tfo:
        loop.set_exception_handler(silent_tpo_timeout_err_handler)

    try:
        server = loop.run_until_complete(
            asyncio.start_server(dispatch_proxy, 'localhost', config.port))
    except OSError:
        die('wstan client failed to bind on localhost:%d' % config.port)

    print('wstan client -- SOCKS5/HTTP(S) server listening on localhost:%d' % config.port)
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.close()
        loop.close() 

def start(self):
        """ Start the server socket

        :return: False if an error prevented us from launching a connection thread. True if a connection thread has been started.
        :rtype: bool
        """
        if self._is_listening:
            return False
        try:
            self.logger.info("Starting up TCP server socket {}".format(self.__our_socket))
            self.__loop = asyncio.new_event_loop()
            asyncio.set_event_loop(self.__loop)
            self.__coroutine = asyncio.start_server(self.__handle_connection, self._interfaceip, self._port)
            self.__server = self.__loop.run_until_complete(self.__coroutine)

            self.__listening_thread = threading.Thread(target=self.__listening_thread_worker, name='TCP_Server_{}'.format(self.name))
            self.__listening_thread.daemon = True
            self.__listening_thread.start()
        except:
            return False
        return True 

def _setup_electrum_server(self, server_info):
        async def methods(r, w):
            responses = {
                'server.version': 'mock 1.2 1.2',
                'blockchain.scripthash.listunspent': 'cafebabe',
                'something.subscribe': 'babe',
                'server.ping': True
            }
            while 1:
                data = await r.read(1024)
                if not data:
                    w.close()
                    break
                else:
                    d = json.loads(data.strip().decode())
                    command = d['method']
                    response = {'result': responses[command], 'id': d['id']}
                    res = json.dumps(response) + '\n'
                    w.write(res.encode())
                    await w.drain()

        host = server_info.hostname
        coro = asyncio.start_server(methods, host=host, port=50001, loop=self.loop)
        return coro 

def test_protocol_timeout_on_starttls(
    event_loop, bind_address, hostname, client_tls_context
):
    async def client_connected(reader, writer):
        await asyncio.sleep(1.0)

    server = await asyncio.start_server(
        client_connected, host=bind_address, port=0, family=socket.AF_INET
    )
    server_port = server.sockets[0].getsockname()[1]

    connect_future = event_loop.create_connection(
        SMTPProtocol, host=hostname, port=server_port
    )

    _, protocol = await asyncio.wait_for(connect_future, timeout=1.0)

    with pytest.raises(SMTPTimeoutError):
        # STARTTLS timeout must be > 0
        await protocol.start_tls(client_tls_context, timeout=0.00001)

    server.close()
    await server.wait_closed() 

def test_error_on_readline_with_partial_line(event_loop, bind_address, hostname):
    partial_response = b"499 incomplete response\\"

    async def client_connected(reader, writer):
        writer.write(partial_response)
        writer.write_eof()
        await writer.drain()

    server = await asyncio.start_server(
        client_connected, host=bind_address, port=0, family=socket.AF_INET
    )
    server_port = server.sockets[0].getsockname()[1]

    connect_future = event_loop.create_connection(
        SMTPProtocol, host=hostname, port=server_port
    )

    _, protocol = await asyncio.wait_for(connect_future, timeout=1.0)

    with pytest.raises(SMTPServerDisconnected):
        await protocol.read_response(timeout=1.0)

    server.close()
    await server.wait_closed() 

def main(address='127.0.0.1', port=2323):  # <1>
    port = int(port)
    loop = asyncio.get_event_loop()
    server_coro = asyncio.start_server(handle_queries, address, port,
                                loop=loop) # <2>
    server = loop.run_until_complete(server_coro) # <3>

    host = server.sockets[0].getsockname()  # <4>
    print('Serving on {}. Hit CTRL-C to stop.'.format(host))  # <5>
    try:
        loop.run_forever()  # <6>
    except KeyboardInterrupt:  # CTRL+C pressed
        pass

    print('Server shutting down.')
    server.close()  # <7>
    loop.run_until_complete(server.wait_closed())  # <8>
    loop.close()  # <9> 

def main(address='127.0.0.1', port=8888):
    port = int(port)
    loop = asyncio.get_event_loop()
    coro = asyncio.start_server(handle_queries, address, port, loop=loop)
    server = loop.run_until_complete(coro)

    host = server.sockets[0].getsockname()
    print('Serving on {}. Hit CTRL-C to stop.'.format(host))
    try:
        loop.run_forever()
    except KeyboardInterrupt:  # CTRL+C pressed
        pass

    server.close()
    loop.run_until_complete(server.wait_closed())
    loop.close() 

def main(address='127.0.0.1', port=2323):  # <1>
    port = int(port)
    loop = asyncio.get_event_loop()
    server_coro = asyncio.start_server(handle_queries, address, port,
                                loop=loop) # <2>
    server = loop.run_until_complete(server_coro) # <3>

    host = server.sockets[0].getsockname()  # <4>
    print('Serving on {}. Hit CTRL-C to stop.'.format(host))  # <5>
    try:
        loop.run_forever()  # <6>
    except KeyboardInterrupt:  # CTRL+C pressed
        pass

    print('Server shutting down.')
    server.close()  # <7>
    loop.run_until_complete(server.wait_closed())  # <8>
    loop.close()  # <9> 

def run_server(bind='127.0.0.1', port=8888):

    # Start the server
    loop = asyncio.get_event_loop()
    coro = asyncio.start_server(euclidean_norm_handler, bind, port)
    server = loop.run_until_complete(coro)

    # Serve requests until Ctrl+C is pressed
    print('Serving on {}'.format(server.sockets[0].getsockname()))
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        pass

    # Close the server
    server.close()
    loop.run_until_complete(server.wait_closed())
    loop.close()


# Main execution 

def start(self):
        loop = asyncio.get_event_loop()
        tcp = self.get_config('app.contact.tcp')
        loop.create_task(asyncio.start_server(self.tcp_handler.accept, *tcp.split(':'), loop=loop))
        loop.create_task(self.operation_loop()) 

def _start_tcp_listener(self) -> None:
        # TODO: Support IPv6 addresses as well.
        self._tcp_listener = await asyncio.start_server(
            self.receive_handshake, host="0.0.0.0", port=self.port
        ) 

def __start_server(self):
        """ Run the server until shutdown is called """
        self.server = await asyncio.start_server(
            self.__handle_new_connection,
            "0.0.0.0",
            self.env.slave_config.PORT,
            loop=self.loop,
        )
        Logger.info(
            "Listening on {} for intra-cluster RPC".format(
                self.server.sockets[0].getsockname()
            )
        ) 

def start_server(self):
        coro = asyncio.start_server(self.new_peer, "0.0.0.0", self.port, loop=self.loop)
        self.server = self.loop.run_until_complete(coro)
        Logger.info("Self id {}".format(self.self_id.hex()))
        Logger.info(
            "Listening on {} for p2p".format(self.server.sockets[0].getsockname())
        ) 

def start(self):
        self.start_server()

        self.loop.create_task(
            self.connect_seed(
                self.env.cluster_config.SIMPLE_NETWORK.BOOTSTRAP_HOST,
                self.env.cluster_config.SIMPLE_NETWORK.BOOTSTRAP_PORT,
            )
        )

    # ------------------------------- Cluster Peer Management -------------------------------- 

def start(self):
        self._server = await asyncio.start_server(self._handle_connection,
                                                  self._listen_address,
                                                  self._listen_port,
                                                  loop=self._loop) 

def test_vnc_authentication(self):

    async def vnc_auth():
      reader, writer = await asyncio.open_connection(
          '127.0.0.1', 8888, loop=self.loop)
      # server rfb version
      _ = await reader.readline()
      writer.write(RFB_VERSION)

      # available auth methods
      _ = await reader.read(1024)
      writer.write(VNC_AUTH)

      # challenge
      _ = await reader.read(1024)
      # Pretending, that we encrypt received challenge with DES and send back the result.
      client_response = os.urandom(16)
      writer.write(client_response)

      # security result
      _ = await reader.read(1024)

    options = {'enabled': 'True', 'port': 8888, 'timeout': 30}
    capability = Vnc(options, self.loop)

    server_coro = asyncio.start_server(
        capability.handle_session, '0.0.0.0', 8888, loop=self.loop)
    self.server = self.loop.run_until_complete(server_coro)

    self.loop.run_until_complete(vnc_auth()) 

def test_socks_authentication(self):

    async def socks_auth():
      reader, writer = await asyncio.open_connection(
          '127.0.0.1', 8888, loop=self.loop)

      # Greeting to the server. version+authmethod number+authmethod
      client_greeting = socks.SOCKS_VERSION + b"\x01" + socks.AUTH_METHOD
      writer.write(client_greeting)

      # Receive version+chosen authmethod
      _ = await reader.read(2)

      # Send credentials.
      # version+username len+username+password len+password
      credentials = b"\x05\x08username\x08password"
      writer.write(credentials)

      # Receive authmethod+\xff
      res = await reader.read(2)
      self.assertEqual(res, socks.AUTH_METHOD + socks.SOCKS_FAIL)

    options = {'enabled': 'True', 'port': 8888, 'timeout': 30}
    capability = socks.Socks5(options, self.loop)

    server_coro = asyncio.start_server(
        capability.handle_session, '127.0.0.1', 8888, loop=self.loop)
    self.server = self.loop.run_until_complete(server_coro)
    self.loop.run_until_complete(socks_auth()) 

def test_LOGIN(self):
    """Testing different login combinations using simple login auth mechanism."""

    def imap_login():
      login_sequences = [('kajoj_admin', 'thebestpassword'),
                         ('\"kajoj_admin\"', 'the best password')]

      imap_obj = imaplib.IMAP4('127.0.0.1', port=8888)
      for sequence in login_sequences:
        with self.assertRaises(imaplib.IMAP4.error) as error:
          imap_obj.login(sequence[0], sequence[1])
        imap_exception = error.exception
        self.assertEqual(imap_exception.args[0], b'Authentication failed')
      imap_obj.logout()

    options = {
        'enabled': 'True',
        'port': 143,
        'timeout': 30,
        'protocol_specific_data': {
            'max_attempts': 3,
            'banner': '* OK IMAP4rev1 Server Ready'
        }
    }
    capability = Imap(options, self.loop)
    server_coro = asyncio.start_server(
        capability.handle_session, '0.0.0.0', 8888, loop=self.loop)
    self.server = self.loop.run_until_complete(server_coro)

    imap_task = self.loop.run_in_executor(None, imap_login)
    self.loop.run_until_complete(imap_task) 

def test_AUTH_CRAM_MD5_reject(self):
    """ Makes sure the server rejects all invalid login attempts that use the
            CRAM-MD5 Authentication method.
        """

    def encode_cram_md5(challenge, user, password):
      challenge = base64.decodebytes(challenge)
      response = user + b' ' + bytes(
          hmac.HMAC(password, challenge, digestmod="md5").hexdigest(), 'utf-8')
      return str(base64.b64encode(response), 'utf-8')

    def smtp_auth_cram_md5():
      smtp_ = smtplib.SMTP(
          '127.0.0.1', 8888, local_hostname='localhost', timeout=15)
      _, resp = smtp_.docmd('AUTH', 'CRAM-MD5')
      code, resp = smtp_.docmd(encode_cram_md5(resp, b'test', b'test'))
      smtp_.quit()
      # For now, the server's going to return a 535 code.
      self.assertEqual(code, 535)

    options = {
        'enabled': 'True',
        'port': 8888,
        'protocol_specific_data': {
            'banner': 'Test'
        },
        'users': {
            'someguy': 'test'
        }
    }
    smtp_cap = smtp.smtp(options, self.loop)

    server_coro = asyncio.start_server(
        smtp_cap.handle_session, '0.0.0.0', 8888, loop=self.loop)
    self.server = self.loop.run_until_complete(server_coro)

    smtp_task = self.loop.run_in_executor(None, smtp_auth_cram_md5)
    self.loop.run_until_complete(smtp_task) 

def test_AUTH_PLAIN_reject(self):
    """ Makes sure the server rejects all invalid login attempts that use the PLAIN Authentication method.
        """

    def smtp_auth_plain_reject():
      smtp_ = smtplib.SMTP(
          '127.0.0.1', 8888, local_hostname='localhost', timeout=15)
      arg = bytes('\0{0}\0{1}'.format('test', 'test'), 'utf-8')
      code, _ = smtp_.docmd('AUTH',
                            'PLAIN ' + str(base64.b64encode(arg), 'utf-8'))
      smtp_.quit()
      self.assertEqual(code, 535)

    options = {
        'enabled': 'True',
        'port': 0,
        'protocol_specific_data': {
            'banner': 'Test'
        },
        'users': {
            'someguy': 'test'
        }
    }

    smtp_cap = smtp.smtp(options, self.loop)

    server_coro = asyncio.start_server(
        smtp_cap.handle_session, '0.0.0.0', 8888, loop=self.loop)
    self.server = self.loop.run_until_complete(server_coro)

    smtp_task = self.loop.run_in_executor(None, smtp_auth_plain_reject)
    self.loop.run_until_complete(smtp_task) 

def test_AUTH_LOGIN_reject(self):
    """ Makes sure the server rejects all invalid login attempts that use the LOGIN Authentication method.
        """

    def smtp_auth_login_reject():
      smtp_ = smtplib.SMTP(
          '127.0.0.1', 8888, local_hostname='localhost', timeout=15)
      smtp_.docmd('AUTH', 'LOGIN')
      smtp_.docmd(str(base64.b64encode(b'test'), 'utf-8'))
      code, _ = smtp_.docmd(str(base64.b64encode(b'test'), 'utf-8'))
      smtp_.quit()
      self.assertEqual(code, 535)

    options = {
        'enabled': 'True',
        'port': 0,
        'protocol_specific_data': {
            'banner': 'Test'
        },
        'users': {
            'someguy': 'test'
        }
    }

    smtp_cap = smtp.smtp(options, self.loop)

    server_coro = asyncio.start_server(
        smtp_cap.handle_session, '0.0.0.0', 8888, loop=self.loop)
    self.server = self.loop.run_until_complete(server_coro)

    smtp_task = self.loop.run_in_executor(None, smtp_auth_login_reject)
    self.loop.run_until_complete(smtp_task) 

def test_login(self):
    """Testing different login combinations"""

    def ftp_login():
      ftp_client = FTP()
      ftp_client.connect('127.0.0.1', 8888, 1)
      # expect perm exception
      try:
        ftp_client.login('james', 'bond')
        _ = ftp_client.getresp()  # NOQA
      except ftplib.error_perm:
        ftp_client.quit()

    options = {
        'enabled': 'True',
        'port': 0,
        'banner': 'Test Banner',
        'users': {
            'test': 'test'
        },
        'protocol_specific_data': {
            'max_attempts': 3,
            'banner': 'test banner',
            'syst_type': 'Test Type'
        }
    }

    ftp_capability = ftp.ftp(options, self.loop)

    server_coro = asyncio.start_server(
        ftp_capability.handle_session, '0.0.0.0', 8888, loop=self.loop)
    self.server = self.loop.run_until_complete(server_coro)

    ftp_task = self.loop.run_in_executor(None, ftp_login)
    self.loop.run_until_complete(ftp_task) 

def test_invalid_login(self):
    """Tests if mysql server responds correctly to a invalid login attempt."""

    def mysql_login():
      try:
        pymysql.connect(
            host="0.0.0.0",
            port=8306,
            user="tuser",
            password="tpass",
            db="testdb")
      except pymysql.err.OperationalError as e:
        return e
      return None

    options = {'enabled': 'True', 'port': 8306}
    mysql_cap = mysql.MySQL(options, self.loop)

    server_coro = asyncio.start_server(
        mysql_cap.handle_session, '0.0.0.0', 8306, loop=self.loop)
    self.server = self.loop.run_until_complete(server_coro)

    mysql_task = self.loop.run_in_executor(None, mysql_login)
    login_exception = self.loop.run_until_complete(mysql_task)

    self.assertIsInstance(login_exception, pymysql.err.OperationalError)
    self.assertEqual(
        str(login_exception),
        '(1045, "Access denied for user \'tuser\'@\'127.0.0.1\' (using password: YES)")'
    ) 

def create_tcp_server(core: Core, family, host_port):
    assert family == socket.AF_INET or family == socket.AF_INET6
    coroutine = asyncio.start_server(
        core.initial_connection_check, host_port[0], host_port[1],
        family=family, backlog=core.backlog, loop=loop)
    abstract_server = loop.run_until_complete(coroutine)
    for sock in abstract_server.sockets:
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    return abstract_server 

def listener():
    def _acceptClient(clientReader, clientWriter):
        pass

    loop = asyncio.get_event_loop()
    server = loop.run_until_complete(
        asyncio.start_server(_acceptClient,
                             host=STATS_SERVER_IP, port=STATS_SERVER_PORT,
                             loop=loop))
    yield server
    server.close()
    loop.run_until_complete(server.wait_closed()) 

def create_servers(loop):
    servers = []

    reuse_port = hasattr(socket, "SO_REUSEPORT")
    has_unix = hasattr(socket, "AF_UNIX")

    if config.LISTEN_ADDR_IPV4:
        task = asyncio.start_server(handle_client_wrapper, config.LISTEN_ADDR_IPV4, config.PORT,
                                    limit=get_to_tg_bufsize(), reuse_port=reuse_port)
        servers.append(loop.run_until_complete(task))

    if config.LISTEN_ADDR_IPV6 and socket.has_ipv6:
        task = asyncio.start_server(handle_client_wrapper, config.LISTEN_ADDR_IPV6, config.PORT,
                                    limit=get_to_tg_bufsize(), reuse_port=reuse_port)
        servers.append(loop.run_until_complete(task))

    if config.LISTEN_UNIX_SOCK and has_unix:
        remove_unix_socket(config.LISTEN_UNIX_SOCK)
        task = asyncio.start_unix_server(handle_client_wrapper, config.LISTEN_UNIX_SOCK,
                                         limit=get_to_tg_bufsize())
        servers.append(loop.run_until_complete(task))
        os.chmod(config.LISTEN_UNIX_SOCK, 0o666)

    if config.METRICS_PORT is not None:
        if config.METRICS_LISTEN_ADDR_IPV4:
            task = asyncio.start_server(handle_metrics, config.METRICS_LISTEN_ADDR_IPV4,
                                        config.METRICS_PORT)
            servers.append(loop.run_until_complete(task))
        if config.METRICS_LISTEN_ADDR_IPV6 and socket.has_ipv6:
            task = asyncio.start_server(handle_metrics, config.METRICS_LISTEN_ADDR_IPV6,
                                        config.METRICS_PORT)
            servers.append(loop.run_until_complete(task))

    return servers 

def _pre_start(self) -> None:
        await super()._pre_start()
        if self._auth_cookie_file is None:
            cookie_dir = self._stack.enter_context(
                tempfile.TemporaryDirectory(
                    prefix=__package__ + '_authcookie_'))
            self._auth_cookie_file = os.path.join(
                cookie_dir, AUTH_COOKIE_FILENAME)
        self._authenticator.write_cookie_file(self._auth_cookie_file)
        self._server = await asyncio.start_server(
            self._ext_or_port_handler, self._ext_host, self._ext_port,
            family=self._ext_family)
        await self._stack.enter_async_context(self._server)