ferrix/crates/integration-tests/tests/http_api.rs

//! Drives the HTTP API against a real server with a real BGP peer attached.

use std::net::Ipv4Addr;
use std::time::Duration;

use bgp_integration_tests::test_peer::TestPeer;
use bgp_integration_tests::test_server::TestServer;
use bgp_packet::message::{
    AsPathAttribute, CommunitiesPathAttribute, LargeCommunity, Message, NextHopPathAttribute,
    OriginPathAttribute, PathAttribute, UpdateMessage,
};
use bgp_primitives::ip_prefix::IpPrefix;
use bgp_rib::{LocalRoute, LocalRouteAttrs};
use eyre::Result;
use serde_json::{Value, json};

const LOCAL_ASN: u32 = 65000;
const PEER_ASN: u32 = 65001;
const TIMEOUT: Duration = Duration::from_secs(5);

fn prefix(s: &str) -> IpPrefix {
    IpPrefix::try_from(s).unwrap()
}

async fn established_peer(server: &TestServer) -> Result<TestPeer> {
    let mut peer = TestPeer::connect(server.addr).await?;
    peer.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 1), 90).await?;
    tokio::time::timeout(TIMEOUT, async {
        while server.registry.list().is_empty() {
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    })
    .await?;
    Ok(peer)
}

async fn get_json(client: &reqwest::Client, url: &str) -> Result<Value> {
    Ok(client.get(url).send().await?.error_for_status()?.json().await?)
}

#[tokio::test]
async fn test_status_and_peers() -> Result<()> {
    let server = TestServer::start(LOCAL_ASN).await?;
    let (base, _http) = server.serve_http().await?;
    let client = reqwest::Client::new();

    let config = get_json(&client, &format!("{base}/api/config")).await?;
    assert_eq!(config["path"], "test.textpb");
    assert_eq!(config["text"], format!("local_asn: {LOCAL_ASN}\n"));

    let status = get_json(&client, &format!("{base}/api/status")).await?;
    assert_eq!(status["local_asn"], LOCAL_ASN);
    assert_eq!(status["router_id"], "127.0.0.1");
    assert_eq!(status["sessions"], 0);
    assert!(status["started_at_unix"].as_u64().unwrap() > 1_700_000_000);
    assert!(!status["hostname"].as_str().unwrap().is_empty());

    let _peer = established_peer(&server).await?;

    let status = get_json(&client, &format!("{base}/api/status")).await?;
    assert_eq!(status["sessions"], 1);

    let peers = get_json(&client, &format!("{base}/api/peers")).await?;
    assert_eq!(peers.as_array().unwrap().len(), 1);
    assert_eq!(peers[0]["peer_asn"], PEER_ASN);
    assert_eq!(peers[0]["local_addr"], server.addr.to_string());
    assert_eq!(peers[0]["four_octet_asn"], true);
    Ok(())
}

/// Originated routes are listed with their attributes as text.
#[tokio::test]
async fn test_originated_routes_are_listed() -> Result<()> {
    let attrs = LocalRouteAttrs {
        prepend: 1,
        med: Some(10),
        communities: vec![(65000, 100), (65535, 65281)],
        large_communities: vec![LargeCommunity { global_admin: 65000, data_1: 1, data_2: 2 }],
        ..Default::default()
    };
    let server = TestServer::start_with(LOCAL_ASN, |c| {
        c.originate = vec![LocalRoute::new(prefix("10.100.1.0/24"), attrs)];
    })
    .await?;
    let (base, _http) = server.serve_http().await?;
    let client = reqwest::Client::new();

    let local = get_json(&client, &format!("{base}/api/routes/local")).await?;
    assert_eq!(local.as_array().unwrap().len(), 1);
    assert_eq!(local[0]["prefix"], "10.100.1.0/24");
    assert_eq!(local[0]["origin"], "IGP");
    assert_eq!(local[0]["prepend"], 1);
    assert_eq!(local[0]["med"], 10);
    assert_eq!(local[0]["communities"], json!(["65000:100", "65535:65281"]));
    assert_eq!(local[0]["large_communities"], json!(["65000:1:2"]));
    assert_eq!(get_json(&client, &format!("{base}/api/status")).await?["local_routes"], 1);
    Ok(())
}

/// The API only reports. Nothing reachable over HTTP changes what the
/// router announces.
#[tokio::test]
async fn test_api_is_read_only() -> Result<()> {
    let server = TestServer::start(LOCAL_ASN).await?;
    let (base, _http) = server.serve_http().await?;
    let client = reqwest::Client::new();

    let body = json!({ "prefix": "10.100.1.0/24" });
    for path in ["/api/routes/local", "/api/routes/received", "/api/peers", "/api/status"] {
        let post = client.post(format!("{base}{path}")).json(&body).send().await?;
        assert_eq!(post.status(), 405, "POST {path}");
        let delete = client.delete(format!("{base}{path}")).json(&body).send().await?;
        assert_eq!(delete.status(), 405, "DELETE {path}");
    }
    assert!(server.local_rib.snapshot().await.is_empty());
    Ok(())
}

