Introduce test based on setup/teardown infrastructure

package.json

@@ -186,6 +186,8 @@
     "integration-test-run-all": "mocha-isolated --pass-through-aws-creds --skip-fs-cleanup-check --max-workers=20 \"test/integration/**/*.test.js\"",
     "integration-test-run-basic": "mocha test/integrationBasic.test.js",
     "integration-test-run-package": "mocha-isolated --skip-fs-cleanup-check test/integrationPackage/**/*.tests.js",
+    "integration-test-setup-infrastructure": "node ./scripts/test/setup-integration-infra.js",
+    "integration-test-teardown-infrastructure": "node ./scripts/test/teardown-integration-infra.js",
     "lint": "eslint .",
     "lint:updated": "pipe-git-updated --ext=js -- eslint",
     "pkg:build": "node ./scripts/pkg/build.js",

scripts/test/setup-integration-infra.js

@@ -0,0 +1,64 @@
+'use strict';
+
+const awsRequest = require('@serverless/test/aws-request');
+const fs = require('fs');
+const path = require('path');
+const { SHARED_INFRA_TESTS_CLOUDFORMATION_STACK } = require('../../test/utils/cludformation');
+
+(async () => {
+  process.stdout.write('Starting setup of integration infrastructure...\n');
+  const cfnTemplate = fs.readFileSync(path.join(__dirname, 'cloudformation.yml'), 'utf8');
+  const kafkaServerProperties = fs.readFileSync(path.join(__dirname, 'kafka.server.properties'));
+
+  process.stdout.write('Checking if integration tests CloudFormation stack already exists...\n');
+  try {
+    await awsRequest('CloudFormation', 'describeStacks', {
+      StackName: SHARED_INFRA_TESTS_CLOUDFORMATION_STACK,
+    });
+    process.stdout.write('Integration tests CloudFormation stack already exists. Quitting.\n');
+    return;
+  } catch (e) {
+    process.stdout.write('Integration tests CloudFormation does not exist. Continuing.\n');
+  }
+
+  const clusterName = 'integration-tests-msk-cluster';
+  const clusterConfName = 'integration-tests-msk-cluster-configuration';
+
+  process.stdout.write('Creating MSK Cluster configuration...\n');
+  let clusterConfResponse;
+  try {
+    clusterConfResponse = await awsRequest('Kafka', 'createConfiguration', {
+      Name: clusterConfName,
+      ServerProperties: kafkaServerProperties,
+      KafkaVersions: ['2.2.1'],
+    });
+  } catch (e) {
+    process.stdout.write(
+      `Error: ${e} while trying to create MSK Cluster configuration. Quitting. \n`
+    );
+    return;
+  }
+
+  const clusterConfigurationArn = clusterConfResponse.Arn;
+  const clusterConfigurationRevision = clusterConfResponse.LatestRevision.Revision.toString();
+
+  process.stdout.write('Deploying integration tests CloudFormation stack...\n');
+  await awsRequest('CloudFormation', 'createStack', {
+    StackName: SHARED_INFRA_TESTS_CLOUDFORMATION_STACK,
+    TemplateBody: cfnTemplate,
+    Parameters: [
+      { ParameterKey: 'ClusterName', ParameterValue: clusterName },
+      { ParameterKey: 'ClusterConfigurationArn', ParameterValue: clusterConfigurationArn },
+      {
+        ParameterKey: 'ClusterConfigurationRevision',
+        ParameterValue: clusterConfigurationRevision,
+      },
+    ],
+  });
+
+  await awsRequest('CloudFormation', 'waitFor', 'stackCreateComplete', {
+    StackName: SHARED_INFRA_TESTS_CLOUDFORMATION_STACK,
+  });
+  process.stdout.write('Deployed integration tests CloudFormation stack!\n');
+  process.stdout.write('Setup of integration infrastructure finished\n');
+})();

scripts/test/teardown-integration-infra.js

@@ -0,0 +1,55 @@
+'use strict';
+
+const awsRequest = require('@serverless/test/aws-request');
+const {
+  SHARED_INFRA_TESTS_CLOUDFORMATION_STACK,
+  getDependencyStackOutputMap,
+} = require('../../test/utils/cludformation');
+
+(async () => {
+  process.stdout.write('Starting teardown of integration infrastructure...\n');
+  const describeClustersResponse = await awsRequest('Kafka', 'listClusters');
+  const clusterConfArn =
+    describeClustersResponse.ClusterInfoList[0].CurrentBrokerSoftwareInfo.ConfigurationArn;
+
+  const outputMap = await getDependencyStackOutputMap();
+
+  process.stdout.write('Removing leftover ENI...\n');
+  const describeResponse = await awsRequest('EC2', 'describeNetworkInterfaces', {
+    Filters: [
+      {
+        Name: 'vpc-id',
+        Values: [outputMap.VPC],
+      },
+      {
+        Name: 'status',
+        Values: ['available'],
+      },
+    ],
+  });
+  try {
+    await Promise.all(
+      describeResponse.NetworkInterfaces.map(networkInterface =>
+        awsRequest('EC2', 'deleteNetworkInterface', {
+          NetworkInterfaceId: networkInterface.NetworkInterfaceId,
+        })
+      )
+    );
+  } catch (e) {
+    process.stdout.write(`Error: ${e} while trying to remove leftover ENIs\n`);
+  }
+  process.stdout.write('Removing integration tests CloudFormation stack...\n');
+  await awsRequest('CloudFormation', 'deleteStack', {
+    StackName: SHARED_INFRA_TESTS_CLOUDFORMATION_STACK,
+  });
+  await awsRequest('CloudFormation', 'waitFor', 'stackDeleteComplete', {
+    StackName: SHARED_INFRA_TESTS_CLOUDFORMATION_STACK,
+  });
+  process.stdout.write('Removed integration tests CloudFormation stack!\n');
+  process.stdout.write('Removing MSK Cluster configuration...\n');
+  await awsRequest('Kafka', 'deleteConfiguration', {
+    Arn: clusterConfArn,
+  });
+  process.stdout.write('Removed MSK Cluster configuration\n');
+  process.stdout.write('Teardown of integration infrastructure finished\n');
+})();

