From 4aa24088f87fca393d391bc1b1babb648b81d4d5 Mon Sep 17 00:00:00 2001
From: Mariusz Nowak <medyk@medikoo.com>
Date: Mon, 13 Jul 2020 11:56:16 +0200
Subject: [PATCH] test(AWS Stream): Improve reliabiity of tests

To avoid race condition as observed here: https://github.com/serverless/serverless/runs/858566550
---
 tests/integration-all/stream/tests.js | 41 ++++++++++++---------------
 1 file changed, 18 insertions(+), 23 deletions(-)

diff --git a/tests/integration-all/stream/tests.js b/tests/integration-all/stream/tests.js
index c03fd7f8b..e6dc729f6 100644
--- a/tests/integration-all/stream/tests.js
+++ b/tests/integration-all/stream/tests.js
@@ -10,13 +10,8 @@ const {
   putKinesisRecord,
 } = require('../../utils/kinesis');
 const { putDynamoDbItem } = require('../../utils/dynamodb');
-const {
-  createTestService,
-  deployService,
-  removeService,
-  waitForFunctionLogs,
-} = require('../../utils/integration');
-const { getMarkers } = require('../shared/utils');
+const { confirmCloudWatchLogs } = require('../../utils/misc');
+const { createTestService, deployService, removeService } = require('../../utils/integration');
 
 describe('AWS - Stream Integration Test', function() {
   this.timeout(1000 * 60 * 100); // Involves time-taking deploys
@@ -70,32 +65,32 @@ describe('AWS - Stream Integration Test', function() {
   describe('Kinesis Streams', () => {
     it('should invoke on kinesis messages from the trim horizon', () => {
       const functionName = 'streamKinesis';
-      const markers = getMarkers(functionName);
       const message = 'Hello from Kinesis!';
 
-      return putKinesisRecord(streamName, message)
-        .then(() => waitForFunctionLogs(tmpDirPath, functionName, markers.start, markers.end))
-        .then(logs => {
-          expect(logs).to.include(functionName);
-          expect(logs).to.include(message);
-          expect(logs).to.include(historicStreamMessage);
-        });
+      return confirmCloudWatchLogs(`/aws/lambda/${stackName}-${functionName}`, () =>
+        putKinesisRecord(streamName, message)
+      ).then(events => {
+        const logs = events.reduce((data, event) => data + event.message, '');
+        expect(logs).to.include(functionName);
+        expect(logs).to.include(message);
+        expect(logs).to.include(historicStreamMessage);
+      });
     });
   });
 
   describe('DynamoDB Streams', () => {
     it('should invoke on dynamodb messages from the latest position', () => {
       const functionName = 'streamDynamoDb';
-      const markers = getMarkers(functionName);
       const item = { id: 'message', hello: 'from dynamo!' };
+      return confirmCloudWatchLogs(`/aws/lambda/${stackName}-${functionName}`, () =>
+        putDynamoDbItem(tableName, item)
+      ).then(events => {
+        const logs = events.reduce((data, event) => data + event.message, '');
 
-      return putDynamoDbItem(tableName, item)
-        .then(() => waitForFunctionLogs(tmpDirPath, functionName, markers.start, markers.end))
-        .then(logs => {
-          expect(logs).to.include(functionName);
-          expect(logs).to.include('INSERT');
-          expect(logs).to.include(item.id);
-        });
+        expect(logs).to.include(functionName);
+        expect(logs).to.include('INSERT');
+        expect(logs).to.include(item.id);
+      });
     });
   });
 });