tests: add test_basic

This is very basic and just meant for verifying test helpers.

Cargo.toml

@@ -2,6 +2,7 @@
 name = "flow"
 version = "0.2.0"
 edition = "2021"
+default-run = "flow"
 
 [[bin]]
 name = "DONTSHIPIT-flow-gen"
@@ -17,6 +18,7 @@ clap = { version = "4.5.20", features = ["derive"] }
 [dependencies]
 anstyle = "1.0.8"
 anyhow = "1.0.86"
+async-tempfile = "0.6.0"
 clap = { workspace = true }
 clap-verbosity-flag = "3.0.0"
 clap_complete = "4.5.38"
@@ -51,6 +53,7 @@ tokio = { version = "1.38.0", features = [
   "sync",
   "macros",
   "io-util",
+  "io-std",
   "signal",
 ] }
 tokio-util = "0.7.11"

src/integration_tests/config/basic.bird.conf

@@ -0,0 +1,43 @@
+router id 10.234.56.78;
+
+flow4 table myflow4;
+flow6 table myflow6;
+
+protocol static f4 {
+  flow4 { table myflow4; };
+
+  route flow4 {
+    dst 10.0.0.0/8;
+    length > 1024;
+  } {
+    bgp_ext_community.add((unknown 0x810c, 1.1.1.1, 0));
+  };
+}
+
+protocol static f6 {
+  flow6 { table myflow6; };
+
+  route flow6 {
+    dst fec0:1122:3344:5566:7788:99aa:bbcc:ddee/128;
+    tcp flags 0x03/0x0f && !0/0xff || 0x33/0x33;
+    dport = 6000;
+    fragment !is_fragment || !first_fragment;
+  } {
+    bgp_community.add((65001, 11451));
+    bgp_ext_community.add((generic, 0x080060000, 0));
+    bgp_ext_community.add((ro, 65002, 1919));
+  };
+}
+
+protocol bgp flow_test {
+  debug all;
+
+  local as 65000;
+  neighbor ::1 port 1179 as 65000;
+  multihop;
+
+  ipv4 { import none; export all; };
+  ipv6 { import none; export all; };
+  flow4 { table myflow4; import none; export all; };
+  flow6 { table myflow6; import none; export all; };
+}

src/integration_tests/helpers/bird.rs

@@ -1,9 +1,12 @@
 use anyhow::bail;
+use async_tempfile::{TempDir, TempFile};
 use std::borrow::Cow;
 use std::ffi::OsStr;
-use std::process::{Command, Stdio};
+use std::process::Stdio;
 use std::sync::LazyLock;
 use std::{env, io};
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::process::{Child, Command};
 use version_compare::compare_to;
 
 static BIRD_PATH: LazyLock<Cow<'static, OsStr>> = LazyLock::new(|| {
