use std::sync::atomic::{AtomicU64, Ordering};

use super::*;
use crate::document::replica::Epoch;
use crate::transport::{HeadsSummary, Status};

/// Memory ports are process-wide.
static NEXT_PORT: AtomicU64 = AtomicU64::new(40_000);

/// A node on the in-memory transport, listening on a port of its own (or nowhere),
/// and its address with its peer ID (or just its peer ID).
fn memory_node_with(
    listen: bool,
    relays: Vec<Multiaddr>,
    relay: bool,
) -> (Libp2pTransport, Multiaddr) {
    let port = NEXT_PORT.fetch_add(1, Ordering::Relaxed);
    let key = Keypair::generate_ed25519();
    let address: Multiaddr = format!("/memory/{port}").parse().unwrap();
    let peer = key.public().to_peer_id();
    let transport = Libp2pTransport::start(Libp2pConfig {
        key,
        listen: if listen {
            vec![address.clone()]
        } else {
            Vec::new()
        },
        announce: Vec::new(),
        relays,
        relay,
        continuum: true,
        mdns: false,
        upnp: false,
        ignore: Vec::new(),
        prefer: Vec::new(),
        internet_hosts: Vec::new(),
        public: false,
        memory: true,
        keepalive: Duration::from_secs(60),
    })
    .unwrap();
    let address = if listen {
        address.with_p2p(peer).unwrap()
    } else {
        Multiaddr::empty().with(Protocol::P2p(peer))
    };
    (transport, address)
}

fn memory_node() -> (Libp2pTransport, Multiaddr) {
    memory_node_with(true, Vec::new(), false)
}

fn status() -> ParsedMessage {
    ParsedMessage::Status(Status {
        heads: HeadsSummary::of(Vec::new()),
        epoch: Epoch::GENESIS,
        epoch_seq: 3,
        ready: true,
        syncs_with_you: false,
        leaving: false,
    })
}

/// Starts `transport`'s consumer, forwarding what it receives.
fn received(transport: Arc<Libp2pTransport>) -> mpsc::UnboundedReceiver<(String, ParsedMessage)> {
    let (sender, receiver) = mpsc::unbounded_channel();
    tokio::spawn(async move {
        transport
            .start_consumer(Box::new(move |from, message, _| {
                let sender = sender.clone();
                Box::pin(async move {
                    let _ = sender.send((from, message));
                    Ok(())
                })
            }))
            .await
    });
    receiver
}

#[tokio::test]
async fn messages_both_ways_over_one_connection() {
    let (a, a_address) = memory_node();
    let (b, _) = memory_node();
    let (a, b) = (Arc::new(a), Arc::new(b));
    let mut at_a = received(Arc::clone(&a));
    let mut at_b = received(Arc::clone(&b));

    assert_eq!(
        b.resolve_seed(&a_address.to_string()).await.unwrap(),
        a.local_id()
    );
    b.send_message(a.local_id(), &status()).await.unwrap();
    let (from, message) = at_a.recv().await.unwrap();
    assert_eq!(from, b.local_id());
    assert!(matches!(message, ParsedMessage::Status(status) if status.epoch_seq == 3));

    // a knows no address of b, but b's connection is there.
    a.send_message(b.local_id(), &status()).await.unwrap();
    let (from, _) = at_b.recv().await.unwrap();
    assert_eq!(from, a.local_id());

    assert_eq!(a.connection(b.local_id()), Some(Connection::Direct));
    assert_eq!(b.connection(a.local_id()), Some(Connection::Direct));
    let stats = a.stats().libp2p.unwrap();
    assert!(stats.connections.direct >= 1);
    assert_eq!(stats.connections.relayed, 0);
}

#[tokio::test]
async fn a_seed_without_a_peer_id_is_connected_to_to_find_out() {
    let (a, a_address) = memory_node();
    let (b, _) = memory_node();
    let mut bare = a_address.clone();
    bare.pop();
    assert_eq!(
        b.resolve_seed(&bare.to_string()).await.unwrap(),
        a.local_id()
    );
    // And b now knows where a is.
    assert_eq!(b.locations(a.local_id()), [a_address.to_string()]);
}

