Merge pull request #180 from biomage-ltd/add-authjwt

auth jwt added to error message pipeline fail

pipeline-runner/R/handle_data.R

@@ -103,11 +103,11 @@ send_output_to_api <- function(pipeline_config, input, plot_data_keys, output) {
   return(result$MessageId)
 }
 
-send_gem2s_update_to_api <- function(pipeline_config, experiment_id, task_name, data, auth_JWT) {
+send_gem2s_update_to_api <- function(pipeline_config, experiment_id, task_name, data, input) {
   message("Sending to SNS topic ", pipeline_config$sns_topic)
   sns <- paws::sns(config = pipeline_config$aws_config)
-
-  msg <- c(data, taskName = list(task_name), experimentId = list(experiment_id), authJWT = list(auth_JWT))
+  # TODO -REMOVE DUPLICATE AUTHJWT IN RESPONSE
+  msg <- c(data, taskName = list(task_name), experimentId = list(experiment_id), authJWT = list(input$auth_JWT), input = list(input))
 
   result <- sns$publish(
     Message = RJSONIO::toJSON(msg),
@@ -124,11 +124,17 @@ send_gem2s_update_to_api <- function(pipeline_config, experiment_id, task_name,
   return(result$MessageId)
 }
 
-send_pipeline_fail_update <- function(pipeline_config, experiment_id, process_name, error_message) {
+send_pipeline_fail_update <- function(pipeline_config, input, error_message) {
+  process_name <- input$processName
+
   error_msg <- list()
-  error_msg$experimentId <- experiment_id
-  error_msg$response$error <- error_message
 
+  # TODO - REMOVE THE DUPLICATE EXPERIMETN ID FROM INPUT RESPONSE
+
+  error_msg$experimentId <- input$experimentId
+  error_msg$taskName <- input$taskName
+  error_msg$response$error <- process_name
+  error_msg$input <- input
   sns <- paws::sns(config = pipeline_config$aws_config)
 
   string_value <- ""

pipeline-runner/R/qc-5-filter_doublets.R

@@ -80,7 +80,6 @@ filter_doublets <- function(scdata, config, sample_id, cells_id, task_name = "do
     config = config,
     plotData = guidata
   )
-
   return(result)
 }
 

pipeline-runner/init.R

@@ -174,7 +174,7 @@ call_gem2s <- function(task_name, input, pipeline_config) {
     c(data, task_out) %<-% run_gem2s_step(task_name, input, pipeline_config, prev_out)
     assign("prev_out", task_out, pos = ".GlobalEnv")
 
-    message_id <- send_gem2s_update_to_api(pipeline_config, experiment_id, task_name, data, input$authJWT)
+    message_id <- send_gem2s_update_to_api(pipeline_config, experiment_id, task_name, data, input)
 
     return(message_id)
 }
@@ -358,7 +358,7 @@ init <- function() {
                     cause = error_txt
                 )
 
-                send_pipeline_fail_update(pipeline_config, input_parse$experimentId, input_parse$processName, error_txt)
+                send_pipeline_fail_update(pipeline_config, input_parse, error_txt)
 
                 message("Sent task failure to state machine task: ", taskToken)
                 message("recovered from error:", e$message)

pipeline-runner/renv.lock

@@ -931,6 +931,13 @@
       "Repository": "CRAN",
       "Hash": "fec5f52652d60615fdb3957b3d74324a"
     },
+    "mockery": {
+      "Package": "mockery",
+      "Version": "0.4.2",
+      "Source": "Repository",
+      "Repository": "CRAN",
+      "Hash": "313fa6504824ba5aab9308412135fb5f"
+    },
     "munsell": {
       "Package": "munsell",
       "Version": "0.5.0",

pipeline-runner/tests/testthat/test-handle_data.R

@@ -0,0 +1,26 @@
+
+mock_sns <- function(config) {
+    return(
+        list(publish = function (Message, TopicArn, MessageAttributes) {
+            return (list(MessageId = 'ok'))
+        }
+    ))
+}
+
+test_that("send_gem2s_update_to_api completes successfully", {
+    pipeline_config <- list(
+        sns_topic = 'ExampleTopic',
+        aws_config = NULL
+    )
+
+    mockery::stub(send_gem2s_update_to_api, 'paws::sns', mock_sns)
+
+    response <- send_gem2s_update_to_api(pipeline_config,
+                            experiment_id = 'dfgdfg',
+                            task_name = 'dsfdsdf',
+                            data = 1:5,
+                            input = list(auth_JWT='ayylmao'))
+
+
+    expect_true(response == 'ok')
+})