spate-test 0.3.0

//! The address the admin server bound, as the start reports it.
//!
//! `admin.listen` accepts a port of `0`, and the kernel then picks the port, so
//! the configuration file does not say where the probes and the exposition are
//! served. This asserts that the start names the address it took, and that the
//! address on that line is the one answering.
//!
//! One test in its own binary, because the capture below is a process-wide
//! global subscriber and `cargo test` shares one process across a binary.

use spate_core::config::PipelineConfig;
use spate_core::ops::chain_owned;
use spate_core::pipeline::{Pipeline, RuntimeOptions};
use spate_core::sink::KeyHashRouter;
use spate_test::{BytesPassthrough, LogCapture, TestEncoder, capture_sink, memory_source};
use spate_test_support::http;
use std::net::SocketAddr;
use std::time::Duration;

const DEADLINE: Duration = Duration::from_secs(10);

// Port 0 is the case under test. `exporter: none` leaves `/metrics` a 404 and
// installs no process-global recorder; the probes do not depend on one.
const CONFIG: &str = r#"
pipeline: { name: admin-addr-test, threads: 1, io_threads: 1 }
admin: { listen: "127.0.0.1:0" }
metrics: { exporter: none }
source: { memory: {} }
sink: { capture: {} }
"#;

/// The `addr` field of the line announcing the bound admin address, once one
/// has been written.
///
/// Reading the field rather than testing the line for the absence of `:0`: a
/// formatter that stopped rendering fields this way produces that absence too,
/// and such an assertion would pass having checked nothing.
fn logged_addr(capture: &LogCapture) -> Option<SocketAddr> {
    capture
        .lines()
        .iter()
        .find(|l| l.contains("admin server listening"))
        .map(|l| {
            l.split_whitespace()
                .find_map(|f| f.strip_prefix("addr="))
                .unwrap_or_else(|| panic!("no `addr` field to read on: {l}"))
                .parse()
                .expect("addr is a socket address")
        })
}

#[test]
fn the_bound_admin_address_is_logged_and_serves() {
    let capture = LogCapture::new();
    // `info` is the level a deployment runs at, and the level
    // `spate_core::telemetry` assigns to a startup milestone.
    tracing_subscriber::fmt()
        .with_writer(capture.clone())
        .with_max_level(tracing_subscriber::filter::LevelFilter::INFO)
        .without_time()
        .init();

    let (source, _handle) = memory_source();
    let (sink, _script) = capture_sink(1, 1);

    let runtime = Pipeline::from_config(PipelineConfig::from_str(CONFIG).expect("config"))
        .expect("builder")
        .sink(sink)
        .expect("sink")
        .chains(|ctx| {
            let chunk_cfg = ctx.chunk();
            chain_owned::<Vec<u8>, _>(BytesPassthrough)
                .with_metrics(ctx.pipeline, "main")
                .sink(
                    TestEncoder,
                    KeyHashRouter,
                    chunk_cfg,
                    ctx.queues,
                    ctx.budget,
                )
                .build()
        })
        .runtime_options(RuntimeOptions {
            handle_signals: false,
            ..RuntimeOptions::default()
        })
        .into_runtime(source)
        .expect("into_runtime");

    let shutdown = runtime.shutdown_handle();
    let join = std::thread::spawn(move || runtime.run());

    let addr = capture.wait_for(DEADLINE, "the admin server to announce its address", || {
        logged_addr(&capture)
    });
    assert_ne!(
        addr.port(),
        0,
        "the line must carry the port the kernel picked, not the one asked for"
    );

    // The address on the line is the one serving. The wait is for the endpoint
    // to answer at all, which needs the spawned server to reach its accept
    // loop; what it answers is asserted once, so a probe reporting 503 fails
    // here rather than being polled past.
    let (status, body) =
        capture.wait_for(DEADLINE, "/healthz to answer at the logged address", || {
            http(addr, "GET", "/healthz").ok()
        });
    assert_eq!(status, 200, "the logged address serves the probes");
    assert_eq!(body, "ok");

    shutdown.trigger();
    let report = join.join().expect("join").expect("run");
    assert_eq!(report.exit_code(), 0, "clean drain");
}