integration_tests: add simple test for multibackend

Author: fcangialosi

tests/basic.rs

@@ -76,5 +76,9 @@ impl IntegrationTest for TestBasicSerialize {
 fn basic() {
     let log = libccp_integration::logger();
     info!(log, "starting basic test");
-    libccp_integration::run_test::<TestBasicSerialize>(log, 1);
+    libccp_integration::run_test::<TestBasicSerialize>(log, 1, false);
+
+    let log2 = libccp_integration::logger();
+    info!(log2, "starting basic test (multi)");
+    libccp_integration::run_test::<TestBasicSerialize>(log2, 1, true);
 }

tests/libccp_integration/mod.rs

@@ -80,7 +80,7 @@ impl<I: Ipc, T: IntegrationTest> Flow for TestBase<I, T> {
 }
 
 use portus::ipc::chan::Socket;
-use portus::ipc::{SingleBackendBuilder, Blocking};
+use portus::ipc::{Blocking, MultiBackendBuilder, SingleBackendBuilder};
 use std;
 use std::thread;
 
@@ -100,8 +100,27 @@ fn start_ccp<T: IntegrationTest + 'static + Send>(
     )
 }
 
+fn start_ccp_multi<T: IntegrationTest + 'static + Send>(
+    sks: Vec<Socket<Blocking>>,
+    log: slog::Logger,
+    tx: mpsc::Sender<Result<(), ()>>,
+) -> portus::CCPHandle {
+    let b = MultiBackendBuilder { socks: sks };
+    portus::spawn::<Socket<Blocking>, TestBaseConfig<T>, MultiBackendBuilder<Socket<Blocking>>>(
+        b,
+        portus::Config {
+            logger: Some(log.clone()),
+        },
+        TestBaseConfig(tx, Some(log.clone()), PhantomData::<T>),
+    )
+}
+
 // Runs a specific intergration test
-pub fn run_test<T: IntegrationTest + 'static + Send>(log: slog::Logger, num_flows: usize) {
+pub fn run_test<T: IntegrationTest + 'static + Send>(
+    log: slog::Logger,
+    num_flows: usize,
+    use_multi: bool,
+) {
     let (tx, rx) = std::sync::mpsc::channel();
 
     // Channel for IPC
@@ -113,14 +132,17 @@ pub fn run_test<T: IntegrationTest + 'static + Send>(log: slog::Logger, num_flow
     let (dp_handle, conn_handles) = mock_datapath::start(dp_log, num_flows, s2, r1);
 
     let sk = Socket::<Blocking>::new(s1, r2);
-    let ccp_handle = start_ccp::<T>(sk, log.clone(), tx);
+    let ccp_handle = match use_multi {
+        false => start_ccp::<T>(sk, log.clone(), tx),
+        true => start_ccp_multi::<T>(vec![sk], log.clone(), tx),
+    };
 
     // wait for program to finish
     let wait_for_done = thread::spawn(move || {
         rx.recv_timeout(std::time::Duration::from_secs(20))
             .unwrap()
             .unwrap();
-        ccp_handle.kill(); // causes backend to stop iterating
+        ccp_handle.kill();
         ccp_handle.wait().unwrap();
 
         for h in conn_handles {

tests/preset.rs

@@ -69,5 +69,5 @@ impl IntegrationTest for TestPresetVars {
 fn preset() {
     let log = libccp_integration::logger();
     info!(log, "starting preset test");
-    libccp_integration::run_test::<TestPresetVars>(log, 1);
+    libccp_integration::run_test::<TestPresetVars>(log, 1, false);
 }

tests/timing.rs

@@ -83,5 +83,5 @@ impl IntegrationTest for TestTiming {
 fn timing() {
     let log = libccp_integration::logger();
     info!(log, "starting timing test");
-    libccp_integration::run_test::<TestTiming>(log, 1);
+    libccp_integration::run_test::<TestTiming>(log, 1, false);
 }

tests/twoflow.rs

@@ -88,5 +88,5 @@ impl IntegrationTest for TestTwoFlows {
 fn twoflow() {
     let log = libccp_integration::logger();
     info!(log, "starting twoflow test");
-    libccp_integration::run_test::<TestTwoFlows>(log, 2);
+    libccp_integration::run_test::<TestTwoFlows>(log, 2, false);
 }

tests/update.rs

@@ -88,5 +88,5 @@ impl IntegrationTest for TestUpdateFields {
 fn update() {
     let log = libccp_integration::logger();
     info!(log, "starting update test");
-    libccp_integration::run_test::<TestUpdateFields>(log, 1);
+    libccp_integration::run_test::<TestUpdateFields>(log, 1, false);
 }

tests/volatile.rs

@@ -74,5 +74,5 @@ impl IntegrationTest for TestVolatileVars {
 fn volatile() {
     let log = libccp_integration::logger();
     info!(log, "starting volatile test");
-    libccp_integration::run_test::<TestVolatileVars>(log, 1);
+    libccp_integration::run_test::<TestVolatileVars>(log, 1, false);
 }