process: Move files to their own directory

tokio-process/tests/issue_42.rs

@@ -0,0 +1,51 @@
+#![cfg(unix)]
+
+extern crate futures;
+extern crate tokio_process;
+
+use futures::{stream, Future, IntoFuture, Stream};
+use std::process::{Command, Stdio};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::thread;
+use std::time::Duration;
+use tokio_process::CommandExt;
+
+fn run_test() {
+    let finished = Arc::new(AtomicBool::new(false));
+    let finished_clone = finished.clone();
+
+    thread::spawn(move || {
+        let _ = stream::iter_ok(0..2)
+            .map(|i| {
+                Command::new("echo")
+                    .arg(format!("I am spawned process #{}", i))
+                    .stdin(Stdio::null())
+                    .stdout(Stdio::null())
+                    .stderr(Stdio::null())
+                    .spawn_async()
+                    .into_future()
+                    .flatten()
+            })
+            .buffered(2)
+            .collect()
+            .wait();
+
+        finished_clone.store(true, Ordering::SeqCst);
+    });
+
+    thread::sleep(Duration::from_millis(100));
+    assert!(
+        finished.load(Ordering::SeqCst),
+        "FINISHED flag not set, maybe we deadlocked?"
+    );
+}
+
+#[test]
+fn issue_42() {
+    let max = 10;
+    for i in 0..max {
+        println!("running {}/{}", i, max);
+        run_test()
+    }
+}

tokio-process/tests/smoke.rs

@@ -0,0 +1,22 @@
+extern crate tokio_process;
+
+use tokio_process::CommandExt;
+
+mod support;
+
+#[test]
+fn simple() {
+    let mut cmd = support::cmd("exit");
+    cmd.arg("2");
+
+    let mut child = cmd.spawn_async().unwrap();
+
+    let id = child.id();
+    assert!(id > 0);
+
+    let status = support::run_with_timeout(&mut child).expect("failed to run future");
+    assert_eq!(status.code(), Some(2));
+
+    assert_eq!(child.id(), id);
+    drop(child.kill());
+}

tokio-process/tests/stdio.rs

