Add lambda test

Signed-off-by: Yves Bastide <[email protected]>
botify-labs · Jun 25, 2017 · 6351703 · 6351703
1 parent 366bd20
commit 6351703
Show file tree

Hide file tree

Showing 7 changed files with 155 additions and 8 deletions.
diff --git a/examples/lambda.py b/examples/lambda.py
@@ -1,7 +1,22 @@
-from simpleflow import Workflow, futures
+from simpleflow import Workflow
 from simpleflow.lambda_function import LambdaFunction
 from simpleflow.swf.task import LambdaFunctionTask
 
+"""
+The lambda function is:
+
+from __future__ import print_function
+
+import json
+
+print('Loading function')
+
+
+def lambda_handler(event, context):
+    print("Received event: " + json.dumps(event, indent=2))
+    return 42
+"""
+
 
 class LambdaWorkflow(Workflow):
     name = 'basic'
@@ -20,4 +35,4 @@ def run(self):
                 foo='bar',
             )
         )
-        print(futures.wait(future))
+        print(future.result)
diff --git a/simpleflow/lambda_function.py b/simpleflow/lambda_function.py
@@ -5,7 +5,10 @@ class LambdaFunction(Submittable):
     def __init__(self,
                  name,
                  start_to_close_timeout=None,
-                 idempotent=None):
+                 idempotent=None,
+                 is_python_function=True,
+                 ):
         self.name = name
         self.start_to_close_timeout = start_to_close_timeout
         self.idempotent = idempotent
+        self.is_python_function = is_python_function
diff --git a/simpleflow/swf/task.py b/simpleflow/swf/task.py
@@ -373,10 +373,13 @@ def name(self):
         return self.lambda_function.name
 
     def schedule(self, *args, **kwargs):