#[tokio::test]
async fn listen_addresses_carry_the_peer_id() {
    let (a, a_address) = memory_node();
    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    while a.local_addresses().is_empty() && tokio::time::Instant::now() < deadline {
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
    assert_eq!(a.local_addresses(), [a_address.to_string()]);
    assert_eq!(
        a.locate(&a_address.to_string()).as_deref(),
        Some(a.local_id())
    );
}

/// Waits up to 5 s for `condition`.
async fn eventually(mut condition: impl FnMut() -> bool) -> bool {
    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    while !condition() {
        if tokio::time::Instant::now() >= deadline {
            return false;
        }
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
    true
}

/// A node learns where a peer that dialled it listens (identify), and can then
/// pass it on.
#[tokio::test]
async fn a_dialled_node_learns_where_the_dialler_listens() {
    let (a, a_address) = memory_node();
    let (b, b_address) = memory_node();
    let (a, b) = (Arc::new(a), Arc::new(b));
    let _at_a = received(Arc::clone(&a));
    b.resolve_seed(&a_address.to_string()).await.unwrap();
    b.send_message(a.local_id(), &status()).await.unwrap();
    let b_id = b.local_id().to_owned();
    assert!(
        eventually(|| a.locations(&b_id) == [b_address.to_string()]).await,
        "a knows {:?} of b",
        a.locations(&b_id)
    );
}

#[tokio::test]
async fn announced_addresses_are_published_instead() {
    let key = Keypair::generate_ed25519();
    let peer = key.public().to_peer_id();
    let port = NEXT_PORT.fetch_add(1, Ordering::Relaxed);
    let announced: Multiaddr = "/ip4/203.0.113.7/tcp/9661".parse().unwrap();
    let transport = Libp2pTransport::start(Libp2pConfig {
        key,
        listen: vec![format!("/memory/{port}").parse().unwrap()],
        announce: vec![announced.clone()],
        relays: Vec::new(),
        relay: false,
        continuum: true,
        mdns: false,
        upnp: false,
        ignore: Vec::new(),
        prefer: Vec::new(),
        internet_hosts: Vec::new(),
        public: false,
        memory: true,
        keepalive: Duration::from_secs(60),
    })
    .unwrap();
    assert_eq!(
        transport.local_addresses(),
        [announced.with_p2p(peer).unwrap().to_string()]
    );
}

#[test]
fn address_ranges() {
    let range = |text: &str| text.parse::<IpRange>().unwrap();
    let address = |text: &str| text.parse::<Multiaddr>().unwrap();
    let ranges = [range("10.244.0.0/16"), range("42::/16"), range("192.0.2.7")];
    for left_out in [
        "/ip4/10.244.3.2/udp/9661/quic-v1",
        "/ip6/42:1:a46e:be4a:cb58:e460:bf0:142c/tcp/9661",
        "/ip4/192.0.2.7/tcp/1",
    ] {
        assert!(ignored(&ranges, &address(left_out)), "{left_out}");
    }
    for kept in [
        "/ip4/10.23.1.144/udp/9661/quic-v1",
        "/ip4/10.245.0.1/tcp/1",
        "/ip6/2a0a:4cc0:c1:5eb0::1/udp/9661/quic-v1",
        "/ip4/192.0.2.8/tcp/1",
        "/dns4/example.org/tcp/1",
    ] {
        assert!(!ignored(&ranges, &address(kept)), "{kept}");
    }
    // Through a relay: judged by the connection to the relay, not here.
    let relayed = format!("/ip4/10.244.0.1/tcp/1/p2p/{}/p2p-circuit", peer());
    assert!(!ignored(&ranges, &address(&relayed)));
    assert!(range("0.0.0.0/0").contains("203.0.113.1".parse().unwrap()));
    assert!(!range("0.0.0.0/0").contains("::1".parse().unwrap()));
    for wrong in ["10.244.0.0/33", "42::/129", "nowhere", "10.0.0.0/x"] {
        assert!(wrong.parse::<IpRange>().is_err(), "{wrong}");
    }
    // Mycelium's range: a prefix that isn't a multiple of 8.
    let mycelium = [range("400::/7")];
    assert!(ignored(
        &mycelium,
        &address("/ip6/45f:fa21:4bdd:1::1/tcp/9661")
    ));
    assert!(ignored(&mycelium, &address("/ip6/5ff::1/tcp/9661")));
    assert!(!ignored(&mycelium, &address("/ip6/600::1/tcp/9661")));
}

/// A preferred range all ignored is refused; one with an ignored range in it is fine
/// (with a warning).
#[test]
fn preferred_and_ignored_ranges() {
    let check = |ignore: &[&str], prefer: &[&str]| {
        let given = |ranges: &[&str]| {
            ranges
                .iter()
                .map(|range| range.to_string())
                .collect::<Vec<_>>()
        };
        let (ignore, prefer) = (given(ignore), given(prefer));
        super::config::check_preferred(
            &super::config::parse_ranges("libp2p-ignore", &ignore).unwrap(),
            &super::config::parse_ranges("libp2p-prefer", &prefer).unwrap(),
            &ignore,
            &prefer,
        )
    };
    assert!(check(&["10.244.0.0/16"], &["10.23.0.0/16"]).is_ok());
    assert!(check(&["10.0.0.0/8"], &["10.23.0.0/16"]).is_err());
    assert!(check(&["10.23.0.0/16"], &["10.23.0.0/16"]).is_err());
    assert!(check(&["10.23.5.0/24"], &["10.23.0.0/16"]).is_ok());
    assert!(check(&["42::/16"], &["10.0.0.0/8"]).is_ok());
}

/// Ranges left out aren't published; announced addresses are, whatever they are.
#[test]
fn ignored_addresses_are_not_published() {
    let parse = |text: &str| text.parse::<Multiaddr>().unwrap();
    let lan = parse("/ip4/10.23.1.5/udp/9663/quic-v1");
    let vpn = parse("/ip4/10.244.3.2/udp/9663/quic-v1");
    let mut book = Book {
        listening: vec![lan.clone(), vpn.clone()],
        ignored: vec!["10.244.0.0/16".parse().unwrap()],
        ..Book::default()
    };
    assert_eq!(book.direct(), [lan]);
    book.announced = vec![vpn.clone()];
    assert_eq!(book.direct(), [vpn]);
}

/// A port forwarded by hand: once AutoNAT found the node dialable at the router's
/// address, that is published with the listen addresses; announced ones still win.
#[test]
fn the_address_autonat_confirmed_is_published() {
    let parse = |text: &str| text.parse::<Multiaddr>().unwrap();
    let lan = parse("/ip4/10.23.1.5/udp/9663/quic-v1");
    let public = parse("/ip4/91.65.180.254/udp/9663/quic-v1");
    let mut book = Book {
        listening: vec![lan.clone()],
        ..Book::default()
    };
    assert_eq!(book.direct(), [lan.clone()]);
    book.public_address = Some(public.clone().with_p2p(peer()).unwrap());
    assert_eq!(book.direct(), [public.clone(), lan.clone()]);
    let announced = parse("/dns4/example.org/udp/9663/quic-v1");
    book.announced = vec![announced.clone()];
    assert_eq!(book.direct(), [announced]);
}

#[test]
fn local_addresses_are_left_out_unless_there_are_only_local_ones() {
    let parse = |text: &str| text.parse::<Multiaddr>().unwrap();
    let loopback = parse("/ip4/127.0.0.1/tcp/1");
    let link = parse("/ip6/fe80::1/tcp/1");
    let lan = parse("/ip4/192.168.1.2/tcp/1");
    let wireguard = parse("/ip6/fd00::2/tcp/1");
    assert_eq!(
        published(&[
            loopback.clone(),
            link.clone(),
            lan.clone(),
            wireguard.clone()
        ]),
        [lan, wireguard]
    );
    assert_eq!(
        published(&[loopback.clone(), link.clone()]),
        [loopback, link]
    );
}

/// b listens nowhere but through the relay r; a reaches it there.
#[tokio::test]
async fn a_node_nobody_can_dial_is_reached_through_a_relay() {
    let (_r, r_address) = memory_node_with(true, Vec::new(), true);
    let (b, _) = memory_node_with(false, vec![r_address.clone()], false);
    let b = Arc::new(b);
    let mut at_b = received(Arc::clone(&b));
    assert!(
        eventually(|| !b.local_addresses().is_empty()).await,
        "b got no slot on the relay"
    );
    let relayed = b.local_addresses()[0].clone();
    assert!(
        relayed.starts_with(&format!("{r_address}/p2p-circuit/")),
        "{relayed}"
    );

    assert!(eventually(|| b.stats().libp2p.unwrap().relay_slots.held == 1).await);
    let slots = b.stats().libp2p.unwrap().relay_slots;
    assert_eq!((slots.granted, slots.not_granted), (1, 0));

    let (a, _) = memory_node();
    assert_eq!(a.resolve_seed(&relayed).await.unwrap(), b.local_id());
    a.send_message(b.local_id(), &status()).await.unwrap();
    let (from, _) = at_b.recv().await.unwrap();
    assert_eq!(from, a.local_id());
    assert!(a.connection(b.local_id()).is_some());
}

/// A seed that's a peer ID only is reached through the relays the node knows.
#[tokio::test]
async fn a_peer_id_seed_is_reached_through_a_relay() {
    let (_r, r_address) = memory_node_with(true, Vec::new(), true);
    let (b, _) = memory_node_with(false, vec![r_address.clone()], false);
    let b = Arc::new(b);
    let mut at_b = received(Arc::clone(&b));
    assert!(
        eventually(|| !b.local_addresses().is_empty()).await,
        "b got no slot on the relay"
    );
    let (a, _) = memory_node_with(true, vec![r_address.clone()], false);
    let seed = format!("/p2p/{}", b.local_id());
    assert_eq!(a.resolve_seed(&seed).await.unwrap(), b.local_id());
    assert_eq!(a.resolve_seed(b.local_id()).await.unwrap(), b.local_id());
    // Nothing to dial is remembered.
    assert!(!lock(&a.book).addresses.contains_key(&b.peer_id));
    a.send_message(b.local_id(), &status()).await.unwrap();
    let (from, _) = at_b.recv().await.unwrap();
    assert_eq!(from, a.local_id());

    // Without a relay, there's nowhere to look.
    let (c, _) = memory_node();
    assert!(c.resolve_seed(&seed).await.is_err());
}

/// b listens nowhere; a, which can be dialled, reaches it through the relay r: b
/// then dials a directly (identify told it where), and both keep that.
#[tokio::test]
async fn a_relayed_connection_is_upgraded_where_one_side_can_be_dialled() {
    let (_r, r_address) = memory_node_with(true, Vec::new(), true);
    let (b, _) = memory_node_with(false, vec![r_address.clone()], false);
    let b = Arc::new(b);
    let mut at_b = received(Arc::clone(&b));
    assert!(
        eventually(|| !b.local_addresses().is_empty()).await,
        "b got no slot on the relay"
    );
    let (a, _) = memory_node();
    let relayed = b.local_addresses()[0].clone();
    assert_eq!(a.resolve_seed(&relayed).await.unwrap(), b.local_id());
    a.send_message(b.local_id(), &status()).await.unwrap();
    at_b.recv().await.unwrap();
    assert!(
        eventually(|| b.connection(a.local_id()) == Some(Connection::Direct)).await,
        "b didn't dial a directly"
    );
    assert_eq!(a.connection(b.local_id()), Some(Connection::Direct));
    // The relayed one goes (`DUPLICATE_GRACE`); the direct one stays.
    let deadline = tokio::time::Instant::now() + DUPLICATE_GRACE * 2;
    while a.stats().libp2p.unwrap().connections.relayed > 0 {
        assert!(
            tokio::time::Instant::now() < deadline,
            "the relayed connection stayed"
        );
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    b.send_message(a.local_id(), &status()).await.unwrap();
    assert_eq!(a.connection(b.local_id()), Some(Connection::Direct));
}

/// A relay that is no node (`--relay-only`) relays, and takes no messages itself.
#[tokio::test]
async fn a_relay_that_is_no_node() {
    let port = NEXT_PORT.fetch_add(1, Ordering::Relaxed);
    let key = Keypair::generate_ed25519();
    let r_peer = key.public().to_peer_id();
    let r_listen: Multiaddr = format!("/memory/{port}").parse().unwrap();
    let r = Libp2pTransport::start(Libp2pConfig {
        key,
        listen: vec![r_listen.clone()],
        announce: Vec::new(),
        relays: Vec::new(),
        relay: true,
        continuum: false,
        mdns: false,
        upnp: false,
        ignore: Vec::new(),
        prefer: Vec::new(),
        internet_hosts: Vec::new(),
        public: false,
        memory: true,
        keepalive: Duration::from_secs(60),
    })
    .unwrap();
    let r_address = r_listen.with_p2p(r_peer).unwrap();
    let (b, _) = memory_node_with(false, vec![r_address.clone()], false);
    let b = Arc::new(b);
    let mut at_b = received(Arc::clone(&b));
    assert!(
        eventually(|| !b.local_addresses().is_empty()).await,
        "b got no slot on the relay"
    );
    let (a, _) = memory_node();
    let relayed = b.local_addresses()[0].clone();
    assert_eq!(a.resolve_seed(&relayed).await.unwrap(), b.local_id());
    a.send_message(b.local_id(), &status()).await.unwrap();
    let (from, _) = at_b.recv().await.unwrap();
    assert_eq!(from, a.local_id());

    // The relay itself doesn't speak continuum: it takes no messages, and a send to
    // it is refused once identify has told (the relay's own refusal comes too late:
    // the protocol goes with the first data, `swarm_config`), saying what to give
    // as a seed instead.
    assert_eq!(
        a.resolve_seed(&r_address.to_string()).await.unwrap(),
        r.local_id()
    );
    assert!(
        eventually(|| lock(&a.book).not_nodes.get(&r.peer_id) == Some(&true)).await,
        "identify didn't tell"
    );
    let refused = a.send_message(r.local_id(), &status()).await.unwrap_err();
    assert!(
        refused.to_string().contains("is a relay, not a node"),
        "{refused:#}"
    );
    assert!(
        r.start_consumer(Box::new(|_, _, _| Box::pin(async { Ok(()) })))
            .await
            .is_err()
    );
}

/// A relay configured without its peer ID is found out, and used.
#[tokio::test]
async fn a_relay_configured_without_its_peer_id() {
    let (_r, r_address) = memory_node_with(true, Vec::new(), true);
    let mut bare = r_address.clone();
    bare.pop();
    let (b, _) = memory_node_with(false, vec![bare], false);
    assert!(
        eventually(|| !b.local_addresses().is_empty()).await,
        "b got no slot on the relay"
    );
    assert!(b.local_addresses()[0].starts_with(&format!("{r_address}/p2p-circuit/")));
}

fn peer() -> PeerId {
    Keypair::generate_ed25519().public().to_peer_id()
}

#[test]
fn relays_are_picked_when_needed() {
    use ::libp2p::autonat::NatStatus;
    let local = peer();
    let offers: HashMap<PeerId, Multiaddr> = (0..4)
        .map(|port| (peer(), format!("/memory/{port}").parse().unwrap()))
        .collect();
    let check =
        |direct, nat: &NatStatus, using: &[PeerId]| relay_check(local, direct, nat, &offers, using);
    // Dialable, or not known yet with an address to dial: none.
    assert!(check(true, &NatStatus::Public(Multiaddr::empty()), &[]).is_empty());
    assert!(check(false, &NatStatus::Public(Multiaddr::empty()), &[]).is_empty());
    assert!(check(true, &NatStatus::Unknown, &[]).is_empty());
    // No address to dial, or AutoNAT says it can't be dialled: some.
    assert_eq!(check(false, &NatStatus::Unknown, &[]).len(), AUTO_RELAYS);
    let picked = check(true, &NatStatus::Private, &[]);
    assert_eq!(picked.len(), AUTO_RELAYS);
    // One in use: the next in the same order.
    let more = check(true, &NatStatus::Private, &[picked[0].0]);
    assert_eq!(more.len(), AUTO_RELAYS - 1);
    assert_eq!(more[0].0, picked[1].0);
    let all: Vec<PeerId> = picked.iter().map(|(relay, _)| *relay).collect();
    assert!(check(true, &NatStatus::Private, &all).is_empty());
    // Never itself.
    let only_us = HashMap::from([(local, Multiaddr::empty())]);
    assert!(relay_check(local, false, &NatStatus::Private, &only_us, &[]).is_empty());
}

/// A relay says so in its addresses; b, which nobody can dial, learns of it (as from
/// the member list) and becomes reachable through it.
#[tokio::test]
async fn a_member_that_relays_is_picked() {
    let (r, _) = memory_node_with(true, Vec::new(), true);
    assert!(eventually(|| !r.local_addresses().is_empty()).await);
    let offer = r
        .local_addresses()
        .into_iter()
        .find(|address| address.ends_with("/p2p-circuit"))
        .expect("the relay offers itself");
    let (b, _) = memory_node_with(false, Vec::new(), false);
    assert_eq!(b.locate(&offer).as_deref(), Some(r.local_id()));
    assert!(
        eventually(|| b
            .local_addresses()
            .iter()
            .any(|address| address.contains("/p2p-circuit/")))
        .await,
        "b isn't reachable through r: {:?}",
        b.local_addresses()
    );
}

/// A peer on the LAN that speaks continuum becomes a seed once, at its address.
#[test]
fn public_ranges() {
    let public = |ip: &str| book::is_public(ip.parse().unwrap());
    assert!(public("203.0.114.5"));
    assert!(public("2a01:4f8::1"));
    for not in [
        "10.1.2.3",
        "100.64.0.1",
        "192.0.2.1",
        "198.18.0.1",
        "224.0.0.1",
        "255.255.255.255",
        "::1",
        "fd00::1",
        "fe80::1",
        "2001:db8::1",
        "64:ff9b::1",
        // Mycelium's overlay.
        "45f:fa21:4bdd::1",
    ] {
        assert!(!public(not), "{not}");
    }
}

/// An address where a dial reached another peer isn't kept for the one dialled, even
/// when it tells it again (identify) or the member list does.
#[test]
fn addresses_of_another_peer_are_forgotten() {
    let mut book = Book::default();
    let them = peer();
    let lan: Multiaddr = "/ip4/192.168.1.5/tcp/9661".parse().unwrap();
    let elsewhere: Multiaddr = "/ip4/203.0.114.9/tcp/9661".parse().unwrap();
    book.remember(them, lan.clone());
    book.remember(them, elsewhere.clone());
    book.not_theirs(them, lan.clone().with_p2p(them).unwrap());
    assert_eq!(book.addresses[&them], [elsewhere.clone()]);
    book.remember(them, lan.clone());
    book.identified(them, vec![lan, elsewhere.clone()]);
    assert_eq!(book.addresses[&them], [elsewhere]);
}

#[test]
fn lan_peers_become_seeds_once() {
    let mut book = Book::default();
    let lan = peer();
    let address: Multiaddr = "/ip4/192.168.1.5/tcp/9661".parse().unwrap();
    book.discover(lan); // No address known: nothing.
    assert!(book.discovered.is_empty());
    book.remember(lan, address.clone());
    book.discover(lan);
    book.discover(lan);
    assert_eq!(book.discovered, [address.with_p2p(lan).unwrap()]);
}

#[test]
fn named_relays() {
    let (local, r1, r2) = (peer(), peer(), peer());
    let at = |port: u16| -> Multiaddr { format!("/ip6/fd00::1/tcp/{port}").parse().unwrap() };
    let mut book = Book::default();
    // With a peer ID: known at once; without: to be found out, once.
    assert_eq!(
        book.name_relay(at(1).with_p2p(r1).unwrap(), Named::Configured, local),
        None
    );
    assert_eq!(
        book.name_relay(at(2), Named::Configured, local),
        Some(at(2))
    );
    assert_eq!(book.name_relay(at(2), Named::Configured, local), None);
    assert_eq!(
        book.name_relay(at(3).with_p2p(local).unwrap(), Named::Configured, local),
        None
    );
    assert!(book.relay_found(at(2), r2, local));
    assert_eq!(book.named(Named::Configured).len(), 2);
    // A relay named with its peer ID isn't found out again; one without is.
    assert_eq!(book.find_again(r1), None);
    assert_eq!(book.find_again(r2), Some(at(2)));
    assert!(!book.named(Named::Configured).contains_key(&r2));

    // Listed: configured wins; unlisted ones go, unless also configured.
    let r3 = peer();
    let listed = vec![
        at(1).with_p2p(r1).unwrap(),
        at(4).with_p2p(r3).unwrap(),
        at(5),
    ];
    assert_eq!(book.learn_listed(listed.clone(), local), Some(vec![at(5)]));
    assert_eq!(book.learn_listed(listed, local), None);
    assert_eq!(book.named_relays[&r1].named, Named::Configured);
    assert_eq!(book.named(Named::Listed).keys().collect::<Vec<_>>(), [&r3]);
    assert_eq!(book.learn_listed(Vec::new(), local), Some(Vec::new()));
    assert!(book.named(Named::Listed).is_empty());
    assert!(!book.finding.contains_key(&at(5)));
    assert!(!book.relay_found(at(5), peer(), local)); // No longer listed.
    assert_eq!(book.named(Named::Configured).len(), 1);
}

#[test]
fn known_addresses() {
    let mut book = Book::default();
    let (relay, us) = (peer(), peer());
    let direct: Multiaddr = "/ip6/fd00::1/tcp/9661".parse().unwrap();
    let relayed: Multiaddr = format!("/ip6/fd00::2/tcp/9661/p2p/{relay}/p2p-circuit/p2p/{us}")
        .parse()
        .unwrap();
    assert!(!book.knows(us, &[direct.clone()]));
    book.remember(us, relayed.clone());
    book.remember(us, direct.clone());
    // In any order, with or without the peer ID.
    assert!(book.knows(us, &[relayed.clone(), direct.clone()]));
    assert!(book.knows(us, &[direct.clone().with_p2p(us).unwrap()]));
    assert!(!book.knows(us, &["/ip6/fd00::3/tcp/9661".parse().unwrap()]));
}

/// A connected peer's identify says where it is: older direct addresses go, and others
/// aren't taken while it stays connected; those through relays and by name stay.
#[test]
fn identify_replaces_direct_addresses() {
    let mut book = Book::default();
    let (relay, them) = (peer(), peer());
    let old: Multiaddr = "/ip4/10.244.0.7/tcp/9661".parse().unwrap();
    let lan: Multiaddr = "/ip4/192.168.1.7/tcp/9661".parse().unwrap();
    let named: Multiaddr = "/dns4/node.example/tcp/9661".parse().unwrap();
    let relayed: Multiaddr = format!("/ip4/192.0.2.1/tcp/9661/p2p/{relay}/p2p-circuit")
        .parse()
        .unwrap();
    book.remember(them, old.clone());
    book.remember(them, named.clone());
    book.remember(them, relayed.clone());

    book.identified(them, vec![lan.clone().with_p2p(them).unwrap()]);
    assert!(book.knows(them, &[lan.clone(), named.clone(), relayed.clone()]));
    assert!(!book.knows(them, &[old.clone()]));
    // The member list, say, still has the old one.
    book.remember(them, old.clone());
    assert!(!book.knows(them, &[old.clone()]));

    // Disconnected (`driver`): taken again, it may have moved there.
    book.identified.remove(&them);
    book.remember(them, old.clone());
    assert!(book.knows(them, &[old]));
}

/// Of a peer with more addresses than are kept, those that matter stay: in the ranges
/// `libp2p-prefer` names first, then public ones, then through relays, then other
/// private ones.
#[test]
fn identified_addresses_preferred_first() {
    let mut book = Book {
        preferred: vec!["10.23.0.0/16".parse().unwrap()],
        ..Book::default()
    };
    let (relay, them) = (peer(), peer());
    let lan: Multiaddr = "/ip4/10.23.7.7/tcp/9661".parse().unwrap();
    // Outside the documentation ranges, which aren't public (`is_public`).
    let public: Multiaddr = "/ip4/203.0.114.7/tcp/9661".parse().unwrap();
    let relayed: Multiaddr = format!("/ip4/192.0.2.1/tcp/9661/p2p/{relay}/p2p-circuit")
        .parse()
        .unwrap();
    let bridges: Vec<Multiaddr> = (0..ADDRESSES_PER_PEER)
        .map(|i| format!("/ip4/172.17.{i}.1/tcp/9661").parse().unwrap())
        .collect();
    let mut listen = bridges.clone();
    listen.extend([relayed.clone(), public.clone(), lan.clone()]);

    book.identified(them, listen);
    let known = &book.addresses[&them];
    assert_eq!(known.len(), ADDRESSES_PER_PEER);
    assert_eq!(known[..3], [lan.clone(), public, relayed]);

    // Our own: those listened on in the preferred ranges published first.
    book.listening = vec![bridges[0].clone(), lan.clone()];
    assert_eq!(book.direct(), [lan, bridges[0].clone()]);
}

/// A peer's addresses in a range this node leaves out aren't kept: never dialled, they
/// would take a place, and be passed on.
#[test]
fn ignored_addresses_not_kept() {
    let mut book = Book {
        ignored: vec!["10.244.0.0/16".parse().unwrap()],
        ..Book::default()
    };
    let them = peer();
    let vpn: Multiaddr = "/ip4/10.244.3.2/tcp/9661".parse().unwrap();
    let lan: Multiaddr = "/ip4/10.23.1.144/tcp/9661".parse().unwrap();
    let relay = peer();
    // Through a relay there: dialling the relay is refused.
    let via_vpn: Multiaddr = format!("/ip4/10.244.0.1/tcp/9661/p2p/{relay}/p2p-circuit")
        .parse()
        .unwrap();
    let via_lan: Multiaddr = format!("/ip4/10.23.0.1/tcp/9661/p2p/{relay}/p2p-circuit")
        .parse()
        .unwrap();
    book.remember(them, vpn.clone());
    book.remember(them, via_vpn.clone());
    book.identified(them, vec![vpn.clone(), lan.clone(), via_lan.clone()]);
    assert!(book.knows(them, &[lan, via_lan]));
    assert!(!book.knows(them, &[vpn]));
    assert!(!book.knows(them, &[via_vpn]));
}

/// A node connected directly over the WLAN dials the ethernet address it later hears
/// of, once; and when an interface of ours comes, every address again, the one in use
/// too.
#[test]
fn better_paths_explored() {
    let mut book = Book::default();
    let them = peer();
    let wlan: Multiaddr = "/ip4/192.168.1.7/tcp/9661".parse().unwrap();
    let ethernet: Multiaddr = "/ip4/192.168.2.7/tcp/9661".parse().unwrap();
    book.continuum.insert(them);
    book.remember(them, wlan.clone());
    // Connected (inbound: from another port) over the WLAN.
    book.connections.insert(
        ConnectionId::new_unchecked(1),
        (them, false, "/ip4/192.168.1.7/tcp/40000".parse().unwrap()),
    );
    assert!(book.to_explore(them, false).is_empty());
    book.identified(them, vec![wlan.clone(), ethernet.clone()]);
    assert_eq!(book.to_explore(them, false), [ethernet.clone()]);
    assert!(book.to_explore(them, false).is_empty());
    // An interface of ours came (`driver`): all of them again.
    book.explored.clear();
    let mut again = book.to_explore(them, true);
    again.sort();
    let mut both = vec![wlan, ethernet];
    both.sort();
    assert_eq!(again, both);
}

/// Addresses without a route from here are left out, as if ignored: not kept, nor
/// dialled; and own public ones of a family without a route to the internet (by
/// `--libp2p-internet-hosts`) aren't published. The routing table is asked for real only by `has_route`.
#[test]
fn unroutable_addresses_left_out() {
    assert!(has_route("127.0.0.1".parse().unwrap()));
    // No IPv6 here.
    let mut book = Book {
        routes: Routes::with_probe(|ip| ip.is_ipv4()),
        listening: vec![
            "/ip4/192.168.1.2/tcp/9661".parse().unwrap(),
            "/ip6/2a0a:4cc0::2/tcp/9661".parse().unwrap(),
        ],
        ..Book::default()
    };
    let them = peer();
    let v4: Multiaddr = "/ip4/192.168.1.7/tcp/9661".parse().unwrap();
    let v6: Multiaddr = "/ip6/2a0a:4cc0::7/tcp/9661".parse().unwrap();
    book.identified(them, vec![v4.clone(), v6.clone()]);
    assert!(book.knows(them, &[v4]));
    assert!(!book.knows(them, &[v6]));
    // Nobody asked whether there's a route to the internet: everything is published.
    assert_eq!(book.direct().len(), 2);
    book.routes.internet(&[
        "1.1.1.1".parse().unwrap(),
        "2606:4700:4700::1111".parse().unwrap(),
    ]);
    assert_eq!(
        book.direct(),
        ["/ip4/192.168.1.2/tcp/9661".parse::<Multiaddr>().unwrap()]
    );
}

/// `--libp2p-internet-hosts` takes addresses as they are (names are resolved).
#[tokio::test]
async fn internet_hosts_resolved() {
    let resolved = resolve_internet_hosts(&["1.1.1.1".to_owned(), "::1".to_owned()]).await;
    assert_eq!(
        resolved,
        [
            "1.1.1.1".parse::<std::net::IpAddr>().unwrap(),
            "::1".parse().unwrap()
        ]
    );
}

/// Peers nobody mentions any more are forgotten, but connected ones.
#[test]
fn stale_peers_are_forgotten() {
    let (old, current, connected) = (peer(), peer(), peer());
    let address: Multiaddr = "/ip6/fd00::1/tcp/9661".parse().unwrap();
    let mut book = Book::default();
    for peer in [old, current, connected] {
        book.remember(peer, address.clone());
    }
    book.relays.insert(old, address.clone());
    book.continuum.insert(old);
    // From after the mentions above (taken at `Instant::now()`), however fast or
    // slow the clock moved meanwhile.
    let later = tokio::time::Instant::now() + STALE_AFTER;
    // `current` was mentioned again since.
    book.seen.insert(current, later);
    assert_eq!(book.prune(later, |peer| *peer == connected), 1);
    assert!(!book.addresses.contains_key(&old));
    assert!(!book.relays.contains_key(&old) && !book.continuum.contains(&old));
    assert!(book.addresses.contains_key(&current));
    assert!(book.addresses.contains_key(&connected));
    assert_eq!(book.prune(later, |_| false), 1); // `connected`, once it isn't.
}

#[test]
fn relay_failures_are_short() {
    assert_eq!(
        relay_failure_text(
            "Custom { kind: Other, error: Other(Left(Left(Left(Reservation(Protocol(NoAddressesInReservation)))))) }",
            "x"
        ),
        "the relay gave no address to be reached through it"
    );
    assert_eq!(
        relay_failure_text(
            "Codec(Error(Custom { kind: Other, error: Failed }))",
            "Failed"
        ),
        "the relay refused the circuit relay protocol (is it a relay?)"
    );
    assert_eq!(
        relay_failure_text("Io(…)", "connection reset"),
        "connection reset"
    );
    assert_eq!(relay_failure(&Ok(())), "closed");
}

/// Only refusals pause a relay; a slot that ended otherwise is asked for again.
#[test]
fn only_refusals_pause_relays() {
    let refused = [
        ("Reservation(Protocol(NoAddressesInReservation))", "x"),
        ("Reservation(Status(ResourceLimitExceeded))", "x"),
        ("Reservation(Status(PermissionDenied))", "x"),
        ("Denied { cause: \"connection gated\" }", "x"),
        ("Upgrade(UnsupportedProtocol)", "x"),
        ("Dial(WrongPeerId { obtained: … })", "x"),
        (
            "Codec(Error(Custom { kind: Other, error: Failed }))",
            "Failed",
        ),
    ];
    for (debug, innermost) in refused {
        assert!(relay_refused(debug, innermost), "{debug}");
    }
    let not = [
        ("Io(Custom { kind: Other, error: Timeout })", "Timeout"),
        (
            "Io(Custom { kind: Other, error: Right(Closed) })",
            "connection is closed",
        ),
        ("Io(…)", "connection reset"),
    ];
    for (debug, innermost) in not {
        assert!(!relay_refused(debug, innermost), "{debug}");
    }
}

#[test]
fn listen_addresses_offered_for_hole_punching() {
    let parse = |text: &str| text.parse::<Multiaddr>().unwrap();
    assert!(offered(&parse("/ip4/192.168.1.2/udp/9661/quic-v1")));
    assert!(offered(&parse("/ip6/fd00::2/tcp/9661")));
    assert!(!offered(&parse("/ip4/127.0.0.1/udp/9661/quic-v1")));
    assert!(!offered(&parse("/ip6/fe80::1/tcp/9661")));
    let relay = PeerId::random();
    assert!(!offered(&parse(&format!(
        "/ip4/192.168.1.2/tcp/9661/p2p/{relay}/p2p-circuit"
    ))));
}

/// libp2p's nested dial errors come out as each address's cause, once.
#[test]
fn dial_errors_are_short() {
    use ::libp2p::core::transport::TransportError;
    let peer = peer();
    let address: Multiaddr = format!("/ip6/fd00::1/tcp/9661/p2p/{peer}").parse().unwrap();
    let nested = std::io::Error::other(
        "Multiple dial errors occurred:\n - Network is unreachable (os error 101): Multiple dial errors occurred:\n - Network is unreachable (os error 101): Network is unreachable (os error 101)",
    );
    let error =
        ::libp2p::swarm::DialError::Transport(vec![(address, TransportError::Other(nested))]);
    assert_eq!(
        describe_dial_error(&error),
        "/ip6/fd00::1/tcp/9661: Network is unreachable (os error 101)"
    );
    let refused = ::libp2p::swarm::DialError::Transport(vec![(
        "/ip4/10.0.0.1/tcp/1".parse().unwrap(),
        TransportError::Other(std::io::Error::from(std::io::ErrorKind::ConnectionRefused)),
    )]);
    assert_eq!(
        describe_dial_error(&refused),
        "/ip4/10.0.0.1/tcp/1: connection refused"
    );
}

/// A relay doesn't dial the nodes it relays for through itself.
#[test]
fn no_circuits_through_ourselves() {
    let (us, other_relay, target) = (peer(), peer(), peer());
    let through = |relay: PeerId| -> Multiaddr {
        format!("/ip6/fd00::1/tcp/9661/p2p/{relay}/p2p-circuit")
            .parse()
            .unwrap()
    };
    let direct: Multiaddr = "/ip6/fd00::2/tcp/9661".parse().unwrap();
    assert_eq!(relay_of(&through(us)), Some(us));
    assert_eq!(relay_of(&direct), None);
    assert_eq!(relay_of(&direct.clone().with_p2p(target).unwrap()), None);
    assert_eq!(
        dialable(&[through(us), through(other_relay), direct.clone()], us),
        [through(other_relay), direct]
    );
}

#[tokio::test]
async fn unknown_peers_are_refused() {
    let (b, _) = memory_node();
    let stranger = Keypair::generate_ed25519()
        .public()
        .to_peer_id()
        .to_string();
    assert!(b.send_message(&stranger, &status()).await.is_err());
    assert!(b.resolve_seed(&stranger).await.is_err());
    assert_eq!(b.connection(&stranger), None);
}

#[test]
fn a_direct_connection_wins_over_a_relayed_one() {
    let mut book = Book::default();
    let (relayed_only, both) = (peer(), peer());
    let mut id = 0;
    let mut connect = |book: &mut Book, peer: PeerId, relayed: bool| {
        id += 1;
        let address: Multiaddr = format!("/ip4/10.0.0.{id}/tcp/1").parse().unwrap();
        book.connections
            .insert(ConnectionId::new_unchecked(id), (peer, relayed, address));
    };
    connect(&mut book, relayed_only, true);
    connect(&mut book, both, true);
    connect(&mut book, both, false);
    assert_eq!(book.connection(&relayed_only), Some(Connection::Relayed));
    assert_eq!(book.connection(&both), Some(Connection::Direct));
    assert_eq!(book.connection(&peer()), None);
    assert_eq!(
        book.location(&both).map(|address| address.to_string()),
        Some("/ip4/10.0.0.3/tcp/1".to_owned())
    );
    let connections = book.stats().connections;
    assert_eq!((connections.direct, connections.relayed), (1, 2));
    // Of several, the oldest direct one stays; a relayed one only without a direct.
    connect(&mut book, both, false);
    assert_eq!(
        book.duplicates(&both),
        [
            ConnectionId::new_unchecked(4),
            ConnectionId::new_unchecked(2)
        ]
    );
    assert!(book.duplicates(&relayed_only).is_empty());
    // Also where one of them carries a slot, ours or the peer's: messages would go over
    // either.
    book.slots.insert(both);
    assert_eq!(book.duplicates(&both).len(), 2);
    book.slots.clear();
    book.slots_here.insert(both);
    assert_eq!(book.duplicates(&both).len(), 2);
    book.slots_here.clear();
    // Once measured, the fastest direct one stays, however young.
    book.rtts
        .insert(ConnectionId::new_unchecked(3), Duration::from_millis(40));
    book.rtts
        .insert(ConnectionId::new_unchecked(4), Duration::from_millis(2));
    assert_eq!(
        book.duplicates(&both),
        [
            ConnectionId::new_unchecked(3),
            ConnectionId::new_unchecked(2)
        ]
    );
    assert_eq!(
        book.location(&both).map(|address| address.to_string()),
        Some("/ip4/10.0.0.4/tcp/1".to_owned())
    );
}