### Question:
RlpxAgent { public void connect(final Stream<? extends Peer> peerStream) { if (!localNode.isReady()) { return; } final int availablePeerSlots = Math.max(0, maxConnections - getConnectionCount()); peerStream .filter(peer -> !connectionsById.containsKey(peer.getId())) .filter(peer -> peer.getEnodeURL().isListening()) .filter(peerPermissions::allowNewOutboundConnectionTo) .limit(availablePeerSlots) .forEach(this::connect); } private  RlpxAgent(
      final LocalNode localNode,
      final PeerConnectionEvents connectionEvents,
      final ConnectionInitializer connectionInitializer,
      final PeerRlpxPermissions peerPermissions,
      final PeerPrivileges peerPrivileges,
      final int maxConnections,
      final int maxRemotelyInitiatedConnections,
      final MetricsSystem metricsSystem); static Builder builder(); CompletableFuture<Integer> start(); CompletableFuture<Void> stop(); Stream<? extends PeerConnection> streamConnections(); int getConnectionCount(); void connect(final Stream<? extends Peer> peerStream); void disconnect(final Bytes peerId, final DisconnectReason reason); Optional<CompletableFuture<PeerConnection>> getPeerConnection(final Peer peer); CompletableFuture<PeerConnection> connect(final Peer peer); void subscribeMessage(final Capability capability, final MessageCallback callback); void subscribeConnect(final ConnectCallback callback); void subscribeDisconnect(final DisconnectCallback callback);  }### Answer:
@Test public void connect_fails() { connectionInitializer.setAutocompleteConnections(false); startAgent(); final Peer peer = createPeer(); final CompletableFuture<PeerConnection> connection = agent.connect(peer); connection.completeExceptionally(new RuntimeException("whoopsies")); assertPeerConnectionNotTracked(peer); }
@Test public void connect_toDiscoveryPeerUpdatesStats() { startAgent(); final DiscoveryPeer peer = DiscoveryPeer.fromEnode(enode()); assertThat(peer.getLastAttemptedConnection()).isEqualTo(0); final CompletableFuture<PeerConnection> connection = agent.connect(peer); assertThat(connection).isDone(); assertThat(connection).isNotCompletedExceptionally(); assertThat(peer.getLastAttemptedConnection()).isGreaterThan(0); }
@Test public void connect_failsIfAgentIsNotReady() { final Peer peer = createPeer(); final CompletableFuture<PeerConnection> connection = agent.connect(peer); assertThat(connection).isDone(); assertThat(connection).isCompletedExceptionally(); assertThatThrownBy(connection::get) .hasCauseInstanceOf(IllegalStateException.class) .hasMessageContaining("Cannot connect before") .hasMessageContaining("has finished starting"); assertPeerConnectionNotTracked(peer); }
@Test public void connect_failsIfPeerIsNotListening() { startAgent(); final Peer peer = DefaultPeer.fromEnodeURL(enodeBuilder().disableListening().build()); final CompletableFuture<PeerConnection> connection = agent.connect(peer); assertThat(connection).isDone(); assertThat(connection).isCompletedExceptionally(); assertThatThrownBy(connection::get) .hasCauseInstanceOf(IllegalArgumentException.class) .hasMessageContaining("Attempt to connect to peer with no listening port"); assertPeerConnectionNotTracked(peer); }
@Test public void connect_failsIfPeerIsNotPermitted() { startAgent(); final Peer peer = createPeer(); doReturn(false) .when(peerPermissions) .isPermitted( eq(localNode.getPeer()), eq(peer), eq(PeerPermissions.Action.RLPX_ALLOW_NEW_OUTBOUND_CONNECTION)); final CompletableFuture<PeerConnection> connection = agent.connect(peer); assertThat(connection).isDone(); assertThat(connection).isCompletedExceptionally(); assertThatThrownBy(connection::get).hasCauseInstanceOf(PeerPermissionsException.class); assertPeerConnectionNotTracked(peer); }