migrated connect_to_unbanned_peer test (near#8196)

Migrated connect_to_unbanned_peer test from integration-tests to near-network.
It was flaky when executed in presubmit. I've made it more synchronized and faked out time, so now it shouldn't cause problems any more. I've tested it locally for flakiness (multiple concurrent runs in a loop) and observed no problems.

To accomodate the test logic, I reorganized a bit the implementation of some PeerStore methods, but no semantic change was introduced there.

chain/network/src/peer_manager/tests/routing.rs

@@ -1378,3 +1378,63 @@ async fn connect_to_unbanned_peer() {
     drop(pm0);
     drop(pm1);
 }
+
+/// Awaits for ConnectionClosed event for a given `stream_id`.
+async fn wait_for_stream_closed(
+    events: &mut broadcast::Receiver<Event>,
+    stream_id: tcp::StreamId,
+) -> ClosingReason {
+    events
+        .recv_until(|ev| match ev {
+            Event::PeerManager(PME::ConnectionClosed(ev)) if ev.stream_id == stream_id => {
+                Some(ev.reason)
+            }
+            _ => None,
+        })
+        .await
+}
+
+/// Check two peers are able to connect again after one peers is banned and unbanned.
+#[tokio::test]
+async fn connect_to_unbanned_peer() {
+    init_test_logger();
+    let mut rng = make_rng(921853233);
+    let rng = &mut rng;
+    let mut clock = time::FakeClock::default();
+    let chain = Arc::new(data::Chain::make(&mut clock, rng, 10));
+
+    let mut pm0 =
+        start_pm(clock.clock(), TestDB::new(), chain.make_config(rng), chain.clone()).await;
+    let mut pm1 =
+        start_pm(clock.clock(), TestDB::new(), chain.make_config(rng), chain.clone()).await;
+
+    tracing::info!(target:"test", "pm0 connects to pm1");
+    let stream_id = pm0.connect_to(&pm1.peer_info(), tcp::Tier::T2).await;
+
+    tracing::info!(target:"test", "pm1 bans pm0");
+    let ban_reason = ReasonForBan::BadBlock;
+    pm1.disconnect_and_ban(&clock.clock(), &pm0.cfg.node_id(), ban_reason).await;
+    wait_for_stream_closed(&mut pm0.events, stream_id).await;
+    assert_eq!(
+        ClosingReason::Ban(ban_reason),
+        wait_for_stream_closed(&mut pm1.events, stream_id).await
+    );
+
+    tracing::info!(target:"test", "pm0 fails to reconnect to pm1");
+    let got_reason = pm1
+        .start_inbound(chain.clone(), pm0.cfg.clone())
+        .await
+        .manager_fail_handshake(&clock.clock())
+        .await;
+    assert_eq!(ClosingReason::RejectedByPeerManager(RegisterPeerError::Banned), got_reason);
+
+    tracing::info!(target:"test", "pm1 unbans pm0");
+    clock.advance(pm1.cfg.peer_store.ban_window);
+    pm1.peer_store_update(&clock.clock()).await;
+
+    tracing::info!(target:"test", "pm0 reconnects to pm1");
+    pm0.connect_to(&pm1.peer_info(), tcp::Tier::T2).await;
+
+    drop(pm0);
+    drop(pm1);
+}