-        input = {
-            'args': self.args,
-            'kwargs': self.kwargs,
-        }
+        if self.lambda_function.is_python_function:
+            input = {
+                'args': self.args,
+                'kwargs': self.kwargs,
+            }
+        else:
+            input = self.kwargs or self.args
 
         decision = swf.models.decision.LambdaFunctionDecision(
             'schedule',

diff --git a/tests/integration/cassettes/test_lambda.yaml b/tests/integration/cassettes/test_lambda.yaml
@@ -0,0 +1,87 @@
+interactions:
+- request:
+    body: '{"domain": "TestDomain", "workflowType": {"name": "basic", "version": "example"}}'
+    headers:
+      Authorization: ['AWS4-HMAC-SHA256 Credential=1234AB/20160823/us-east-1/swf/aws4_request,SignedHeaders=host;x-amz-date;x-amz-target,Signature=foobar']
+      Content-Encoding: [amz-1.0]
+      Content-Length: ['81']
+      Content-Type: [application/json; charset=UTF-8]
+      Host: [swf.us-east-1.amazonaws.com]
+      User-Agent: [Boto/2.47.0 Python/3.6.1 Linux/4.10.0-21-generic]
+      X-Amz-Date: [20170610T183557Z]
+      X-Amz-Target: [com.amazonaws.swf.service.model.SimpleWorkflowService.DescribeWorkflowType]
+    method: POST
+    uri: https://swf.us-east-1.amazonaws.com/
+  response:
+    body: {string: '{"configuration":{"defaultChildPolicy":"TERMINATE","defaultExecutionStartToCloseTimeout":"300","defaultTaskList":{"name":"None"},"defaultTaskStartToCloseTimeout":"300"},"typeInfo":{"creationDate":1.488630704586E9,"status":"REGISTERED","workflowType":{"name":"basic","version":"example"}}}'}
+    headers:
+      Content-Length: ['288']
+      Content-Type: [application/json]
+      x-amzn-RequestId: [a4ff48f5-4e0b-11e7-a675-552ffed790c9]
+    status: {code: 200, message: OK}
+- request:
+    body: '{"domain": "TestDomain", "workflowId": "test-simpleflow-workflow", "workflowType":
+      {"name": "basic", "version": "example"}, "taskList": {"name": "test-simpleflow-workflow-7d4d6b0513764d3ba38c60c4061483ba"},
+      "childPolicy": "TERMINATE", "input": "{\"args\":[],\"kwargs\":{}}", "lambdaRole":
+      "arn:aws:iam::111111000000:role/swf-lambda"}'
+    headers:
+      Authorization: ['AWS4-HMAC-SHA256 Credential=1234AB/20160823/us-east-1/swf/aws4_request,SignedHeaders=host;x-amz-date;x-amz-target,Signature=foobar']
+      Content-Encoding: [amz-1.0]
+      Content-Length: ['333']
+      Content-Type: [application/json; charset=UTF-8]
+      Host: [swf.us-east-1.amazonaws.com]
+      User-Agent: [Boto/2.47.0 Python/3.6.1 Linux/4.10.0-21-generic]
+      X-Amz-Date: [20170610T183557Z]
+      X-Amz-Target: [com.amazonaws.swf.service.model.SimpleWorkflowService.StartWorkflowExecution]
+    method: POST
+    uri: https://swf.us-east-1.amazonaws.com/
+  response:
+    body: {string: '{"runId":"22Yr7TxK3TwETHtk/p2S2dWiEyzwD8e9bjDaOKtBXy1wE="}'}
+    headers:
+      Content-Length: ['58']
+      Content-Type: [application/json]
+      x-amzn-RequestId: [a54bbb7e-4e0b-11e7-818d-3f4a87b50f20]
+    status: {code: 200, message: OK}
+- request:
+    body: '{"domain": "TestDomain", "execution": {"runId": "22Yr7TxK3TwETHtk/p2S2dWiEyzwD8e9bjDaOKtBXy1wE=",
+      "workflowId": "test-simpleflow-workflow"}}'
+    headers:
+      Authorization: ['AWS4-HMAC-SHA256 Credential=1234AB/20160823/us-east-1/swf/aws4_request,SignedHeaders=host;x-amz-date;x-amz-target,Signature=foobar']
+      Content-Encoding: [amz-1.0]
+      Content-Length: ['140']
+      Content-Type: [application/json; charset=UTF-8]
+      Host: [swf.us-east-1.amazonaws.com]
+      User-Agent: [Boto/2.47.0 Python/3.6.1 Linux/4.10.0-21-generic]
+      X-Amz-Date: [20170610T183600Z]
+      X-Amz-Target: [com.amazonaws.swf.service.model.SimpleWorkflowService.DescribeWorkflowExecution]
+    method: POST
+    uri: https://swf.us-east-1.amazonaws.com/
+  response:
+    body: {string: '{"executionConfiguration":{"childPolicy":"TERMINATE","executionStartToCloseTimeout":"300","lambdaRole":"arn:aws:iam::111111000000:role/swf-lambda","taskList":{"name":"test-simpleflow-workflow-7d4d6b0513764d3ba38c60c4061483ba"},"taskStartToCloseTimeout":"300"},"executionInfo":{"cancelRequested":false,"closeStatus":"COMPLETED","closeTimestamp":1.497119759876E9,"execution":{"runId":"22Yr7TxK3TwETHtk/p2S2dWiEyzwD8e9bjDaOKtBXy1wE=","workflowId":"test-simpleflow-workflow"},"executionStatus":"CLOSED","startTimestamp":1.497119758313E9,"workflowType":{"name":"basic","version":"example"}},"openCounts":{"openActivityTasks":0,"openChildWorkflowExecutions":0,"openDecisionTasks":0,"openLambdaFunctions":0,"openTimers":0}}'}
+    headers:
+      Content-Length: ['716']
+      Content-Type: [application/json]
+      x-amzn-RequestId: [a6d36cb2-4e0b-11e7-8e25-3b3c40344394]
+    status: {code: 200, message: OK}
+- request:
+    body: '{"domain": "TestDomain", "execution": {"runId": "22Yr7TxK3TwETHtk/p2S2dWiEyzwD8e9bjDaOKtBXy1wE=",
+      "workflowId": "test-simpleflow-workflow"}}'
+    headers:
+      Authorization: ['AWS4-HMAC-SHA256 Credential=1234AB/20160823/us-east-1/swf/aws4_request,SignedHeaders=host;x-amz-date;x-amz-target,Signature=foobar']
+      Content-Encoding: [amz-1.0]
+      Content-Length: ['140']
+      Content-Type: [application/json; charset=UTF-8]
+      Host: [swf.us-east-1.amazonaws.com]
+      User-Agent: [Boto/2.47.0 Python/3.6.1 Linux/4.10.0-21-generic]
+      X-Amz-Date: [20170610T183701Z]
+      X-Amz-Target: [com.amazonaws.swf.service.model.SimpleWorkflowService.GetWorkflowExecutionHistory]
+    method: POST
+    uri: https://swf.us-east-1.amazonaws.com/
+  response:
+    body: {string: '{"events":[{"eventId":1,"eventTimestamp":1.497119758313E9,"eventType":"WorkflowExecutionStarted","workflowExecutionStartedEventAttributes":{"childPolicy":"TERMINATE","executionStartToCloseTimeout":"300","input":"{\"args\":[],\"kwargs\":{}}","lambdaRole":"arn:aws:iam::111111000000:role/swf-lambda","parentInitiatedEventId":0,"taskList":{"name":"test-simpleflow-workflow-7d4d6b0513764d3ba38c60c4061483ba"},"taskStartToCloseTimeout":"300","workflowType":{"name":"basic","version":"example"}}},{"decisionTaskScheduledEventAttributes":{"startToCloseTimeout":"300","taskList":{"name":"test-simpleflow-workflow-7d4d6b0513764d3ba38c60c4061483ba"}},"eventId":2,"eventTimestamp":1.497119758313E9,"eventType":"DecisionTaskScheduled"},{"decisionTaskStartedEventAttributes":{"identity":"{\"hostname\":\"zeb-P5Q-E\",\"pid\":6614,\"user\":\"zeb\"}","scheduledEventId":2},"eventId":3,"eventTimestamp":1.497119758358E9,"eventType":"DecisionTaskStarted"},{"decisionTaskCompletedEventAttributes":{"scheduledEventId":2,"startedEventId":3},"eventId":4,"eventTimestamp":1.497119758915E9,"eventType":"DecisionTaskCompleted"},{"eventId":5,"eventTimestamp":1.497119758915E9,"eventType":"LambdaFunctionScheduled","lambdaFunctionScheduledEventAttributes":{"decisionTaskCompletedEventId":4,"id":"hello-world-python-5ced6d5d9fae58fd0a7eb0d3908cabed","input":"{\"args\":[8],\"kwargs\":{\"foo\":\"bar\"}}","name":"hello-world-python"}},{"eventId":6,"eventTimestamp":1.497119758941E9,"eventType":"LambdaFunctionStarted","lambdaFunctionStartedEventAttributes":{"scheduledEventId":5}},{"eventId":7,"eventTimestamp":1.497119759328E9,"eventType":"LambdaFunctionCompleted","lambdaFunctionCompletedEventAttributes":{"result":"42","scheduledEventId":5,"startedEventId":6}},{"decisionTaskScheduledEventAttributes":{"startToCloseTimeout":"300","taskList":{"name":"test-simpleflow-workflow-7d4d6b0513764d3ba38c60c4061483ba"}},"eventId":8,"eventTimestamp":1.497119759328E9,"eventType":"DecisionTaskScheduled"},{"decisionTaskStartedEventAttributes":{"identity":"{\"hostname\":\"zeb-P5Q-E\",\"pid\":6613,\"user\":\"zeb\"}","scheduledEventId":8},"eventId":9,"eventTimestamp":1.497119759353E9,"eventType":"DecisionTaskStarted"},{"decisionTaskCompletedEventAttributes":{"scheduledEventId":8,"startedEventId":9},"eventId":10,"eventTimestamp":1.497119759876E9,"eventType":"DecisionTaskCompleted"},{"eventId":11,"eventTimestamp":1.497119759876E9,"eventType":"WorkflowExecutionCompleted","workflowExecutionCompletedEventAttributes":{"decisionTaskCompletedEventId":10,"result":"null"}}]}'}
+    headers:
+      Content-Length: ['2535']
+      Content-Type: [application/json]
+      x-amzn-RequestId: [cb1ece9f-4e0b-11e7-8c8a-29c4fe87cb8e]
+    status: {code: 200, message: OK}
+version: 1
diff --git a/tests/integration/test_lambda.py b/tests/integration/test_lambda.py
@@ -0,0 +1,17 @@
+from sure import expect
+
+from tests.integration import VCRIntegrationTest, vcr
+
+
+class TestLambda(VCRIntegrationTest):
+    @vcr.use_cassette
+    def test_lambda(self):
+        events = self.run_standalone('tests.integration.workflow.LambdaWorkflow')
+        expect(len(events)).should.equal(11)
+        expect(events[0]['workflowExecutionStartedEventAttributes']).should.have.key('lambdaRole')
+        expect(events[4]['eventType']).should.equal('LambdaFunctionScheduled')
+        expect(events[4]['lambdaFunctionScheduledEventAttributes']['name']).should.equal('hello-world-python')
+        expect(events[5]['eventType']).should.equal('LambdaFunctionStarted')
+        expect(events[6]['eventType']).should.equal('LambdaFunctionCompleted')
+        expect(events[6]['lambdaFunctionCompletedEventAttributes']['result']).should.equal('42')
+        print(events)
diff --git a/tests/integration/test_task_list.py b/tests/integration/test_task_list.py
@@ -56,6 +56,7 @@ def test_not_standalone(self):
             input='[]',
             input_file=None,
             local=False,