/// Routes the peer sends us show up in the received table with a summary.
#[tokio::test]
async fn test_received_routes_summary() -> Result<()> {
    let server = TestServer::start(LOCAL_ASN).await?;
    let (base, _http) = server.serve_http().await?;
    let client = reqwest::Client::new();
    let mut peer = established_peer(&server).await?;

    peer.send(&Message::Update(UpdateMessage {
        withdrawn_routes: vec![],
        path_attributes: vec![
            PathAttribute::Origin(OriginPathAttribute::IGP),
            AsPathAttribute::from_asns(vec![PEER_ASN, 65002]),
            PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(10, 0, 0, 1))),
            PathAttribute::Communities(CommunitiesPathAttribute(vec![(65001, 7)])),
        ],
        announced_routes: vec![prefix("203.0.113.0/24")],
    }))
    .await?;

    let received = tokio::time::timeout(TIMEOUT, async {
        loop {
            let v = get_json(&client, &format!("{base}/api/routes/received")).await?;
            if !v.as_array().unwrap().is_empty() {
                return Ok::<_, eyre::Error>(v);
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
    })
    .await??;
    assert_eq!(received[0]["prefix"], "203.0.113.0/24");
    assert_eq!(received[0]["peer_asn"], PEER_ASN);
    assert_eq!(received[0]["summary"]["as_path"], json!([PEER_ASN, 65002]));
    assert_eq!(received[0]["summary"]["next_hop"], "10.0.0.1");
    assert_eq!(received[0]["summary"]["communities"], json!(["65001:7"]));
    Ok(())
}

/// The SSE stream reports session and route changes as they happen.
#[tokio::test]
async fn test_event_stream() -> Result<()> {
    let server = TestServer::start(LOCAL_ASN).await?;
    let (base, _http) = server.serve_http().await?;
    let client = reqwest::Client::new();

    let mut resp = client.get(format!("{base}/api/events")).send().await?.error_for_status()?;
    assert!(resp.headers()["content-type"].to_str()?.starts_with("text/event-stream"));

    let _peer = established_peer(&server).await?;
    server
        .local_rib
        .announce(LocalRoute::new(prefix("10.100.1.0/24"), Default::default()))
        .await?;

    // Collect SSE frames until we have seen both events.
    let mut buf = String::new();
    let mut seen_up = false;
    let mut seen_announce = false;
    tokio::time::timeout(TIMEOUT, async {
        while !(seen_up && seen_announce) {
            let chunk = resp.chunk().await?.ok_or_else(|| eyre::eyre!("stream ended"))?;
            buf.push_str(std::str::from_utf8(&chunk)?);
            for line in buf.lines().filter_map(|l| l.strip_prefix("data:")) {
                let ev: Value = serde_json::from_str(line.trim())?;
                match ev["type"].as_str() {
                    Some("session_up") if ev["peer"]["peer_asn"] == PEER_ASN => seen_up = true,
                    Some("local_announced") if ev["route"]["prefix"] == "10.100.1.0/24" => seen_announce = true,
                    _ => {}
                }
            }
        }
        Ok::<_, eyre::Error>(())
    })
    .await??;
    Ok(())
}

#[tokio::test]
async fn test_statusz_page_is_served() -> Result<()> {
    let server = TestServer::start(LOCAL_ASN).await?;
    let (base, _http) = server.serve_http().await?;
    let client = reqwest::Client::new();

    let resp = client.get(format!("{base}/statusz")).send().await?;
    assert_eq!(resp.status(), 200);
    assert!(resp.headers()["content-type"].to_str()?.starts_with("text/html"));
    let body = resp.text().await?;
    assert!(body.contains("<title>Status for ferrix_bgp</title>"));
    assert!(body.contains("/api/events"));

    // The root redirects to the page.
    let no_redirect = reqwest::Client::builder().redirect(reqwest::redirect::Policy::none()).build()?;
    let resp = no_redirect.get(format!("{base}/")).send().await?;
    assert_eq!(resp.status(), 307);
    assert_eq!(resp.headers()["location"], "/statusz");
    Ok(())
}

#[tokio::test]
async fn test_metricz() -> Result<()> {
    let server = TestServer::start(LOCAL_ASN).await?;
    let (base, _http) = server.serve_http().await?;
    let client = reqwest::Client::new();
    let _peer = established_peer(&server).await?;

    let text = client.get(format!("{base}/metricz")).send().await?.error_for_status()?.text().await?;
    // Metrics are process-global, so other tests in this binary contribute
    // too; assert on presence and a lower bound rather than exact values.
    let metric = |name: &str| -> Option<i64> {
        text.lines()
            .find(|l| l.starts_with(name) && l[name.len()..].starts_with(' '))
            .and_then(|l| l.rsplit(' ').next()?.parse().ok())
    };
    assert!(metric("bgp_sessions_established").unwrap_or(0) >= 1, "{text}");
    assert!(metric("bgp_messages_sent_total{type=\"open\"}").unwrap_or(0) >= 1, "{text}");
    assert!(metric("bgp_messages_received_total{type=\"keepalive\"}").unwrap_or(0) >= 1, "{text}");
    assert!(metric("bgp_local_routes").is_some(), "{text}");
    assert!(metric("bgp_adj_rib_in_routes").is_some(), "{text}");
    Ok(())
}