@@ -13,7 +16,10 @@ static BIRD_PATH: LazyLock<Cow<'static, OsStr>> = LazyLock::new(|| {
 });
 
 static BIRD_VERSION: LazyLock<Result<Option<String>, String>> = LazyLock::new(|| {
-  let output = Command::new(&*BIRD_PATH).arg("--version").stdin(Stdio::null()).output();
+  let output = std::process::Command::new(&*BIRD_PATH)
+    .arg("--version")
+    .stdin(Stdio::null())
+    .output();
   let mut stderr = match output {
     Ok(output) => output.stderr,
     Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(None),
@@ -111,3 +117,33 @@ fn check_bird_2_16() -> anyhow::Result<()> {
       for more information.",
   )
 }
+
+pub async fn run_bird(config: &str) -> anyhow::Result<(Child, (TempFile, TempDir))> {
+  let mut config_file = TempFile::new().await?;
+  config_file.write_all(config.as_bytes()).await?;
+  config_file.flush().await?;
+
+  let sock_dir = TempDir::new().await?;
+  let sock_path = sock_dir.join("bird.sock");
+
+  let mut bird = Command::new(&*BIRD_PATH)
+    .arg("-d")
+    .args(["-c".as_ref(), config_file.file_path().as_os_str()])
+    .args(["-s".as_ref(), sock_path.as_os_str()])
+    .stdin(Stdio::null())
+    .stdout(Stdio::piped())
+    .stderr(Stdio::piped())
+    .spawn()?;
+
+  let mut bird_stderr = BufReader::new(bird.stderr.take().unwrap());
+  tokio::spawn(async move {
+    let mut buf = String::new();
+    while bird_stderr.read_line(&mut buf).await? != 0 {
+      eprint!("{buf}");
+      buf.clear();
+    }
+    anyhow::Ok(())
+  });
+
+  Ok((bird, (config_file, sock_dir)))
+}

src/integration_tests/helpers/cli.rs

@@ -0,0 +1,20 @@
+use crate::args::Cli;
+use crate::cli_entry;
+use std::future::Future;
+use std::process::ExitCode;
+use tokio::task::{JoinHandle, LocalSet};
+
+pub type CliChild = JoinHandle<anyhow::Result<ExitCode>>;
+
+pub async fn run_cli<F, Fut, R>(options: Cli, f: F) -> anyhow::Result<R>
+where
+  F: FnOnce(CliChild, &LocalSet) -> Fut,
+  Fut: Future<Output = anyhow::Result<R>>,
+{
+  let ls = LocalSet::new();
+  let cli = ls.spawn_local(async {
+    let exit_code = cli_entry(options).await;
+    anyhow::Ok(exit_code)
+  });
+  ls.run_until(f(cli, &ls)).await
+}

src/integration_tests/helpers/kernel.rs

@@ -1,6 +1,26 @@
 use anyhow::bail;
 use nix::unistd::Uid;
 
+#[cfg(rtnetlink_supported)]
+pub async fn ensure_loopback_up() -> anyhow::Result<()> {
+  use rtnetlink::{LinkMessageBuilder, LinkUnspec};
+
+  if !Uid::effective().is_root() {
+    return Ok(());
+  }
+  let (conn, handle, _) = rtnetlink::new_connection()?;
+  tokio::spawn(conn);
+  handle
+    .link()
+    .set(LinkMessageBuilder::<LinkUnspec>::new().index(1).up().build())
+    .execute()
+    .await?;
+  Ok(())
+}
+
+#[cfg(not(rtnetlink_supported))]
+pub async fn ensure_loopback_up() -> anyhow::Result<()> {}
+
 #[test]
 fn check_root() -> anyhow::Result<()> {
   if !Uid::effective().is_root() {

src/integration_tests/helpers/mod.rs

@@ -1,2 +1,3 @@
 pub mod bird;
+pub mod cli;
 pub mod kernel;

src/integration_tests/mod.rs

@@ -2,3 +2,36 @@
 //! tests.
 
 mod helpers;
+
+use crate::args::Cli;
+use clap::Parser;
+use helpers::bird::{ensure_bird_ver_ge, run_bird};
+use helpers::cli::run_cli;
+use helpers::kernel::ensure_loopback_up;
+use std::ffi::OsString;
+use std::time::Duration;
+use tokio::time::sleep;
+
+#[tokio::test]
+async fn test_basic() -> anyhow::Result<()> {
+  ensure_bird_ver_ge!("2");
+  ensure_loopback_up().await?;
+
+  let (mut bird, (_f, temp_dir)) = run_bird(include_str!("config/basic.bird.conf")).await?;
+  let temp_dir_path = temp_dir.as_os_str().into();
+  let cli_opt = Cli::try_parse_from(
+    ["flow", "run", "-v", "--dry-run", "--bind=[::1]:1179", "--run-dir"]
+      .into_iter()
+      .map(OsString::from)
+      .chain(Some(temp_dir_path)),
+  )?;
+
+  // TODO: implement events when cfg(test) in CLI
+  let fut = run_cli(cli_opt, |_cli, _ls| async {
+    sleep(Duration::from_secs(7)).await;
+    bird.kill().await?;
+    Ok(())
+  });
+
+  fut.await
+}

src/main.rs

@@ -15,8 +15,6 @@ use args::{Cli, Command, RunArgs, ShowArgs};
 use bgp::{Session, StateView};
 use clap::Parser;
 use env_logger::fmt::Formatter;
-use futures::future::select;
-use futures::FutureExt;
 use ipc::{get_sock_path, IpcServer};
 use itertools::Itertools;
 use log::{error, info, warn, Level, LevelFilter, Record};
@@ -28,10 +26,19 @@ use std::path::Path;
 use std::process::ExitCode;
 use tokio::io::BufReader;
 use tokio::net::TcpListener;
-use tokio::signal::unix::{signal, SignalKind};
-use tokio::{pin, select};
+use tokio::select;
 use util::{BOLD, FG_GREEN_BOLD, RESET};
 
+#[cfg(test)]
+use std::future::pending;
+
+#[cfg(not(test))]
+use {
+  futures::future::{select, FutureExt},
+  tokio::pin,
+  tokio::signal::unix::{signal, SignalKind},
+};
+
 async fn run(mut args: RunArgs, sock_path: &str) -> anyhow::Result<ExitCode> {
   if let Some(file) = args.file {
     let cmd = std::env::args().next().unwrap();
@@ -63,15 +70,23 @@ async fn run(mut args: RunArgs, sock_path: &str) -> anyhow::Result<ExitCode> {
 
   info!("Flow listening to {bind:?} as AS{local_as}, router ID {router_id}");
 
-  let mut sigint = signal(SignalKind::interrupt()).context("failed to register signal handler")?;
-  let mut sigterm = signal(SignalKind::terminate()).context("failed to register signal handler")?;
+  #[cfg(not(test))]
+  let (mut sigint, mut sigterm) = (
+    signal(SignalKind::interrupt()).context("failed to register signal handler")?,
+    signal(SignalKind::terminate()).context("failed to register signal handler")?,
+  );
 
   loop {
     let select = async {
+      #[cfg(not(test))]
       pin! {
         let sigint = sigint.recv().map(|_| "SIGINT");
         let sigterm = sigterm.recv().map(|_| "SIGTERM");
+        let signal_select = select(sigint, sigterm);
       }
+      #[cfg(test)]
+      let signal_select = pending();
+
       select! {
         result = listener.accept(), if matches!(bgp.state(), bgp::State::Active) => {
           let (stream, mut addr) = result.context("failed to accept TCP connection")?;
@@ -88,10 +103,14 @@ async fn run(mut args: RunArgs, sock_path: &str) -> anyhow::Result<ExitCode> {
           let mut stream = result.context("failed to accept IPC connection")?;
           bgp.write_states(&mut stream).await.context("failed to write to IPC channel")?;
         },
-        signal = select(sigint, sigterm) => {
-          let (signal, _) = signal.factor_first();
-          warn!("{signal} received, exiting");
-          return Ok(Some(ExitCode::SUCCESS))
+
+        _signal = signal_select => {
+          #[cfg(not(test))]
+          {
+            let (signal, _) = _signal.factor_first();
+            warn!("{signal} received, exiting");
+            return Ok(Some(ExitCode::SUCCESS))
+          }
         }
       }
       anyhow::Ok(None)
@@ -176,11 +195,14 @@ fn format_log(f: &mut Formatter, record: &Record<'_>) -> io::Result<()> {
 
 pub async fn cli_entry(cli: Cli) -> ExitCode {
   let sock_path = get_sock_path(&cli.run_dir).unwrap();
-  env_logger::builder()
+  let mut builder = env_logger::builder();
+  builder
     .filter_level(cli.verbosity.log_level_filter())
     .format(format_log)
-    .filter_module("netlink", LevelFilter::Off)
-    .init();
+    .filter_module("netlink", LevelFilter::Off);
+  #[cfg(test)]
+  builder.is_test(true);
+  builder.init();
   match cli.command {
     Command::Run(args) => match run(args, &sock_path).await {
       Ok(x) => x,