update firecrest api, fix coman squash downloading, fix tests

Author: Panaetius

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "coman"
-version = "0.5.0"
+version = "0.5.1"
 edition = "2024"
 description = "Compute Manager for managing HPC compute"
 authors = ["Ralf Grubenmann <ralf.grubenmann@sdsc.ethz.ch>"]
@@ -94,4 +94,3 @@ current_dir = "0.1.2"
 injectorpp = "0.4.0"
 rstest = "0.26.1"
 tempfile = "3.24.0"
-tmp_env = "0.1.1"

coman/Cargo.toml

@@ -449,7 +449,6 @@ mod tests {
         assert_eq!(layer.data.to_string(), "[cscs]\nvalue=10\nother_value = 20\n");
     }
 
-    #[cfg(target_family = "unix")]
     #[test]
     fn test_config_read_write() {
         let project_dir = tempdir().expect("couldn't create temp dir");
@@ -474,8 +473,33 @@ mod tests {
         std::fs::create_dir_all(global_config.parent().unwrap()).expect("couldn't create config dir");
         std::fs::write(&global_config, "[cscs]\ncurrent_system = \"global\"").expect("couldn't create config file");
 
-        let _tmp_env = tmp_env::set_var("HOME", home_dir.path().as_os_str());
-        let mut conf = Config::new().expect("couldn't load config");
+        let default_layer: DocumentMut = DEFAULT_CONFIG_TOML.parse().expect("couldn't parse default config");
+        let global_layer = Layer::from_path(global_config).expect("couldn't create global layer");
+        let project_layer = Layer::from_path(project_config).expect("couldn't create project layer");
+        let mut builder =
+            config::Config::builder().add_source(config::File::from_str(DEFAULT_CONFIG_TOML, config::FileFormat::Toml));
+        builder = builder
+            .add_source(config::File::from_str(
+                &global_layer.data.to_string(),
+                config::FileFormat::Toml,
+            ))
+            .add_source(config::File::from_str(
+                &project_layer.data.to_string(),
+                config::FileFormat::Toml,
+            ));
+
+        let cfg: ComanConfig = builder
+            .build()
+            .expect("couldn't build")
+            .try_deserialize()
+            .expect("couldn't deserialize");
+        let mut conf = Config {
+            values: cfg,
+            default_layer,
+            global_layer,
+            project_layer: Some(project_layer),
+        };
+
         assert_eq!(conf.values.cscs.current_system, "project");
         conf.set("cscs.current_system", "global2", true)
             .expect("couldn't set global config value");

coman/src/config.rs

@@ -9,9 +9,9 @@ use firecrest_client::{
         get_compute_system_jobs, post_compute_system_job,
     },
     filesystem_api::{
-        get_filesystem_ops_download, get_filesystem_ops_ls, get_filesystem_ops_stat, get_filesystem_ops_tail,
-        post_filesystem_ops_mkdir, post_filesystem_ops_upload, post_filesystem_transfer_download,
-        post_filesystem_transfer_upload, put_filesystem_ops_chmod,
+        delete_filesystem_ops_rm, get_filesystem_ops_download, get_filesystem_ops_ls, get_filesystem_ops_stat,
+        get_filesystem_ops_tail, post_filesystem_ops_mkdir, post_filesystem_ops_upload,
+        post_filesystem_transfer_download, post_filesystem_transfer_upload, put_filesystem_ops_chmod,
     },
     status_api::{get_status_systems, get_status_userinfo},
     types::DownloadFileResponseTransferDirectives,
@@ -161,6 +161,12 @@ impl CscsApi {
             .wrap_err("couldn't change directory permission")?;
         Ok(())
     }
+    pub async fn rm_path(&self, system_name: &str, path: PathBuf) -> Result<()> {
+        delete_filesystem_ops_rm(&self.client, system_name, path)
+            .await
+            .wrap_err("couldn't remove remote file")?;
+        Ok(())
+    }
     pub async fn upload(&self, system_name: &str, target: PathBuf, file: Vec<u8>) -> Result<()> {
         post_filesystem_ops_upload(&self.client, system_name, target, file)
             .await
@@ -242,8 +248,8 @@ impl CscsApi {
 mod tests {
     use claim::*;
     use firecrest_client::types::{
-        DownloadFileResponse, FirecrestFilesystemTransferModelsTransferJob, GetJobMetadataResponse, GetJobResponse,
-        GetSystemsResponse, HPCCluster, JobMetadataModel, JobModel, JobStatus, PostJobSubmissionResponse,
+        DownloadFileResponse, GetJobMetadataResponse, GetJobResponse, GetSystemsResponse, HPCCluster, JobMetadataModel,
+        JobModel, JobStatus, LibDatatransfersDatatransferBaseTransferJob, PostJobSubmissionResponse,
         S3TransferResponse, UploadFileResponse,
     };
     use injectorpp::interface::injector::*;
@@ -459,7 +465,7 @@ mod tests {
             ))
             .will_return_async(injectorpp::async_return!(
                 Ok(UploadFileResponse {
-                    transfer_job: FirecrestFilesystemTransferModelsTransferJob {
+                    transfer_job: LibDatatransfersDatatransferBaseTransferJob {
                         job_id: 1,
                         ..Default::default()
                     },
@@ -493,7 +499,7 @@ mod tests {
             ))
             .will_return_async(injectorpp::async_return!(
                 Ok(DownloadFileResponse {
-                    transfer_job: FirecrestFilesystemTransferModelsTransferJob {
+                    transfer_job: LibDatatransfersDatatransferBaseTransferJob {
                         job_id: 1,
                         ..Default::default()
                     },

coman/src/cscs/api_client/client.rs

@@ -382,12 +382,17 @@ async fn inject_coman_squash(
     #[cfg(target_family = "windows")]
     let size = file_meta.file_size() as usize;
 
-    let existing = api_client.list_path(current_system, target.clone()).await?;
-    if !existing.is_empty() {
+    let response = api_client.list_path(current_system, target.clone()).await;
+    if let Ok(existing) = response
+        && !existing.is_empty()
+    {
         //squash file already present on remote, don't upload if it's the same
         let entry = existing.first().unwrap();
         if entry.size.unwrap_or_default() == size {
             return Ok(Some(target));
+        } else {
+            // remove file before upload
+            api_client.rm_path(current_system, target.clone()).await?;
         }
     }
     //upload squash file
@@ -421,7 +426,10 @@ async fn inject_coman_squash(
         .map(|(i, etag)| (i + 1, etag))
         .map(|(i, etag)| format!("<Part><PartNumber>{}</PartNumber><ETag>{}</ETag></Part>", i, etag))
         .join("");
-    let body = format!("<CompleteMultipartUpload>{}</CompleteMultipartUpload>", body);
+    let body = format!(
+        "<CompleteMultipartUpload xmlns=\"http://s3.amazonaws.com/doc/2006-03-01\">{}</CompleteMultipartUpload>",
+        body
+    );
     let req = client.post(transfer_data.1.complete_upload_url).body(body).build()?;
     let resp = client.execute(req).await?;
     resp.error_for_status()?;

coman/src/cscs/handlers.rs

@@ -227,3 +227,14 @@ pub async fn post_filesystem_transfer_download(
     model.transfer_directives = transfer_dir;
     Ok(model)
 }
+
+pub async fn delete_filesystem_ops_rm(client: &FirecrestClient, system_name: &str, path: PathBuf) -> Result<()> {
+    let path = path.as_os_str().to_str().ok_or(eyre!("couldn't cast path to string"))?;
+    let _ = client
+        .delete(
+            format!("filesystem/{system_name}/ops/rm").as_str(),
+            Some(vec![("path", path)]),
+        )
+        .await?;
+    Ok(())
+}

firecrest_client/src/filesystem_api.rs

@@ -9,7 +9,7 @@
 //! AUTO-GENERATED CODE - DO NOT EDIT!
 //!
 //! FirecREST
-//! Version: 2.4.1
+//! Version: 2.4.2
 //! Generated by `oas3-gen v0.21.1`
 //!
 //! No description provided
@@ -178,7 +178,7 @@ pub struct CompressRequest {
 #[derive(Debug, Clone, PartialEq, Deserialize, oas3_gen_support::Default)]
 pub struct CompressResponse {
     #[serde(rename = "transferJob")]
-    pub transfer_job: FirecrestFilesystemTransferModelsTransferJob,
+    pub transfer_job: LibDatatransfersDatatransferBaseTransferJob,
 }
 #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, oas3_gen_support::Default)]
 pub enum CompressionType {
@@ -219,7 +219,7 @@ pub struct CopyRequest {
 #[derive(Debug, Clone, PartialEq, Deserialize, oas3_gen_support::Default)]
 pub struct CopyResponse {
     #[serde(rename = "transferJob")]
-    pub transfer_job: FirecrestFilesystemTransferModelsTransferJob,
+    pub transfer_job: LibDatatransfersDatatransferBaseTransferJob,
 }
 #[derive(Debug, Clone, PartialEq, Deserialize, oas3_gen_support::Default)]
 #[serde(default)]
@@ -278,7 +278,7 @@ impl DeleteJobCancelComputeSystemNameJobsJobIdDeleteRequest {
 #[derive(Debug, Clone, PartialEq, Deserialize, oas3_gen_support::Default)]
 pub struct DeleteResponse {
     #[serde(rename = "transferJob")]
-    pub transfer_job: FirecrestFilesystemTransferModelsTransferJob,
+    pub transfer_job: LibDatatransfersDatatransferBaseTransferJob,
 }
 ///Delete file or directory operation (`rm`)
 #[derive(Debug, Clone, validator::Validate, oas3_gen_support::Default)]
@@ -421,7 +421,7 @@ pub struct DownloadFileResponse {
     #[serde(rename = "transferDirectives")]
     pub transfer_directives: DownloadFileResponseTransferDirectives,
     #[serde(rename = "transferJob")]
-    pub transfer_job: FirecrestFilesystemTransferModelsTransferJob,
+    pub transfer_job: LibDatatransfersDatatransferBaseTransferJob,
 }
 ///Data transfer parameters specific to the transfer method
 #[derive(Debug, Clone, PartialEq, Deserialize, oas3_gen_support::Default)]
@@ -519,7 +519,7 @@ pub struct ExtractRequest {
 #[derive(Debug, Clone, PartialEq, Deserialize, oas3_gen_support::Default)]
 pub struct ExtractResponse {
     #[serde(rename = "transferJob")]
-    pub transfer_job: FirecrestFilesystemTransferModelsTransferJob,
+    pub transfer_job: LibDatatransfersDatatransferBaseTransferJob,
 }
 #[derive(Debug, Clone, PartialEq, Deserialize, oas3_gen_support::Default)]
 pub struct File {
@@ -617,22 +617,6 @@ pub struct FilesystemServiceHealth {
     #[serde(rename = "serviceType")]
     pub service_type: String,
 }
-#[derive(Debug, Clone, PartialEq, Deserialize, oas3_gen_support::Default)]
-pub struct FirecrestFilesystemTransferModelsTransferJob {
-    #[serde(rename = "jobId")]
-    pub job_id: i64,
-    pub logs: FirecrestFilesystemTransferModelsTransferJobLogs,
-    pub system: String,
-    #[serde(rename = "workingDirectory")]
-    pub working_directory: String,
-}
-#[derive(Debug, Clone, PartialEq, Deserialize, oas3_gen_support::Default)]
-pub struct FirecrestFilesystemTransferModelsTransferJobLogs {
-    #[serde(rename = "errorLog")]
-    pub error_log: String,
-    #[serde(rename = "outputLog")]
-    pub output_log: String,
-}
 ///Output the checksum of a file (using SHA-256 algotithm)
 #[derive(Debug, Clone, validator::Validate, oas3_gen_support::Default)]
 pub struct GetChecksumFilesystemSystemNameOpsChecksumGetRequest {
@@ -1059,6 +1043,9 @@ pub struct GetJobsComputeSystemNameJobsGetRequest {
     /// - Location: `Query`
     #[default(Some(false))]
     pub allusers: Option<bool>,
+    ///If specified, filter jobs by account name
+    /// - Location: `Query`
+    pub account: Option<String>,
 }
 impl GetJobsComputeSystemNameJobsGetRequest {
     ///Render the request path with parameters.
@@ -1078,6 +1065,15 @@ impl GetJobsComputeSystemNameJobsGetRequest {
             )
             .unwrap();
         }
+        if let Some(value) = &self.account {
+            prefix = if prefix == '\0' { '?' } else { '&' };
+            write!(
+                &mut path,
+                "{prefix}account={}",
+                oas3_gen_support::percent_encode_query_component(&oas3_gen_support::serialize_query_param(value)?)
+            )
+            .unwrap();
+        }
         Ok(path)
     }
     ///Parse the HTTP response into the response enum.
@@ -1945,6 +1941,22 @@ pub struct JobTime {
     pub start: Option<i64>,
     pub suspended: Option<i64>,
 }
+#[derive(Debug, Clone, PartialEq, Deserialize, oas3_gen_support::Default)]
+pub struct LibDatatransfersDatatransferBaseTransferJob {
+    #[serde(rename = "jobId")]
+    pub job_id: i64,
+    pub logs: LibDatatransfersDatatransferBaseTransferJobLogs,
+    pub system: String,
+    #[serde(rename = "workingDirectory")]
+    pub working_directory: String,
+}
+#[derive(Debug, Clone, PartialEq, Deserialize, oas3_gen_support::Default)]
+pub struct LibDatatransfersDatatransferBaseTransferJobLogs {
+    #[serde(rename = "errorLog")]
+    pub error_log: String,
+    #[serde(rename = "outputLog")]
+    pub output_log: String,
+}
 ///Create move file or directory operation (`mv`) (for files larger than 5242880 Bytes)
 #[derive(Debug, Clone, validator::Validate, oas3_gen_support::Default)]
 pub struct MoveMvFilesystemSystemNameTransferMvPostRequest {
@@ -2010,7 +2022,7 @@ pub struct MoveRequest {
 #[derive(Debug, Clone, PartialEq, Deserialize, oas3_gen_support::Default)]
 pub struct MoveResponse {
     #[serde(rename = "transferJob")]
-    pub transfer_job: FirecrestFilesystemTransferModelsTransferJob,
+    pub transfer_job: LibDatatransfersDatatransferBaseTransferJob,
 }
 ///Configuration for multipart upload behavior.
 #[derive(Debug, Clone, PartialEq, Deserialize, oas3_gen_support::Default)]
@@ -2951,6 +2963,10 @@ pub struct Scheduler {
     ///Scheduler API version.
     #[serde(rename = "apiVersion")]
     pub api_version: Option<String>,
+    ///Scheduler connection mode.
+    #[serde(rename = "connectionMode")]
+    #[default(Some(Default::default()))]
+    pub connection_mode: Option<SchedulerConnectionMode>,
     ///Timeout in seconds for scheduler communication with the API.
     #[default(Some(10i64))]
     pub timeout: Option<i64>,
@@ -2960,6 +2976,17 @@ pub struct Scheduler {
     ///Scheduler version.
     pub version: String,
 }
+///Modes to connect to the schedulers present in the system
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Deserialize, oas3_gen_support::Default)]
+pub enum SchedulerConnectionMode {
+    #[serde(rename = "hybrid")]
+    #[default]
+    Hybrid,
+    #[serde(rename = "rest")]
+    Rest,
+    #[serde(rename = "ssh")]
+    Ssh,
+}
 ///Health check result for the job scheduler.
 #[derive(Debug, Clone, PartialEq, Deserialize, oas3_gen_support::Default)]
 #[serde(default)]
@@ -3044,7 +3071,7 @@ pub struct UploadFileResponse {
     #[serde(rename = "transferDirectives")]
     pub transfer_directives: DownloadFileResponseTransferDirectives,
     #[serde(rename = "transferJob")]
-    pub transfer_job: FirecrestFilesystemTransferModelsTransferJob,
+    pub transfer_job: LibDatatransfersDatatransferBaseTransferJob,
 }
 #[derive(Debug, Clone, PartialEq, Deserialize, oas3_gen_support::Default)]
 pub struct UserInfoResponse {