[CIT-523] Support for update-list command in plugins (#440)

8 files changed, 405 insertions, 52 deletions

diff --git a/sm/json_sm/src/error.rs b/sm/json_sm/src/error.rs
index 923d47f5..05ffef46 100644
--- a/sm/json_sm/src/error.rs
+++ b/sm/json_sm/src/error.rs
@@ -46,6 +46,12 @@ pub enum SoftwareError {
         reason: String,
     },
 
+    #[error("Failed to execute updates for {software_type:?}")]
+    UpdateList {
+        software_type: SoftwareType,
+        reason: String,
+    },
+
     #[error("Unknown {software_type:?} module: {name:?}")]
     UnknownModule {
         software_type: SoftwareType,
@@ -71,8 +77,14 @@ pub enum SoftwareError {
     #[error("The configured default plugin: {0} not found")]
     InvalidDefaultPlugin(String),
 
+    #[error("The update-list command is not supported by this: {0} plugin")]
+    UpdateListNotSupported(String),
+
     #[error("I/O error: {reason:?}")]
     IoError { reason: String },
+
+    #[error("Plugin output contains invalid UTF-8 characters")]
+    NonUtf8Output,
 }
 
 impl From<serde_json::Error> for SoftwareError {
diff --git a/sm/plugin_sm/src/logged_command.rs b/sm/plugin_sm/src/logged_command.rs
index 72cb2b01..11739ce0 100644
--- a/sm/plugin_sm/src/logged_command.rs
+++ b/sm/plugin_sm/src/logged_command.rs
@@ -2,7 +2,26 @@ use std::ffi::OsStr;
 use std::process::{Output, Stdio};
 use tokio::fs::File;
 use tokio::io::{AsyncWriteExt, BufWriter};
-use tokio::process::Command;
+use tokio::process::{Child, Command};
+
+pub struct LoggingChild {
+    command_line: String,
+    pub inner_child: Child,
+}
+
+impl LoggingChild {
+    pub async fn wait_with_output(
+        self,
+        logger: &mut BufWriter<File>,
+    ) -> Result<Output, std::io::Error> {
+        let outcome = self.inner_child.wait_with_output().await;
+        if let Err(err) = LoggedCommand::log_outcome(&self.command_line, &outcome, logger).await {
+            tracing::log::error!("Fail to log the command execution: {}", err);
+        }
+
+        outcome
+    }
+}
 
 /// A command which execution is logged.
 ///
@@ -33,7 +52,7 @@ impl LoggedCommand {
         let mut command = Command::new(program);
         command
             .current_dir("/tmp")
-            .stdin(Stdio::null())
+            .stdin(Stdio::piped())
             .stdout(Stdio::piped())
             .stderr(Stdio::piped());
 
@@ -67,6 +86,14 @@ impl LoggedCommand {
         outcome
     }
 
+    pub fn spawn(&mut self) -> Result<LoggingChild, std::io::Error> {
+        let child = self.command.spawn()?;
+        Ok(LoggingChild {
+            command_line: self.command_line.clone(),
+            inner_child: child,
+        })
+    }
+
     async fn log_outcome(
         command_line: &str,
         result: &Result<Output, std::io::Error>,
diff --git a/sm/plugin_sm/src/plugin.rs b/sm/plugin_sm/src/plugin.rs
index 77cde63e..80d981c8 100644
--- a/sm/plugin_sm/src/plugin.rs
+++ b/sm/plugin_sm/src/plugin.rs
@@ -1,7 +1,9 @@
 use crate::logged_command::LoggedCommand;
 use async_trait::async_trait;
 use download::Downloader;
-use json_sm::*;
+use json_sm::{
+    DownloadInfo, SoftwareError, SoftwareModule, SoftwareModuleUpdate, SoftwareType, DEFAULT,
+};
 use std::{iter::Iterator, path::PathBuf, process::Output};
 use tokio::io::BufWriter;
 use tokio::{fs::File, io::AsyncWriteExt};
@@ -9,21 +11,32 @@ use tokio::{fs::File, io::AsyncWriteExt};
 #[async_trait]
 pub trait Plugin {
     async fn prepare(&self, logger: &mut BufWriter<File>) -> Result<(), SoftwareError>;
+
     async fn install(
         &self,
         module: &SoftwareModule,
         logger: &mut BufWriter<File>,
     ) -> Result<(), SoftwareError>;
+
     async fn remove(
         &self,
         module: &SoftwareModule,
         logger: &mut BufWriter<File>,
     ) -> Result<(), SoftwareError>;
+
+    async fn update_list(
+        &self,
+        modules: &Vec<SoftwareModuleUpdate>,
+        logger: &mut BufWriter<File>,
+    ) -> Result<(), SoftwareError>;
+
     async fn finalize(&self, logger: &mut BufWriter<File>) -> Result<(), SoftwareError>;
+
     async fn list(
         &self,
         logger: &mut BufWriter<File>,
     ) -> Result<Vec<SoftwareModule>, SoftwareError>;
+
     async fn version(
         &self,
         module: &SoftwareModule,
@@ -51,26 +64,63 @@ pub trait Plugin {
 
     async fn apply_all(
         &self,
-        updates: Vec<SoftwareModuleUpdate>,
+        mut updates: Vec<SoftwareModuleUpdate>,
         logger: &mut BufWriter<File>,
     ) -> Vec<SoftwareError> {
         let mut failed_updates = Vec::new();
 
+        // Prepare the updates
         if let Err(prepare_error) = self.prepare(logger).await {
             failed_updates.push(prepare_error);
             return failed_updates;
         }
 
-        for update in updates.iter() {
-            if let Err(error) = self.apply(update, logger).await {
-                failed_updates.push(error);
+        // Download all modules for which a download URL is provided
+        let mut downloaders = Vec::new();
+        for update in updates.iter_mut() {
+            let module = match update {
+                SoftwareModuleUpdate::Remove { module } => module,
+                SoftwareModuleUpdate::Install { module } => module,
             };
+            let module_url = module.url.clone();
+            if let Some(url) = module_url {
+                match Self::download_from_url(module, &url, logger).await {
+                    Err(prepare_error) => {
+                        failed_updates.push(prepare_error);
+                        break;
+                    }
+                    Ok(downloader) => downloaders.push(downloader),
+                }
+            }
+        }
+
+        // Execute the updates
+        if failed_updates.is_empty() {
+            let outcome = self.update_list(&updates, logger).await;
+            if let Err(SoftwareError::UpdateListNotSupported(_)) = outcome {
+                for update in updates.iter() {
+                    if let Err(error) = self.apply(update, logger).await {
+                        failed_updates.push(error);
+                    };
+                }
+            } else if let Err(update_list_error) = outcome {
+                failed_updates.push(update_list_error);
+            }
         }
 
+        // Finalize the updates
         if let Err(finalize_error) = self.finalize(logger).await {
             failed_updates.push(finalize_error);
         }
 
+        // Cleanup all the downloaded modules
+        for downloader in downloaders {
+            if let Err(cleanup_error) = Self::cleanup_downloaded_artefacts(downloader, logger).await
+            {
+                failed_updates.push(cleanup_error);
+            }
+        }
+
         failed_updates
     }
 
@@ -80,6 +130,18 @@ pub trait Plugin {
         url: &DownloadInfo,
         logger: &mut BufWriter<File>,
     ) -> Result<(), SoftwareError> {
+        let downloader = Self::download_from_url(module, url, logger).await?;
+        let result = self.install(module, logger).await;
+        Self::cleanup_downloaded_artefacts(downloader, logger).await?;
+
+        result
+    }
+
+    async fn download_from_url(
+        module: &mut SoftwareModule,
+        url: &DownloadInfo,
+        logger: &mut BufWriter<File>,
+    ) -> Result<Downloader, SoftwareError> {
         let downloader = Downloader::new(&module.name, &module.version, "/tmp");
 
         logger
@@ -110,21 +172,26 @@ pub trait Plugin {
         }
 
         module.file_path = Some(downloader.filename().to_owned());
-        let result = self.install(module, logger).await;
+
+        Ok(downloader)
+    }
+
+    async fn cleanup_downloaded_artefacts(
+        downloader: Downloader,
+        logger: &mut BufWriter<File>,
+    ) -> Result<(), SoftwareError> {
         if let Err(err) = downloader
             .cleanup()
             .await
-            .map_err(|err| SoftwareError::DownloadError {
+            .map_err(|err| SoftwareError::IoError {
                 reason: err.to_string(),
-                url: url.url().to_string(),
             })
         {
             logger
                 .write_all(format!("warn: {}\n", &err).as_bytes())
                 .await?;
         }
-
-        result
+        Ok(())
     }
 }
 
@@ -215,6 +282,7 @@ impl ExternalPluginCommand {
 const PREPARE: &str = "prepare";
 const INSTALL: &str = "install";
 const REMOVE: &str = "remove";
+const UPDATE_LIST: &str = "update-list";
 const FINALIZE: &str = "finalize";
 pub const LIST: &str = "list";
 const VERSION: &str = "version";
@@ -271,6 +339,63 @@ impl Plugin for ExternalPluginCommand {
         }
     }
 
+    async fn update_list(
+        &self,
+        updates: &Vec<SoftwareModuleUpdate>,
+        logger: &mut BufWriter<File>,
+    ) -> Result<(), SoftwareError> {
+        let mut command = self.command(UPDATE_LIST, None)?;
+
+        let mut child = command.spawn()?;
+        let child_stdin =
+            child
+                .inner_child
+                .stdin
+                .as_mut()
+                .ok_or_else(|| SoftwareError::IoError {
+                    reason: "Plugin stdin unavailable".into(),
+                })?;
+
+        for update in updates {
+            let action = match update {
+                SoftwareModuleUpdate::Install { module } => {
+                    format!(
+                        "install\t{}\t{}\t{}\n",
+                        module.name,
+                        module.version.clone().map_or("".into(), |v| v),
+                        module.file_path.clone().map_or("".into(), |v| v
+                            .to_str()
+                            .map_or("".into(), |u| u.to_string()))
+                    )
+                }
+
+                SoftwareModuleUpdate::Remove { module } => {
+                    format!(
+                        "remove\t{}\t{}\t\n",
+                        module.name,
+                        module.version.clone().map_or("".into(), |v| v),
+                    )
+                }
+            };
+
+            child_stdin.write_all(action.as_bytes()).await?
+        }
+
+        let output = child.wait_with_output(logger).await?;
+        match output.status.code() {
+            Some(0) => Ok(()),
+            Some(1) => Err(SoftwareError::UpdateListNotSupported(self.name.clone())),
+            Some(_) => Err(SoftwareError::UpdateList {
+                software_type: self.name.clone(),
+                reason: self.content(output.stderr)?,
+            }),
+            None => Err(SoftwareError::UpdateList {
+                software_type: self.name.clone(),
+                reason: "Interrupted".into(),
+            }),
+        }
+    }
+
     async fn finalize(&self, logger: &mut BufWriter<File>) -> Result<(), SoftwareError> {
         let command = self.command(FINALIZE, None)?;
         let output = self.execute(command, logger).await?;
diff --git a/sm/plugin_sm/src/plugin_manager.rs b/sm/plugin_sm/src/plugin_manager.rs
index e3825823..4311073b 100644
--- a/sm/plugin_sm/src/plugin_manager.rs
+++ b/sm/plugin_sm/src/plugin_manager.rs
@@ -1,6 +1,9 @@
-use crate::log_file::LogFile;
-use crate::plugin::*;
-use json_sm::*;
+use crate::plugin::{Plugin, LIST};
+use crate::{log_file::LogFile, plugin::ExternalPluginCommand};
+use json_sm::{
+    SoftwareError, SoftwareListRequest, SoftwareListResponse, SoftwareType, SoftwareUpdateRequest,
+    SoftwareUpdateResponse, DEFAULT,
+};
 use std::{
     collections::HashMap,
     fs,
diff --git a/sm/plugin_sm/tests/plugin.rs b/sm/plugin_sm/tests/plugin.rs
index 9d9c111a..72f2243a 100644
--- a/sm/plugin_sm/tests/plugin.rs
+++ b/sm/plugin_sm/tests/plugin.rs
@@ -1,7 +1,8 @@
 #[cfg(test)]
 mod tests {
 
-    use json_sm::{SoftwareError, SoftwareModule};
+    use assert_matches::assert_matches;
+    use json_sm::{SoftwareError, SoftwareModule, SoftwareModuleUpdate};
     use plugin_sm::plugin::{ExternalPluginCommand, Plugin};
     use std::{fs, io::Write, path::PathBuf, str::FromStr};
     use tokio::fs::File;
@@ -212,6 +213,81 @@ mod tests {
         assert_eq!(res, Ok(()));
     }
 
+    #[tokio::test]
+    async fn plugin_get_command_update_list() {
+        // Prepare dummy plugin with .0 which will give specific exit code ==0.
+        let (plugin, _plugin_path) = get_dummy_plugin("test");
+
+        // Create list of modules to perform plugin update-list API call containing valid input.
+        let module1 = SoftwareModule {
+            module_type: Some("test".into()),
+            name: "test1".into(),
+            version: None,
+            url: None,
+            file_path: None,
+        };
+        let module2 = SoftwareModule {
+            module_type: Some("test".into()),
+            name: "test2".into(),
+            version: None,
+            url: None,
+            file_path: None,
+        };
+
+        let mut logger = dev_null().await;
+        // Call plugin update-list via API.
+        let res = plugin
+            .update_list(
+                &vec![
+                    SoftwareModuleUpdate::Install { module: module1 },
+                    SoftwareModuleUpdate::Remove { module: module2 },
+                ],
+                &mut logger,
+            )
+            .await;
+
+        // Expect Ok as plugin should exit with code 0. If Ok, there is no response to assert.
+        assert_matches!(res, Err(SoftwareError::UpdateListNotSupported(_)));
+    }
+
+    // Test validating if the plugin will fall back to `install` and `remove` options if the `update-list` option is not supported
+    #[tokio::test]
+    async fn plugin_command_update_list_fallback() {
+        // Prepare dummy plugin with .0 which will give specific exit code ==0.
+        let (plugin, _plugin_path) = get_dummy_plugin("test");
+
+        // Create list of modules to perform plugin update-list API call containing valid input.
+        let module1 = SoftwareModule {
+            module_type: Some("test".into()),
+            name: "test1".into(),
+            version: None,
+            url: None,
+            file_path: None,
+        };
+        let module2 = SoftwareModule {
+            module_type: Some("test".into()),
+            name: "test2".into(),
+            version: None,
+            url: None,
+            file_path: None,
+        };
+
+        let mut logger = dev_null().await;
+        // Call plugin update-list via API.
+        let errors = plugin
+            .apply_all(
+                vec![
+                    SoftwareModuleUpdate::Install { module: module1 },
+                    SoftwareModuleUpdate::Remove { module: module2 },
+                ],
+                &mut logger,
+            )
+            .await;
+
+        // Expect Ok as plugin should exit with code 0. If Ok, there is no response to assert.
+        assert!(errors.is_empty());
+    }
+
     fn get_dummy_plugin_path() -> PathBuf {
         // Return a path to a dummy plugin in target directory.
         let package_dir = std::env::var("CARGO_MANIFEST_DIR").unwrap();
diff --git a/sm/plugins/tedge_apt_plugin/Cargo.toml b/sm/plugins/tedge_apt_plugin/Cargo.toml
index 5228e741..8831b4d4 100644
--- a/sm/plugins/tedge_apt_plugin/Cargo.toml
+++ b/sm/plugins/tedge_apt_plugin/Cargo.toml
@@ -16,6 +16,8 @@ assets = [
 [dependencies]
 structopt = "0.3"
 thiserror = "1.0"
+csv = "1.1"
+serde = { version = "1", features = ["derive"] }
 
 [dev-dependencies]
 anyhow = "1.0"
diff --git a/sm/plugins/tedge_apt_plugin/src/error.rs b/sm/plugins/tedge_apt_plugin/src/error.rs
index 8e1e9012..07c8399a 100644
--- a/sm/plugins/tedge_apt_plugin/src/error.rs
+++ b/sm/plugins/tedge_apt_plugin/src/error.rs
@@ -11,6 +11,16 @@ pub enum InternalError {
 
     #[error("Parsing Debian package failed for `{file}`")]
     ParsingError { file: String },
+
+    #[error(transparent)]
+    FromCsv(#[from] csv::Error),
+
+    #[error("Validation of {package} failed with version mismatch. Installed version: {installed}, Expected version: {expected}")]
+    VersionMismatch {
+        package: String,
+        installed: String,
+        expected: String,
+    },
 }
 
 impl InternalError {
diff --git a/sm/plugins/tedge_apt_plugin/src/main.rs b/sm/plugins/tedge_apt_plugin/src/main.rs
index b270023e..769e4a64 100644
--- a/sm/plugins/tedge_apt_plugin/src/main.rs
+++ b/sm/plugins/tedge_apt_plugin/src/main.rs
@@ -3,6 +3,8 @@ mod module_check;
 
 use crate::error::InternalError;
 use crate::module_check::PackageMetadata;
+use serde::Deserialize;
+use std::io::{self};
 use std::process::{Command, ExitStatus, Stdio};
 use structopt::StructOpt;
 
@@ -33,6 +35,9 @@ pub enum PluginOp {
         version: Option<String>,
     },
 
+    /// Install or remove multiple modules at once
+    UpdateList,
+
     /// Prepare a sequences of install/remove commands
     Prepare,
 
@@ -40,6 +45,22 @@ pub enum PluginOp {
     Finalize,
 }
 
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "lowercase")]
+enum UpdateAction {
+    Install,
+    Remove,
+}
+#[derive(Debug, Deserialize)]
+struct SoftwareModuleUpdate {
+    pub action: UpdateAction,
+    pub name: String,
+    #[serde(default)]
+    pub version: Option<String>,
+    #[serde(default)]
+    pub path: Option<String>,
+}
+
 fn run(operation: PluginOp) -> Result<ExitStatus, InternalError> {
     let status = match operation {
         PluginOp::List {} => {
@@ -70,42 +91,8 @@ fn run(operation: PluginOp) -> Result<ExitStatus, InternalError> {
             version,
             file_path,
         } => {
-            match (&version, &file_path) {
-                (None, None) => {
-                    // normal install
-                    run_cmd("apt-get", &format!("install --quiet --yes {}", module))?
-                }
-
-                (Some(version), None) => run_cmd(
-                    "apt-get",
-                    &format!("install --quiet --yes {}={}", module, version),
-                )?,
-
-                (None, Some(file_path)) => {
-                    let mut package = PackageMetadata::try_new(file_path)?;
-                    let () = package
-                        .validate_package(&[&format!("Package: {}", &module), "Debian package"])?;
-
-                    run_cmd(
-                        "apt-get",
-                        &format!("install --quiet --yes {}", package.file_path().display()),
-                    )?
-                }
-
-                (Some(version), Some(file_path)) => {
-                    let mut package = PackageMetadata::try_new(file_path)?;
-                    let () = package.validate_package(&[
-                        &format!("Version: {}", &version),
-                        &format!("Package: {}", &module),
-                        "Debian package",
-                    ])?;
-
-                    run_cmd(
-                        "apt-get",
-                        &format!("install --quiet --yes {}", package.file_path().display()),
-                    )?
-                }
-            }
+            let (installer, _metadata) = get_installer(module, version, file_path)?;
+            run_cmd("apt-get", &format!("install --quiet --yes {}", installer))?
         }
 
         PluginOp::Remove { module, version } => {
@@ -120,6 +107,55 @@ fn run(operation: PluginOp) -> Result<ExitStatus, InternalError> {
             }
         }
 
+        PluginOp::UpdateList => {
+            let mut updates: Vec<SoftwareModuleUpdate> = Vec::new();
+            let mut rdr = csv::ReaderBuilder::new()
+                .has_headers(false)
+                .delimiter(b'\t')
+                .from_reader(io::stdin());
+            for result in rdr.deserialize() {
+                updates.push(result?);
+            }
+
+            // Maintaining this metadata list to keep the debian package symlinks until the installation is complete,
+            // which will get cleaned up once it goes out of scope after this block
+            let mut metadata_vec = Vec::new();
+            let mut args: Vec<String> = Vec::new();
+            args.push("install".into());
+            args.push("--quiet".into());
+            args.push("--yes".into());
+            for update_module in updates {
+                match update_module.action {
+                    UpdateAction::Install => {
+                        let (installer, metadata) = get_installer(
+                            update_module.name,
+                            update_module.version,
+                            update_module.path,
+                        )?;
+                        args.push(installer);
+                        metadata_vec.push(metadata);
+                    }
+                    UpdateAction::Remove => {
+                        if let Some(version) = update_module.version {
+                            validate_version(update_module.name.as_str(), version.as_str())?
+                        }
+
+                        // Adding a '-' at the end of the package name like 'rolldice-' instructs apt to treat it as removal
+                        args.push(format!("{}-", update_module.name))
+                    }
+                };
+            }
+
+            println!("apt-get install args: {:?}", args);
+            let status = Command::new("apt-get")
+                .args(args)
+                .stdin(Stdio::null())
+                .status()
+                .map_err(|err| InternalError::exec_error("apt-get", err))?;
+
+            return Ok(status);
+        }
+
         PluginOp::Prepare => run_cmd("apt-get", "update --quiet --yes")?,
 
         PluginOp::Finalize => run_cmd("apt-get", "auto-remove --quiet --yes")?,
@@ -128,6 +164,68 @@ fn run(operation: PluginOp) -> Result<ExitStatus, InternalError> {
     Ok(status)
 }
 
+fn get_installer(
+    module: String,
+    version: Option<String>,
+    file_path: Option<String>,
+) -> Result<(String, Option<PackageMetadata>), InternalError> {
+    match (&version, &file_path) {
+        (None, None) => Ok((module, None)),
+
+        (Some(version), None) => Ok((format!("{}={}", module, version), None)),
+
+        (None, Some(file_path)) => {
+            let mut package = PackageMetadata::try_new(file_path)?;
+            let () =
+                package.validate_package(&[&format!("Package: {}", &module), "Debian package"])?;
+
+            Ok((format!("{}", package.file_path().display()), Some(package)))
+        }
+
+        (Some(version), Some(file_path)) => {
+            let mut package = PackageMetadata::try_new(file_path)?;
+            let () = package.validate_package(&[
+                &format!("Version: {}", &version),
+                &format!("Package: {}", &module),
+                "Debian package",
+            ])?;
+
+            Ok((format!("{}", package.file_path().display()), Some(package)))
+        }
+    }
+}
+
+/// Validate if the provided module version matches the currently installed version
+fn validate_version(module_name: &str, module_version: &str) -> Result<(), InternalError> {
+    // Get the current installed version of the provided package
+    let output = Command::new("apt")
+        .arg("list")
+        .arg("--installed")
+        .arg(module_name)
+        .output()
+        .map_err(|err| InternalError::exec_error("apt-get", err))?;
+
+    let stdout = String::from_utf8(output.stdout)?;
+
+    // Check if the installed version and the provided version match
+    let second_line = stdout.lines().nth(1); //Ignore line 0 which is always 'Listing...'
+    if let Some(package_info) = second_line {
+        if let Some(installed_version) = package_info.split_whitespace().nth(1)
+        // Value at index 0 is the package name
+        {
+            if installed_version != module_version {
+                return Err(InternalError::VersionMismatch {
+                    package: module_name.into(),
+                    installed: installed_version.into(),
+                    expected: module_version.into(),
+                });
+            }
+        }
+    }
+
+    Ok(())
+}
+
 fn run_cmd(cmd: &str, args: &str) -> Result<ExitStatus, InternalError> {
     let args: Vec<&str> = args.split_whitespace().collect();
     let status = Command::new(cmd)