netring 0.29.0

//! Integration tests for the flowscope 0.3 pass-through plumbing
//! (plan 19).
//!
//! Covers:
//!
//! 1. `with_idle_timeout_fn` on `FlowStream` is preserved when
//!    converted to `SessionStream` / `DatagramStream` (i.e. the
//!    tracker is moved over, not rebuilt).
//! 2. `with_monotonic_timestamps` is preserved across the same
//!    conversions.
//! 3. `snapshot_flow_stats()` is callable on all three stream
//!    types (signature smoke test).
//!
//! The tests build the stream chain but never poll it — no
//! packets are captured, so behavior is independent of network
//! state. Same shape as `flow_stream_config.rs`.

#![cfg(all(
    feature = "integration-tests",
    feature = "flow",
    feature = "tokio",
    feature = "parse"
))]

mod helpers;

use std::time::Duration;

use flowscope::{
    DatagramParser, DatagramParserFactory, SessionParser, SessionParserFactory, Timestamp,
};
use netring::flow::extract::FiveTuple;
use netring::{AsyncCapture, CaptureBuilder};

#[derive(Default, Clone)]
struct StubSessionParser;

impl SessionParser for StubSessionParser {
    type Message = ();
    fn feed_initiator(&mut self, _: &[u8], _ts: Timestamp, _out: &mut Vec<()>) {}
    fn feed_responder(&mut self, _: &[u8], _ts: Timestamp, _out: &mut Vec<()>) {}
}

#[derive(Default, Clone)]
struct StubSessionFactory;

impl<K> SessionParserFactory<K> for StubSessionFactory {
    type Parser = StubSessionParser;
    fn new_parser(&mut self, _key: &K) -> StubSessionParser {
        StubSessionParser
    }
}

#[derive(Default, Clone)]
struct StubDatagramParser;

impl DatagramParser for StubDatagramParser {
    type Message = ();
    fn parse(&mut self, _: &[u8], _: flowscope::FlowSide, _ts: Timestamp, _out: &mut Vec<()>) {}
}

#[derive(Default, Clone)]
struct StubDatagramFactory;

impl<K> DatagramParserFactory<K> for StubDatagramFactory {
    type Parser = StubDatagramParser;
    fn new_parser(&mut self, _key: &K) -> StubDatagramParser {
        StubDatagramParser
    }
}

fn build_async_capture() -> AsyncCapture<netring::Capture> {
    let rx = CaptureBuilder::default()
        .interface(helpers::LOOPBACK)
        .block_timeout_ms(10)
        .build()
        .expect("build rx");
    AsyncCapture::new(rx).expect("wrap in AsyncFd")
}

#[tokio::test]
async fn flow_stream_idle_timeout_fn_chains() {
    // Smoke test — the closure compiles, the builder accepts it,
    // and the resulting stream is constructible. No polling.
    let _stream = build_async_capture()
        .flow_stream(FiveTuple::bidirectional())
        .with_idle_timeout_fn(|key, _l4| {
            if key.either_port(53) {
                Some(Duration::from_secs(5))
            } else {
                None
            }
        });
}

#[tokio::test]
async fn session_stream_idle_timeout_fn_chains() {
    let _stream = build_async_capture()
        .flow_stream(FiveTuple::bidirectional())
        .session_stream(StubSessionFactory)
        .with_idle_timeout_fn(|_, _| Some(Duration::from_secs(3)));
}

#[tokio::test]
async fn datagram_stream_idle_timeout_fn_chains() {
    let _stream = build_async_capture()
        .flow_stream(FiveTuple::bidirectional())
        .datagram_stream(StubDatagramFactory)
        .with_idle_timeout_fn(|_, _| Some(Duration::from_secs(3)));
}

#[tokio::test]
async fn monotonic_chains_across_conversions() {
    // The bool just toggles state; preservation across conversion
    // is a netring-side property tested via the builder.
    let _session = build_async_capture()
        .flow_stream(FiveTuple::bidirectional())
        .with_monotonic_timestamps(true)
        .session_stream(StubSessionFactory);

    let _datagram = build_async_capture()
        .flow_stream(FiveTuple::bidirectional())
        .with_monotonic_timestamps(true)
        .datagram_stream(StubDatagramFactory);

    let _ds_only = build_async_capture()
        .flow_stream(FiveTuple::bidirectional())
        .datagram_stream(StubDatagramFactory)
        .with_monotonic_timestamps(true);
}

#[tokio::test]
async fn snapshot_flow_stats_callable_on_all_three() {
    let flow = build_async_capture().flow_stream(FiveTuple::bidirectional());
    // Newly-built tracker has no flows.
    assert_eq!(flow.snapshot_flow_stats().count(), 0);

    let session = build_async_capture()
        .flow_stream(FiveTuple::bidirectional())
        .session_stream(StubSessionFactory);
    assert_eq!(session.snapshot_flow_stats().count(), 0);

    let datagram = build_async_capture()
        .flow_stream(FiveTuple::bidirectional())
        .datagram_stream(StubDatagramFactory);
    assert_eq!(datagram.snapshot_flow_stats().count(), 0);
}