platform: add websockets based listener

delivery/template.yaml

@@ -41,10 +41,6 @@ Globals:
         LOG_LEVEL: !Ref LogLevel
 
 
-Conditions:
-  IsNotProd: !Not [!Equals [!Ref Environment, prod]]
-
-
 Resources:
   #########
   # TABLE #
@@ -159,17 +155,4 @@ Resources:
   DeadLetterQueue:
     Type: AWS::CloudFormation::Stack
     Properties:
-      TemplateURL: ../../shared/templates/dlq.yaml
-
-  ############
-  # LISTENER #
-  ############
-  Listener:
-    Type: AWS::CloudFormation::Stack
-    Condition: IsNotProd
-    Properties:
-      TemplateURL: ../../shared/templates/listener.yaml
-      Parameters:
-        Environment: !Ref Environment
-        EventBusName: !Ref EventBusName
-        ServiceName: "delivery"
+      TemplateURL: ../../shared/templates/dlq.yaml

delivery/tests/integ/test_events.py

@@ -45,21 +45,13 @@ def test_table_update_completed(table_name, listener, order):
 
     # Set the item to COMPLETED
     order["status"] = "COMPLETED"
-    table.put_item(Item=order)
-
-    # Listen for messages on EventBridge through a listener SQS queue
-    messages = listener("delivery")
-
-    # Parse messages
-    found = False
-    for message in messages:
-        print("MESSAGE RECEIVED:", message)
-        body = json.loads(message["Body"])
-        if order["orderId"] in body["resources"]:
-            found = True
-            assert body["detail-type"] == "DeliveryCompleted"
 
-    assert found == True
+    # Listen for messages on EventBridge
+    listener(
+        "ecommerce.delivery",
+        lambda: table.put_item(Item=order),
+        lambda m: order["orderId"] in m["resources"] and m["detail-type"] == "DeliveryCompleted"
+    )
 
     # Delete the item
     table.delete_item(Key={"orderId": order["orderId"]})
@@ -76,19 +68,9 @@ def test_table_update_failed(table_name, listener, order):
     order["status"] = "NEW"
     table.put_item(Item=order)
 
-    # Delete the item
-    table.delete_item(Key={"orderId": order["orderId"]})
-
     # Listen for messages on EventBridge through a listener SQS queue
-    messages = listener("delivery")
-
-    # Parse messages
-    found = False
-    for message in messages:
-        print("MESSAGE RECEIVED:", message)
-        body = json.loads(message["Body"])
-        if order["orderId"] in body["resources"]:
-            found = True
-            assert body["detail-type"] == "DeliveryFailed"
-
-    assert found == True
+    listener(
+        "ecommerce.delivery",
+        lambda: table.delete_item(Key={"orderId": order["orderId"]}),
+        lambda m: order["orderId"] in m["resources"] and m["detail-type"] == "DeliveryFailed"
+    )

orders/template.yaml

@@ -53,10 +53,6 @@ Globals:
         LOG_LEVEL: !Ref LogLevel
 
 
-Conditions:
-  IsNotProd: !Not [!Equals [!Ref Environment, prod]]
-
-
 Resources:
   #########
   # TABLE #
@@ -286,19 +282,6 @@ Resources:
       Parameters:
         AlarmAction: !Ref AlarmTopic
 
-  ############
-  # LISTENER #
-  ############
-  Listener:
-    Type: AWS::CloudFormation::Stack
-    Condition: IsNotProd
-    Properties:
-      TemplateURL: ../../shared/templates/listener.yaml
-      Parameters:
-        Environment: !Ref Environment
-        EventBusName: !Ref EventBusName
-        ServiceName: "orders"
-
   #############
   # DASHBOARD #
   #############

orders/tests/integ/test_events.py

@@ -26,66 +26,37 @@ def test_table_update(table_name, listener, order):
     Test that the TableUpdate function reacts to changes to DynamoDB and sends
     events to EventBridge
     """
-    # Add a new item
     table = boto3.resource("dynamodb").Table(table_name) # pylint: disable=no-member
-    table.put_item(Item=order)
 
-    # Listen for messages on EventBridge through a listener SQS queue
-    messages = listener("orders")
-
-    # Parse messages
-    found = False
-    for message in messages:
-        print("MESSAGE RECEIVED:", message)
-        body = json.loads(message["Body"])
-        if order["orderId"] in body["resources"]:
-            found = True
-            assert body["detail-type"] == "OrderCreated"
-
-    assert found == True
-
-    # Change the status to cancelled
-    table.update_item(
-        Key={"orderId": order["orderId"]},
-        UpdateExpression="set #s = :s",
-        ExpressionAttributeNames={
-            "#s": "status"
-        },
-        ExpressionAttributeValues={
-            ":s": "CANCELLED"
-        }
+    # Listen for messages on EventBridge
+    listener(
+        "ecommerce.orders",
+        # Add a new item
+        lambda: table.put_item(Item=order),
+        lambda m: order["orderId"] in m["resources"] and m["detail-type"] == "OrderCreated"
     )
 
-    # Listen for messages on EventBridge through a listener SQS queue
-    messages = listener("orders")
-
-    # Parse messages
-    found = False
-    for message in messages:
-        print("MESSAGE RECEIVED:", message)
-        body = json.loads(message["Body"])
-        if order["orderId"] in body["resources"]:
-            found = True
-            assert body["detail-type"] == "OrderModified"
-            detail = body["detail"]
-            assert "changed" in detail
-            assert "status" in detail["changed"]
-
-    assert found == True
-
-    # Delete the item
-    table.delete_item(Key={"orderId": order["orderId"]})
-
-    # Listen for messages on EventBridge through a listener SQS queue
-    messages = listener("orders")
-
-    # Parse messages
-    found = False
-    for message in messages:
-        print("MESSAGE RECEIVED:", message)
-        body = json.loads(message["Body"])
-        if order["orderId"] in body["resources"]:
-            found = True
-            assert body["detail-type"] == "OrderDeleted"
+    # Listen for messages on EventBridge
+    listener(
+        "ecommerce.orders",
+        # Change the status to cancelled
+        lambda: table.update_item(
+            Key={"orderId": order["orderId"]},
+            UpdateExpression="set #s = :s",
+            ExpressionAttributeNames={
+                "#s": "status"
+            },
+            ExpressionAttributeValues={
+                ":s": "CANCELLED"
+            }
+        ),
+        lambda m: order["orderId"] in m["resources"] and m["detail-type"] == "OrderModified"
+    )
 
-    assert found == True
+    # Listen for messages on EventBridge
+    listener(
+        "ecommerce.orders",
+        # Delete the item
+        lambda: table.delete_item(Key={"orderId": order["orderId"]}),
+        lambda m: order["orderId"] in m["resources"] and m["detail-type"] == "OrderDeleted"
+    )