feat(jstzd): implement async OctezNode

crates/jstzd/Cargo.toml

@@ -12,6 +12,7 @@ async-trait.workspace = true
 bollard.workspace = true
 futures-util.workspace = true
 octez = { path = "../octez" }
+serde.workspace = true
 tokio.workspace = true
 
 [dev-dependencies]

crates/jstzd/src/task/async_octez_node.rs

@@ -0,0 +1,262 @@
+use std::{fs::File, path::PathBuf, process::Stdio};
+
+#[cfg(test)]
+use tests::{Child, Command};
+
+#[cfg(not(test))]
+use tokio::process::{Child, Command};
+
+use anyhow::{anyhow, Result};
+
+fn path_or_default<'a>(path: Option<&'a PathBuf>, default: &'a str) -> &'a str {
+    path.and_then(|bin| bin.to_str()).unwrap_or(default)
+}
+
+async fn run_command(command: &mut Command) -> Result<()> {
+    let output = command.output().await?;
+
+    if !output.status.success() {
+        return Err(anyhow!(
+            "Command {:?} failed:\n {}",
+            command,
+            String::from_utf8_lossy(&output.stderr)
+        ));
+    }
+
+    Ok(())
+}
+
+pub struct AsyncOctezNode {
+    /// Path to the octez-node binary
+    /// If None, the binary will inside PATH will be used
+    pub octez_node_bin: Option<PathBuf>,
+    /// Path to the octez-node directory
+    pub octez_node_dir: PathBuf,
+}
+
+impl AsyncOctezNode {
+    fn command(&self) -> Command {
+        Command::new(path_or_default(self.octez_node_bin.as_ref(), "octez-node"))
+    }
+
+    pub async fn config_init(
+        &self,
+        network: &str,
+        rpc_endpoint: &str,
+        num_connections: u32,
+    ) -> Result<()> {
+        run_command(self.command().args([
+            "config",
+            "init",
+            "--network",
+            network,
+            "--data-dir",
+            self.octez_node_dir.to_str().expect("Invalid path"),
+            "--rpc-addr",
+            rpc_endpoint,
+            "--connections",
+            num_connections.to_string().as_str(),
+        ]))
+        .await
+    }
+
+    pub async fn generate_identity(&self) -> Result<()> {
+        run_command(self.command().args([
+            "identity",
+            "generate",
+            "0",
+            "--data-dir",
+            self.octez_node_dir.to_str().expect("Invalid path"),
+        ]))
+        .await
+    }
+
+    pub async fn run(&self, log_file: &File, options: &[&str]) -> Result<Child> {
+        let mut command = self.command();
+
+        command
+            .args([
+                "run",
+                "--data-dir",
+                self.octez_node_dir.to_str().expect("Invalid path"),
+                "--singleprocess",
+            ])
+            .args(options)
+            .stdout(Stdio::from(log_file.try_clone()?))
+            .stderr(Stdio::from(log_file.try_clone()?));
+
+        Ok(command.spawn()?)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{path_or_default, run_command, AsyncOctezNode};
+    use std::{fs::File, path::PathBuf, process::Stdio};
+
+    pub struct Child {
+        pub cmd: String,
+    }
+
+    pub struct CommandOutputStatus {
+        ok: bool,
+    }
+    impl CommandOutputStatus {
+        pub fn success(&self) -> bool {
+            self.ok
+        }
+    }
+    pub struct CommandOutput {
+        pub status: CommandOutputStatus,
+        pub stderr: Vec<u8>,
+    }
+    #[derive(Debug)]
+    pub struct Command {
+        inner: String,
+    }
+    impl Command {
+        pub fn new(s: &str) -> Self {
+            Command {
+                inner: s.to_owned(),
+            }
+        }
+        pub fn args<I, T>(&mut self, args: I) -> &mut Self
+        where
+            I: IntoIterator<Item = T>,
+            T: AsRef<std::ffi::OsStr>,
+        {
+            let args_str = args
+                .into_iter()
+                .map(|v| v.as_ref().to_str().unwrap().to_owned())
+                .collect::<Vec<String>>()
+                .join(" ");
+            if !args_str.is_empty() {
+                self.inner += " ";
+                self.inner += &args_str;
+            }
+            self
+        }
+        pub async fn output(&self) -> anyhow::Result<CommandOutput> {
+            let mut msg = Vec::new();
+            let mut ok = true;
+            if !self.inner.starts_with("ok") {
+                msg = Vec::from("this is the error message");
+                ok = false;
+            }
+            Ok(CommandOutput {
+                status: CommandOutputStatus { ok },
+                stderr: msg,
+            })
+        }
+        pub fn spawn(&self) -> Result<Child, std::io::Error> {
+            Ok(Child {
+                cmd: self.inner.clone(),
+            })
+        }
+
+        pub fn stdout<T: Into<Stdio>>(&mut self, _cfg: T) -> &mut Self {
+            self
+        }
+        pub fn stderr<T: Into<Stdio>>(&mut self, _cfg: T) -> &mut Self {
+            self
+        }
+    }
+
+    #[tokio::test]
+    async fn run_command_ok() {
+        // No surprise in command
+        let mut command = Command {
+            inner: "ok".to_owned(),
+        };
+        assert!(run_command(&mut command).await.is_ok());
+    }
+
+    #[tokio::test]
+    async fn run_command_err() {
+        // When the command is not "ok", the command fails
+        let mut command = Command {
+            inner: "fail".to_owned(),
+        };
+        assert!(run_command(&mut command)
+            .await
+            .unwrap_err()
+            .to_string()
+            .contains("this is the error message"));
+    }
+
+    #[test]
+    fn test_path_or_default() {
+        let path = PathBuf::from("/foo/bar");
+        assert_eq!(path_or_default(Some(&path), "not_this_one"), "/foo/bar");
+        assert_eq!(
+            path_or_default(None, "should_return_this"),
+            "should_return_this"
+        );
+    }
+
+    #[test]
+    fn octez_node_command() {
+        let node = AsyncOctezNode {
+            octez_node_bin: None,
+            octez_node_dir: PathBuf::from("/some_dir"),
+        };
+        assert_eq!(node.command().inner, "octez-node");
+
+        let node = AsyncOctezNode {
+            octez_node_bin: Some(PathBuf::from("/path/to/bin")),
+            octez_node_dir: PathBuf::from("/some_dir"),
+        };
+        assert_eq!(node.command().inner, "/path/to/bin");
+    }
+
+    #[tokio::test]
+    async fn octez_node_config_init() {
+        // Setting the bin path to `my-node` fails the command.
+        // Then here we can check if the args were passed into the command correctly.
+        let node = AsyncOctezNode {
+            octez_node_bin: Some(PathBuf::from("my-node")),
+            octez_node_dir: PathBuf::from("/some_dir"),
+        };
+        assert!(
+            node.config_init("foo", "bar", 42)
+                .await
+                .unwrap_err()
+                .to_string()
+                .contains(
+                    "my-node config init --network foo --data-dir /some_dir --rpc-addr bar --connections 42"
+                )
+        );
+    }
+
+    #[tokio::test]
+    async fn octez_node_generate_identity() {
+        // Setting the bin path to `my-node` fails the command.
+        // Then here we can check if the args were passed into the command correctly.
+        let node = AsyncOctezNode {
+            octez_node_bin: Some(PathBuf::from("my-node")),
+            octez_node_dir: PathBuf::from("/some_dir"),
+        };
+        assert!(node
+            .generate_identity()
+            .await
+            .unwrap_err()
+            .to_string()
+            .contains("my-node identity generate 0 --data-dir /some_dir"));
+    }
+
+    #[tokio::test]
+    async fn octez_node_run() {
+        // Setting the bin path to `my-node` fails the command.
+        // Then here we can check if the args were passed into the command correctly.
+        let node = AsyncOctezNode {
+            octez_node_bin: Some(PathBuf::from("my-node")),
+            octez_node_dir: PathBuf::from("/some_dir"),
+        };
+        assert!(node
+            .run(&File::open("/dev/null").unwrap(), &[])
+            .await
+            .unwrap()
+            .cmd
+            .contains("my-node run --data-dir /some_dir --singleprocess"));
+    }
+}

crates/jstzd/src/task/mod.rs

@@ -1,3 +1,4 @@
+mod async_octez_node;
 pub mod octez_node;
 
 use anyhow::Result;

crates/jstzd/src/task/octez_node.rs

@@ -6,12 +6,12 @@ use std::{fs::File, path::PathBuf, sync::Arc};
 use tokio::sync::RwLock;
 
 #[cfg(test)]
-use tests::{Child, InnerOctezNode};
+use tests::{AsyncOctezNode, Child};
 
 #[cfg(not(test))]
-use octez::OctezNode as InnerOctezNode;
+use crate::task::async_octez_node::AsyncOctezNode;
 #[cfg(not(test))]
-use std::process::Child;
+use tokio::process::Child;
 
 #[derive(Clone)]
 pub struct OctezNodeConfig {
@@ -105,7 +105,7 @@ struct ChildWrapper {
 impl ChildWrapper {
     pub async fn kill(&mut self) -> anyhow::Result<()> {
         if let Some(mut v) = self.inner.take() {
-            return Ok(v.kill()?);
+            v.kill().await?;
         }
         Ok(())
     }
@@ -129,13 +129,14 @@ impl Task for OctezNode {
 
     /// Spins up the task with the given config.
     async fn spawn(config: Self::Config) -> Result<Self> {
-        let node = InnerOctezNode {
+        let node = AsyncOctezNode {
             octez_node_bin: Some(config.binary_path),
             octez_node_dir: config.data_dir,
         };
 
-        node.config_init(&config.network, "localhost:8731", &config.rpc_endpoint, 0)?;
-        node.generate_identity()?;
+        node.generate_identity().await?;
+        node.config_init(&config.network, &config.rpc_endpoint, 0)
+            .await?;
         Ok(OctezNode {
             inner: Arc::new(RwLock::new(AsyncDropper::new(ChildWrapper {
                 inner: Some(
@@ -147,7 +148,8 @@ impl Task for OctezNode {
                             .map(|s| s as &str)
                             .collect::<Vec<&str>>()
                             .as_slice(),
-                    )?,
+                    )
+                    .await?,
                 ),
             }))),
         })
@@ -178,7 +180,7 @@ mod tests {
     }
 
     impl Child {
-        pub fn kill(&mut self) -> std::io::Result<()> {
+        pub async fn kill(&mut self) -> std::io::Result<()> {
             if self.file_path.is_none() {
                 return Err(std::io::Error::new(
                     std::io::ErrorKind::Other,
@@ -190,23 +192,22 @@ mod tests {
         }
     }
 
-    pub struct InnerOctezNode {
+    pub struct AsyncOctezNode {
         pub octez_node_bin: Option<PathBuf>,
         pub octez_node_dir: PathBuf,
     }
 
-    impl InnerOctezNode {
-        pub fn config_init(
+    impl AsyncOctezNode {
+        pub async fn config_init(
             &self,
             _network: &str,
-            _http_endpoint: &str,
             _rpc_endpoint: &str,
             _num_connections: u32,
         ) -> Result<()> {
             Ok(())
         }
 
-        pub fn run(&self, _log_file: &File, options: &[&str]) -> Result<Child> {
+        pub async fn run(&self, _log_file: &File, options: &[&str]) -> Result<Child> {
             let file_path = if !options.is_empty() {
                 Some(options[0].to_owned())
             } else {
@@ -215,7 +216,7 @@ mod tests {
             Ok(Child { file_path })
         }
 
-        pub fn generate_identity(&self) -> Result<()> {
+        pub async fn generate_identity(&self) -> Result<()> {
             Ok(())
         }
     }