Factorise integration test and stabilise with line processing ✂️

Several lines could be capture captured in same chunk which can cause lambda invocation to be missed.
(hence the serverless hang and travis goes into timeout)

Author: AdrieanKhisbe

tests/serverless-plugins-integration/package-lock.json

@@ -24,6 +24,7 @@
     "aws-sdk": "^2.757.0",
     "lodash": "^4.17.20",
     "minio": "^7.0.16",
+    "pump": "^3.0.0",
     "serverless": "^1.83.0",
     "serverless-offline": "^6.7.0",
     "serverless-offline-dynamodb-streams": "^4.1.0",

tests/serverless-plugins-integration/package.json

@@ -2,6 +2,8 @@ const {Writable} = require('stream');
 const {spawn} = require('child_process');
 const onExit = require('signal-exit');
 const {DynamoDB} = require('aws-sdk');
+const pump = require('pump');
+const {delay, getSplitLinesTransform} = require('./utils');
 
 const client = new DynamoDB({
   region: 'eu-west-1',
@@ -38,9 +40,7 @@ const putItems = () =>
 let setupInProgress = true;
 const populateTables = async () => {
   await unemptyTables();
-  await new Promise(resolve => {
-    setTimeout(resolve, 1200);
-  });
+  await delay(1200);
   setupInProgress = false;
   await putItems();
 };
@@ -56,19 +56,20 @@ const serverless = spawn(
 
 const set = new Set();
 let invocationCount = 0;
-serverless.stdout.pipe(
+pump(
+  serverless.stdout,
+  getSplitLinesTransform(),
   new Writable({
-    write(chunk, enc, cb) {
-      const output = chunk.toString();
-
-      if (/Starting Offline Dynamodb Streams/.test(output)) {
+    objectMode: true,
+    write(line, enc, cb) {
+      if (/Starting Offline Dynamodb Streams/.test(line)) {
         populateTables(); // will run in the background
       }
 
       if (setupInProgress) return cb(); // do not consider lambda executions before we post the real items
 
       const matches = /offline: \(λ: (.*)\) RequestId: .* Duration: .* ms {2}Billed Duration: .* ms/g.exec(
-        output
+        line
       );
 
       if (matches) {

tests/serverless-plugins-integration/test-dynamodb-streams.js

@@ -2,6 +2,8 @@ const {Writable} = require('stream');
 const {spawn} = require('child_process');
 const onExit = require('signal-exit');
 const {Kinesis} = require('aws-sdk');
+const pump = require('pump');
+const {delay, getSplitLinesTransform} = require('./utils');
 
 const client = new Kinesis({
   region: 'eu-west-1',
@@ -11,9 +13,7 @@ const client = new Kinesis({
 });
 
 const putRecords = async () => {
-  await new Promise(resolve => {
-    setTimeout(resolve, 1000);
-  });
+  await delay(1000);
 
   await Promise.all([
     client
@@ -52,19 +52,20 @@ const serverless = spawn('serverless', ['--config', 'serverless.kinesis.yml', 'o
   cwd: __dirname
 });
 
-serverless.stdout.pipe(
+pump(
+  serverless.stdout,
+  getSplitLinesTransform(),
   new Writable({
-    write(chunk, enc, cb) {
-      const output = chunk.toString();
-
-      if (/Starting Offline Kinesis/.test(output)) {
+    objectMode: true,
+    write(line, enc, cb) {
+      if (/Starting Offline Kinesis/.test(line)) {
         putRecords();
       }
 
       this.count =
         (this.count || 0) +
         (
-          output.match(
+          line.match(
             /offline: \(λ: .*\) RequestId: .* Duration: .* ms {2}Billed Duration: .* ms/g
           ) || []
         ).length;

tests/serverless-plugins-integration/test-kinesis.js

@@ -2,6 +2,8 @@ const {Writable} = require('stream');
 const {spawn} = require('child_process');
 const onExit = require('signal-exit');
 const Minio = require('minio');
+const pump = require('pump');
+const {delay, getSplitLinesTransform} = require('./utils');
 
 const client = new Minio.Client({
   region: 'eu-west-1',
@@ -14,9 +16,7 @@ const client = new Minio.Client({
 
 const path = './files/test.txt';
 const uploadFiles = async () => {
-  await new Promise(resolve => {
-    setTimeout(resolve, 1000);
-  });
+  await delay(1000);
 
   await Promise.all([
     client.fPutObject('documents', 'first.txt', path),
@@ -34,19 +34,20 @@ const serverless = spawn('serverless', ['--config', 'serverless.s3.yml', 'offlin
   cwd: __dirname
 });
 
-serverless.stdout.pipe(
+pump(
+  serverless.stdout,
+  getSplitLinesTransform(),
   new Writable({
-    write(chunk, enc, cb) {
-      const output = chunk.toString();
-
-      if (/Starting Offline S3/.test(output)) {
+    objectMode: true,
+    write(line, enc, cb) {
+      if (/Starting Offline S3/.test(line)) {
         uploadFiles();
       }
 
       this.count =
         (this.count || 0) +
         (
-          output.match(
+          line.match(
             /offline: \(λ: .*\) RequestId: .* Duration: .* ms {2}Billed Duration: .* ms/g
           ) || []
         ).length;

tests/serverless-plugins-integration/test-s3.js

@@ -2,6 +2,8 @@ const {Writable} = require('stream');
 const {spawn} = require('child_process');
 const onExit = require('signal-exit');
 const {SQS} = require('aws-sdk');
+const pump = require('pump');
+const {getSplitLinesTransform} = require('./utils');
 
 const client = new SQS({
   region: 'eu-west-1',
@@ -43,12 +45,13 @@ const serverless = spawn(
   }
 );
 
-serverless.stdout.pipe(
+pump(
+  serverless.stdout,
+  getSplitLinesTransform(),
   new Writable({
-    write(chunk, enc, cb) {
-      const output = chunk.toString();
-
-      if (/Starting Offline SQS/.test(output)) {
+    objectMode: true,
+    write(line, enc, cb) {
+      if (/Starting Offline SQS/.test(line)) {
         sendMessages()
           .then(() => console.log('sucessfully send messages'))
           .catch(err => {
@@ -59,7 +62,7 @@ serverless.stdout.pipe(
       this.count =
         (this.count || 0) +
         (
-          output.match(
+          line.match(
             /offline: \(λ: .*\) RequestId: .* Duration: .* ms {2}Billed Duration: .* ms/g
           ) || []
         ).length;

tests/serverless-plugins-integration/test-sqs-autocreate.js

@@ -2,6 +2,8 @@ const {Writable} = require('stream');
 const {spawn} = require('child_process');
 const onExit = require('signal-exit');
 const {SQS} = require('aws-sdk');
+const pump = require('pump');
+const {getSplitLinesTransform} = require('./utils');
 
 const client = new SQS({
   region: 'eu-west-1',
@@ -47,19 +49,20 @@ const serverless = spawn('serverless', ['--config', 'serverless.sqs.yml', 'offli
   cwd: __dirname
 });
 
-serverless.stdout.pipe(
+pump(
+  serverless.stdout,
+  getSplitLinesTransform(),
   new Writable({
-    write(chunk, enc, cb) {
-      const output = chunk.toString();
-
-      if (/Starting Offline SQS/.test(output)) {
+    objectMode: true,
+    write(line, enc, cb) {
+      if (/Starting Offline SQS/.test(line)) {
         sendMessages();
       }
 
       this.count =
         (this.count || 0) +
         (
-          output.match(
+          line.match(
             /offline: \(λ: .*\) RequestId: .* Duration: .* ms {2}Billed Duration: .* ms/g
           ) || []
         ).length;

tests/serverless-plugins-integration/test-sqs.js

@@ -0,0 +1,18 @@
+const {Transform} = require('stream');
+
+const getSplitLinesTransform = () =>
+  new Transform({
+    objectMode: true,
+    transform(chunk, encoding, callback) {
+      const lines = chunk.toString().trim().split('\n');
+      lines.forEach(line => this.push(line));
+      callback();
+    }
+  });
+
+const delay = duration =>
+  new Promise(resolve => {
+    setTimeout(resolve, duration);
+  });
+
+module.exports = {getSplitLinesTransform, delay};