[pgmoneta#1089] fix the integration testing

Signed-off-by: Amr-Shams <amr.shams2015.as@gmail.com>

Author: Amr-Shams

src/include/workflow.h

@@ -123,6 +123,14 @@ struct workflow
 struct workflow*
 pgmoneta_workflow_create(int workflow_type, struct backup* backup);
 
+/**
+ * Create a verify chain: decrypt -> decompress -> verify
+ * @param backup The backup
+ * @return The workflow chain
+ */
+struct workflow*
+pgmoneta_workflow_create_verify_chain(struct backup* backup);
+
 /**
  * Create standard workflow nodes
  * @param server The server

src/libpgmoneta/s3.c

@@ -366,6 +366,12 @@ pgmoneta_restore_s3_objects(int client_fd, int server, char* prefix, uint8_t com
       goto error;
    }
 
+   if (pgmoneta_art_insert(nodes, USER_IDENTIFIER, (uintptr_t)prefix, ValueString))
+   {
+      ec = MANAGEMENT_ERROR_RESTORE_S3_WORKFLOW;
+      goto error;
+   }
+
    workflow = pgmoneta_workflow_create(WORKFLOW_TYPE_S3_RESTORE, NULL);
 
    if (workflow == NULL)
@@ -381,16 +387,26 @@ pgmoneta_restore_s3_objects(int client_fd, int server, char* prefix, uint8_t com
       goto error;
    }
    pgmoneta_workflow_destroy(workflow);
+   workflow = NULL;
+
    base_dir = pgmoneta_get_server_backup(server);
    local_root = pgmoneta_get_server_backup_identifier(server, prefix);
+
    if (pgmoneta_art_insert(nodes, NODE_TARGET_BASE, (uintptr_t)local_root, ValueString))
    {
       pgmoneta_log_error("S3 restore: could not add backup dir to art");
       goto error;
    }
+
+   if (pgmoneta_art_insert(nodes, NODE_TARGET_ROOT, (uintptr_t)local_root, ValueString))
+   {
+      pgmoneta_log_error("S3 restore: could not add target root to art");
+      goto error;
+   }
+
    if (pgmoneta_art_insert(nodes, USER_FILES, (uintptr_t)NODE_ALL, ValueString))
    {
-      pgmoneta_log_error("S3 restore: could not add base dir to art");
+      pgmoneta_log_error("S3 restore: could not add user files to art");
       goto error;
    }
 
@@ -399,7 +415,8 @@ pgmoneta_restore_s3_objects(int client_fd, int server, char* prefix, uint8_t com
       pgmoneta_log_error("S3 restore: could not load info for %s", config->common.servers[server].name);
       goto error;
    }
-   workflow = pgmoneta_workflow_create(WORKFLOW_TYPE_VERIFY, backup);
+
+   workflow = pgmoneta_workflow_create_verify_chain(backup);
    if (pgmoneta_workflow_execute(workflow, nodes, &en, &ec))
    {
       pgmoneta_log_error("S3 restore: workflow failed for %s", config->common.servers[server].name);
@@ -444,8 +461,8 @@ pgmoneta_restore_s3_objects(int client_fd, int server, char* prefix, uint8_t com
    pgmoneta_art_destroy(nodes);
    pgmoneta_workflow_destroy(workflow);
    free(elapsed);
-   free(base_dir);
    free(backup);
+   free(base_dir);
    if (local_root != NULL)
    {
       pgmoneta_delete_directory(local_root);

src/libpgmoneta/workflow.c

@@ -808,6 +808,96 @@ wf_incremental_backup(void)
    return head;
 }
 
+struct workflow*
+pgmoneta_workflow_create_verify_chain(struct backup* backup)
+{
+   struct workflow* head = NULL;
+   struct workflow* current = NULL;
+
+   if (backup->encryption != ENCRYPTION_NONE)
+   {
+      head = pgmoneta_encryption(false);
+      current = head;
+   }
+
+   switch (COMPRESSION_ALGORITHM(backup->compression))
+   {
+      case COMPRESSION_ALG_GZIP:
+         if (head == NULL)
+         {
+            head = pgmoneta_create_gzip(false);
+            current = head;
+         }
+         else
+         {
+            current->next = pgmoneta_create_gzip(false);
+            current = current->next;
+         }
+         break;
+      case COMPRESSION_ALG_ZSTD:
+         if (head == NULL)
+         {
+            head = pgmoneta_create_zstd(false);
+            current = head;
+         }
+         else
+         {
+            current->next = pgmoneta_create_zstd(false);
+            current = current->next;
+         }
+         break;
+      case COMPRESSION_ALG_LZ4:
+         if (head == NULL)
+         {
+            head = pgmoneta_create_lz4(false);
+            current = head;
+         }
+         else
+         {
+            current->next = pgmoneta_create_lz4(false);
+            current = current->next;
+         }
+         break;
+      case COMPRESSION_ALG_BZIP2:
+         if (head == NULL)
+         {
+            head = pgmoneta_create_bzip2(false);
+            current = head;
+         }
+         else
+         {
+            current->next = pgmoneta_create_bzip2(false);
+            current = current->next;
+         }
+         break;
+   }
+
+   if (head == NULL)
+   {
+      head = pgmoneta_create_verify();
+      current = head;
+   }
+   else
+   {
+      current->next = pgmoneta_create_verify();
+      current = current->next;
+   }
+
+#ifdef DEBUG
+   current = head;
+   while (current != NULL)
+   {
+      assert(current->name != NULL);
+      assert(current->setup != NULL);
+      assert(current->execute != NULL);
+      assert(current->teardown != NULL);
+      current = current->next;
+   }
+#endif
+
+   return head;
+}
+
 static struct workflow*
 wf_verify(struct backup* backup)
 {
@@ -852,18 +942,6 @@ wf_verify(struct backup* backup)
    current->next = pgmoneta_create_verify();
    current = current->next;
 
-#ifdef DEBUG
-   current = head;
-   while (current != NULL)
-   {
-      assert(current->name != NULL);
-      assert(current->setup != NULL);
-      assert(current->execute != NULL);
-      assert(current->teardown != NULL);
-      current = current->next;
-   }
-#endif
-
    return head;
 }