+            lambda_role=None,
         )
         while True:
             time.sleep(1)

diff --git a/tests/integration/workflow.py b/tests/integration/workflow.py
@@ -10,6 +10,8 @@
 from simpleflow.canvas import Chain, Group
 from simpleflow.constants import HOUR, MINUTE
 from simpleflow.swf.utils import get_workflow_execution
+from simpleflow.lambda_function import LambdaFunction
+from simpleflow.swf.task import LambdaFunctionTask
 from simpleflow.task import ActivityTask
 
 
@@ -154,6 +156,26 @@ def run(self):
         return future.result
 
 
+class LambdaWorkflow(Workflow):
+    name = 'basic'
+    version = 'example'
+    task_list = 'example'
+    lambda_role = 'arn:aws:iam::111111000000:role/swf-lambda'
+
+    def run(self):
+        future = self.submit(
+            LambdaFunctionTask(
+                LambdaFunction(
+                    'hello-world-python',
+                    idempotent=True,
+                ),
+                8,
+                foo='bar',
+            )
+        )
+        print(future.result)
+
+
 class TimerWorkflow(Workflow):
     name = 'example'
     version = 'example'
@@ -190,7 +212,6 @@ def run(self):
         return future.result
 
 
-
 class WorkflowToCancel(Workflow):
     name = 'example'
     version = 'example'