@@ -0,0 +1,113 @@
+extern crate futures;
+#[macro_use]
+extern crate log;
+extern crate tokio_io;
+extern crate tokio_process;
+
+use std::io;
+use std::process::{Command, ExitStatus, Stdio};
+
+use futures::future::Future;
+use futures::stream::{self, Stream};
+use tokio_io::io::{read_until, write_all};
+use tokio_process::{Child, CommandExt};
+
+mod support;
+
+fn cat() -> Command {
+    let mut cmd = support::cmd("cat");
+    cmd.stdin(Stdio::piped()).stdout(Stdio::piped());
+    cmd
+}
+
+fn feed_cat(mut cat: Child, n: usize) -> Box<Future<Item = ExitStatus, Error = io::Error>> {
+    let stdin = cat.stdin().take().unwrap();
+    let stdout = cat.stdout().take().unwrap();
+
+    debug!("starting to feed");
+    // Produce n lines on the child's stdout.
+    let numbers = stream::iter_ok(0..n);
+    let write = numbers
+        .fold(stdin, |stdin, i| {
+            debug!("sending line {} to child", i);
+            write_all(stdin, format!("line {}\n", i).into_bytes()).map(|p| p.0)
+        })
+        .map(|_| ());
+
+    // Try to read `n + 1` lines, ensuring the last one is empty
+    // (i.e. EOF is reached after `n` lines.
+    let reader = io::BufReader::new(stdout);
+    let expected_numbers = stream::iter_ok(0..=n);
+    let read = expected_numbers.fold((reader, 0), move |(reader, i), _| {
+        let done = i >= n;
+        debug!("starting read from child");
+        read_until(reader, b'\n', Vec::new()).and_then(move |(reader, vec)| {
+            debug!(
+                "read line {} from child ({} bytes, done: {})",
+                i,
+                vec.len(),
+                done
+            );
+            match (done, vec.len()) {
+                (false, 0) => Err(io::Error::new(io::ErrorKind::BrokenPipe, "broken pipe")),
+                (true, n) if n != 0 => Err(io::Error::new(io::ErrorKind::Other, "extraneous data")),
+                _ => {
+                    let s = std::str::from_utf8(&vec).unwrap();
+                    let expected = format!("line {}\n", i);
+                    if done || s == expected {
+                        Ok((reader, i + 1))
+                    } else {
+                        Err(io::Error::new(io::ErrorKind::Other, "unexpected data"))
+                    }
+                }
+            }
+        })
+    });
+
+    // Compose reading and writing concurrently.
+    Box::new(write.join(read).and_then(|_| cat))
+}
+
+/// Check for the following properties when feeding stdin and
+/// consuming stdout of a cat-like process:
+///
+/// - A number of lines that amounts to a number of bytes exceeding a
+///   typical OS buffer size can be fed to the child without
+///   deadlock. This tests that we also consume the stdout
+///   concurrently; otherwise this would deadlock.
+///
+/// - We read the same lines from the child that we fed it.
+///
+/// - The child does produce EOF on stdout after the last line.
+#[test]
+fn feed_a_lot() {
+    let child = cat().spawn_async().unwrap();
+    let status = support::run_with_timeout(feed_cat(child, 10000)).unwrap();
+    assert_eq!(status.code(), Some(0));
+}
+
+#[test]
+fn wait_with_output_captures() {
+    let mut child = cat().spawn_async().unwrap();
+    let stdin = child.stdin().take().unwrap();
+    let out = child.wait_with_output();
+
+    let future = write_all(stdin, b"1234").map(|p| p.1).join(out);
+    let ret = support::run_with_timeout(future).unwrap();
+    let (written, output) = ret;
+
+    assert!(output.status.success());
+    assert_eq!(output.stdout, written);
+    assert_eq!(output.stderr.len(), 0);
+}
+
+#[test]
+fn status_closes_any_pipes() {
+    // Cat will open a pipe between the parent and child.
+    // If `status_async` doesn't ensure the handles are closed,
+    // we would end up blocking forever (and time out).
+    let child = cat().status_async().expect("failed to spawn child");
+
+    support::run_with_timeout(child)
+        .expect("time out exceeded! did we get stuck waiting on the child?");
+}

tokio-process/tests/support/mod.rs

@@ -0,0 +1,43 @@
+extern crate futures;
+extern crate tokio;
+
+use self::futures::Future;
+use self::tokio::timer::Timeout;
+use std::env;
+use std::process::Command;
+use std::time::Duration;
+
+pub use self::tokio::runtime::current_thread::Runtime as CurrentThreadRuntime;
+
+pub fn cmd(s: &str) -> Command {
+    let mut me = env::current_exe().unwrap();
+    me.pop();
+    if me.ends_with("deps") {
+        me.pop();
+    }
+    me.push(s);
+    Command::new(me)
+}
+
+pub fn with_timeout<F: Future>(future: F) -> impl Future<Item = F::Item, Error = F::Error> {
+    Timeout::new(future, Duration::from_secs(3)).map_err(|e| {
+        if e.is_timer() {
+            panic!("failed to register timer");
+        } else if e.is_elapsed() {
+            panic!("timed out")
+        } else {
+            e.into_inner().expect("missing inner error")
+        }
+    })
+}
+
+pub fn run_with_timeout<F>(future: F) -> Result<F::Item, F::Error>
+where
+    F: Future,
+{
+    // NB: Timeout requires a timer registration which is provided by
+    // tokio's `current_thread::Runtime`, but isn't available by just using
+    // tokio's default CurrentThread executor which powers `current_thread::block_on_all`.
+    let mut rt = CurrentThreadRuntime::new().expect("failed to get runtime");
+    rt.block_on(with_timeout(future))
+}