test/README.md

@@ -60,6 +60,19 @@ Pass test file to Mocha directly as follows
 AWS_ACCESS_KEY_ID=XXX AWS_SECRET_ACCESS_KEY=xxx npx mocha tests/integration/{chosen}.test.js
 ```
 
+### Tests that depend on shared infrastructure stack
+
+Due to the fact that some of the tests require a bit more complex infrastructure setup which might be lengthy, two additional commands has been made available:
+
+- `integration-test-setup-infrastructure` - used for setting up all needed intrastructure dependencies
+- `integration-test-teardown-infrastructure` - used for tearing down the infrastructure setup by the above command
+
+Such tests take advantage of `isDependencyStackAvailable` util to check if all needed dependencies are ready. If not, it skips the given test suite.
+
+Examples of such tests:
+
+- [MSK]('./integration/msk.test.js')
+
 ## Testing templates
 
 If you add a new template or want to test a template after changing it you can run the template integration tests. Make sure you have `docker` and `docker-compose` installed as they are required. The `docker` containers we're using through compose are automatically including your `$HOME/.aws` folder so you can deploy to AWS.

test/integration/msk.test.js

@@ -0,0 +1,101 @@
+'use strict';
+
+const { expect } = require('chai');
+const log = require('log').get('serverless:test');
+const fixtures = require('../fixtures');
+const { confirmCloudWatchLogs } = require('../utils/misc');
+const {
+  isDependencyStackAvailable,
+  getDependencyStackOutputMap,
+} = require('../utils/cludformation');
+
+const awsRequest = require('@serverless/test/aws-request');
+const crypto = require('crypto');
+const { deployService, removeService } = require('../utils/integration');
+
+describe('AWS - MSK Integration Test', function() {
+  this.timeout(1000 * 60 * 100); // Involves time-taking deploys
+  let stackName;
+  let servicePath;
+  const stage = 'dev';
+
+  const topicName = `msk-topic-${crypto.randomBytes(8).toString('hex')}`;
+
+  before(async function beforeHook() {
+    const isDepsStackAvailable = await isDependencyStackAvailable();
+    if (!isDepsStackAvailable) {
+      log.notice(
+        'CloudFormation stack with integration test dependencies not found. Skipping test.'
+      );
+      this.skip();
+    }
+
+    const outputMap = await getDependencyStackOutputMap();
+
+    log.notice('Getting MSK Boostrap Brokers URLs...');
+    const getBootstrapBrokersResponse = await awsRequest('Kafka', 'getBootstrapBrokers', {
+      ClusterArn: outputMap.MSKCluster,
+    });
+    const brokerUrls = getBootstrapBrokersResponse.BootstrapBrokerStringTls;
+
+    const serviceData = await fixtures.setup('functionMsk', {
+      configExt: {
+        functions: {
+          producer: {
+            vpc: {
+              subnetIds: [outputMap.PrivateSubnetA],
+              securityGroupIds: [outputMap.SecurityGroup],
+            },
+            environment: {
+              TOPIC_NAME: topicName,
+              BROKER_URLS: brokerUrls,
+            },
+          },
+          consumer: {
+            events: [
+              {
+                msk: {
+                  arn: outputMap.MSKCluster,
+                  topic: topicName,
+                },
+              },
+            ],
+          },
+        },
+      },
+    });
+
+    ({ servicePath } = serviceData);
+
+    const serviceName = serviceData.serviceConfig.service;
+    stackName = `${serviceName}-${stage}`;
+    log.notice(`Deploying "${stackName}" service...`);
+    await deployService(servicePath);
+  });
+
+  after(async () => {
+    if (servicePath) {
+      log.notice('Removing service...');
+      await removeService(servicePath);
+    }
+  });
+
+  it('correctly processes messages from MSK topic', async () => {
+    const functionName = 'consumer';
+    const message = 'Hello from MSK Integration test!';
+
+    return confirmCloudWatchLogs(
+      `/aws/lambda/${stackName}-${functionName}`,
+      async () =>
+        await awsRequest('Lambda', 'invoke', {
+          FunctionName: `${stackName}-producer`,
+          InvocationType: 'RequestResponse',
+        }),
+      { timeout: 120 * 1000 }
+    ).then(events => {
+      const logs = events.reduce((data, event) => data + event.message, '');
+      expect(logs).to.include(functionName);
+      expect(logs).to.include(message);
+    });
+  });
+});