diff --git a/Cargo.lock b/Cargo.lock
index 3bbde48..673e14b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1,6 +1,6 @@
# This file is automatically @generated by Cargo.
# It is not intended for manual editing.
-version = 3
+version = 4
[[package]]
name = "addr2line"
@@ -72,6 +72,15 @@ version = "1.0.95"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "34ac096ce696dc2fcabef30516bb13c0a68a11d30131d3df6f04711467681b04"
+[[package]]
+name = "async-tempfile"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "acb90d9834a8015109afc79f1f548223a0614edcbab62fb35b62d4b707e975e7"
+dependencies = [
+ "tokio",
+]
+
[[package]]
name = "autocfg"
version = "1.4.0"
@@ -247,6 +256,7 @@ version = "0.2.0"
dependencies = [
"anstyle",
"anyhow",
+ "async-tempfile",
"cfg_aliases",
"clap",
"clap-verbosity-flag",
diff --git a/Cargo.toml b/Cargo.toml
index ad0ba64..e257eae 100644
--- a/Cargo.toml
+++ b/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"
diff --git a/src/integration_tests/config/basic.bird.conf b/src/integration_tests/config/basic.bird.conf
new file mode 100644
index 0000000..12335e6
--- /dev/null
+++ b/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; };
+}
diff --git a/src/integration_tests/helpers/bird.rs b/src/integration_tests/helpers/bird.rs
index 444e478..a0b5e5e 100644
--- a/src/integration_tests/helpers/bird.rs
+++ b/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)))
+}
diff --git a/src/integration_tests/helpers/cli.rs b/src/integration_tests/helpers/cli.rs
new file mode 100644
index 0000000..9a57d67
--- /dev/null
+++ b/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
+}
diff --git a/src/integration_tests/helpers/kernel.rs b/src/integration_tests/helpers/kernel.rs
index 35e711e..6c27c3d 100644
--- a/src/integration_tests/helpers/kernel.rs
+++ b/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() {
diff --git a/src/integration_tests/helpers/mod.rs b/src/integration_tests/helpers/mod.rs
index 3f05bbb..dda4cce 100644
--- a/src/integration_tests/helpers/mod.rs
+++ b/src/integration_tests/helpers/mod.rs
@@ -1,2 +1,3 @@
pub mod bird;
+pub mod cli;
pub mod kernel;
diff --git a/src/integration_tests/mod.rs b/src/integration_tests/mod.rs
index 1c47e85..6d92f77 100644
--- a/src/integration_tests/mod.rs
+++ b/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
+}
diff --git a/src/main.rs b/src/main.rs
index e9d762e..f768e5a 100644
--- a/src/main.rs
+++ b/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,