CUMULUS-436: Add activities support to @cumulus/integration-tests

.eslint-ratchet-high-water-mark

@@ -1 +1 @@
-1348
+1344

CHANGELOG.md

@@ -22,6 +22,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
   - moved the tasks that are not converted to use CMA to `tasks/.not_CMA_compliant`
   - updated paths where necessary
 
+### Added
+- `@cumulus/integration-tests` supports testing the output of an activity-type step in addition to a lambda-type step.
+
 ## [v1.2.0] - 2018-03-20
 
 ### Fixed

packages/integration-tests/index.js

@@ -3,7 +3,7 @@
 const uuidv4 = require('uuid/v4');
 const fs = require('fs-extra');
 const { s3, sfn } = require('@cumulus/common/aws');
-const lambda = require('./lambda');
+const sfnStep = require('./sfnStep');
 
 const executionStatusNumRetries = 20;
 const waitPeriodMs = 5000;
@@ -69,11 +69,13 @@ async function waitForCompletedExecution(executionArn) {
   let statusCheckCount = 0;
 
   // While execution is running, check status on a time interval
+  /* eslint-disable no-await-in-loop */
   while (executionStatus === 'RUNNING' && statusCheckCount < executionStatusNumRetries) {
     await timeout(waitPeriodMs);
     executionStatus = await getExecutionStatus(executionArn);
-    statusCheckCount++;
+    statusCheckCount += 1;
   }
+  /* eslint-enable no-await-in-loop */
 
   if (executionStatus === 'RUNNING' && statusCheckCount >= executionStatusNumRetries) {
     //eslint-disable-next-line max-len
@@ -122,8 +124,7 @@ async function startWorkflowExecution(workflowArn, inputFile) {
  */
 async function executeWorkflow(stackName, bucketName, workflowName, inputFile) {
   const workflowArn = await getWorkflowArn(stackName, bucketName, workflowName);
-  const execution = await startWorkflowExecution(workflowArn, inputFile);
-  const executionArn = execution.executionArn;
+  const { executionArn } = await startWorkflowExecution(workflowArn, inputFile);
 
   console.log(`Executing workflow: ${workflowName}. Execution ARN ${executionArn}`);
 
@@ -161,5 +162,10 @@ async function testWorkflow(stackName, bucketName, workflowName, inputFile) {
 module.exports = {
   testWorkflow,
   executeWorkflow,
-  getLambdaOutput: lambda.getLambdaOutput
+  ActivityStep: sfnStep.ActivityStep,
+  LambdaStep: sfnStep.LambdaStep,
+  /**
+   * @deprecated Since version 1.3. To be deleted version 2.0. sfnStep.LambdaStep.getStepOutput instead.
+   */
+  getLambdaOutput: sfnStep.LambdaStep.getStepOutput
 };

packages/integration-tests/sfnStep.js

@@ -0,0 +1,166 @@
+'use strict';
+
+const { sfn } = require('@cumulus/common/aws');
+
+/**
+ * `SfnStep` provides methods for getting the output of a step within an AWS
+ * Step Function for a specific execution.
+*/
+class SfnStep {
+  /**
+   * `getStartEvent` gets the "start" event for a step, given its schedule event
+   *
+   * @param  {Object} executionHistory - AWS Step Function execution history
+   * @param  {Object} scheduleEvent    - AWS Step Function schedule-type event
+   * @returns {Object}                 - AWS Step Function start-type event
+   */
+  getStartEvent(executionHistory, scheduleEvent) {
+    return executionHistory.events.find((event) => {
+      const isStartEvent = this.startEvents.includes(event.type);
+      const previousEventIsScheduleEvent = event.previousEventId === scheduleEvent.id;
+      return isStartEvent && previousEventIsScheduleEvent;
+    });
+  }
+
+  /**
+   * `getCompletionEvent` gets the "completion" event for a step, given its start event
+   *
+   * @param  {Object} executionHistory - AWS Step Function execution history
+   * @param  {Object} startEvent       - AWS Step Function start-type event
+   * @returns {Object}                 - AWS Step Function completion-type event
+   */
+  getCompletionEvent(executionHistory, startEvent) {
+    return executionHistory.events.find((event) => {
+      const isCompletionEvent = this.completionEvents.includes(event.type);
+      const previousEventIsStartEvent = event.previousEventId === startEvent.id;
+      return isCompletionEvent && previousEventIsStartEvent;
+    });
+  }
+
+  /**
+   * Get the events for the step execution for the given workflow execution.
+   * This function currently assumes one execution of the given step (by step name) per workflow.
+   *
+   * @param {string} workflowExecutionArn - Arn of the workflow execution
+   * @param {string} stepName - name of the step
+   * @returns {Object} an object containing a schedule event, start event, and complete
+   * event if exist, null if cannot find the step
+   */
+  async getStepExecution(workflowExecutionArn, stepName) {
+    const executionHistory = (
+      await sfn().getExecutionHistory({ executionArn: workflowExecutionArn }).promise()
+    );
+
+    // Get the event where the step was scheduled
+    const scheduleEvent = executionHistory.events.find((event) => {
+      const eventScheduled = this.scheduleEvents.includes(event.type);
+      const eventDetails = event[this.eventDetailsKeys.scheduled];
+      const isStepEvent = eventDetails && eventDetails.resource.includes(stepName);
+      return eventScheduled && isStepEvent;
+    });
+
+    if (!scheduleEvent) {
+      console.log(`Could not find step ${stepName} in execution.`);
+      return null;
+    }
+
+    let startEvent = null;
+    let completeEvent = null;
+
+    if (scheduleEvent.type !== this.startFailedEvent) {
+      startEvent = this.getStartEvent(executionHistory, scheduleEvent, this);
+
+      if (startEvent !== null && startEvent.type !== this.startFailedEvent) {
+        completeEvent = this.getCompletionEvent(executionHistory, startEvent, this);
+      }
+    }
+
+    return { scheduleEvent, startEvent, completeEvent };
+  }
+
+  /**
+   * Get the output payload from the step, if the step succeeds
+   *
+   * @param {string} workflowExecutionArn - Arn of the workflow execution
+   * @param {string} stepName - name of the step
+   * @returns {Object} object containing the payload, null if error
+   */
+  async getStepOutput(workflowExecutionArn, stepName) {
+    const stepExecution = await this.getStepExecution(workflowExecutionArn, stepName, this);
+
+    if (stepExecution === null) {
+      console.log(`Could not find step ${stepName} in execution.`);
+      return null;
+    }
+
+    if (stepExecution.completeEvent === null ||
+        stepExecution.completeEvent.type !== this.successEvent) {
+      console.log(`Step ${stepName} was not successful.`);
+      return null;
+    }
+
+    return JSON.parse(stepExecution.completeEvent[this.eventDetailsKeys.succeeded].output.toString());
+  }
+}
+
+/**
+ * `LambdaStep` is a step inside a step function that runs an AWS Lambda function.
+ */
+class LambdaStep extends SfnStep {
+  //eslint-disable-next-line require-jsdoc
+  constructor() {
+    super();
+    this.scheduleFailedEvent = 'LambdaFunctionScheduleFailed';
+    this.scheduleEvents = [
+      this.scheduleFailedEvent,
+      'LambdaFunctionScheduled'
+    ];
+    this.startFailedEvent = 'LambdaFunctionStartFailed';
+    this.startEvents = [
+      this.startFailedEvent,
+      'LambdaFunctionStarted'
+    ];
+    this.successEvent = 'LambdaFunctionSucceeded';
+    this.completionEvents = [
+      this.successEvent,
+      'LambdaFunctionFailed',
+      'LambdaFunctionTimedOut'
+    ];
+    this.eventDetailsKeys = {
+      scheduled: 'lambdaFunctionScheduledEventDetails',
+      succeeded: 'lambdaFunctionSucceededEventDetails'
+    };
+  }
+}
+
+/**
+ * `ActivityStep` is a step inside a step function that runs an AWS ECS activity.
+ */
+class ActivityStep extends SfnStep {
+  //eslint-disable-next-line require-jsdoc
+  constructor() {
+    super();
+    this.scheduleFailedEvent = 'ActivityScheduleFailed';
+    this.scheduleEvents = [
+      'ActivityScheduled',
+      this.scheduleFailedEvent
+    ];
+    this.startEvents = ['ActivityStarted'];
+    this.startFailedEvent = undefined; // there is no 'ActivityStartFailed'
+    this.successEvent = 'ActivitySucceeded';
+    this.completionEvents = [
+      this.successEvent,
+      'ActivityFailed',
+      'ActivityTimedOut'
+    ];
+    this.eventDetailsKeys = {
+      scheduled: 'activityScheduledEventDetails',
+      succeeded: 'activitySucceededEventDetails'
+    };
+  }
+}
+
+module.exports = {
+  ActivityStep,
+  LambdaStep
+};