Added Servicebus batch e2e test

Author: gavin-aguiar

tests/endtoend/servicebus_functions/servicebus_functions_stein/function_app.py

@@ -55,23 +55,53 @@ def servicebus_trigger(msg: func.ServiceBusMessage) -> str:
     return result
 
 
+@app.route(route="put_message_batch")
+@app.service_bus_queue_output(
+    arg_name="msg",
+    connection="AzureWebJobsServiceBusConnectionString",
+    queue_name="testqueuebatch")
+def put_message_batch(req: func.HttpRequest, msg: func.Out[str]):
+    msg.set(req.get_body().decode('utf-8'))
+    return 'OK'
+
+
 @app.service_bus_queue_trigger(
     arg_name="msg",
     connection="AzureWebJobsServiceBusConnectionString",
-    queue_name="testqueue",
-    cardinality=func.Cardinality.MANY)
+    queue_name="testqueuebatch", cardinality="many")
 @app.blob_output(arg_name="$return",
-                 path="python-worker-tests/servicebus-triggered_batch.txt",
+                 path="python-worker-tests/test-servicebus-batch.txt",
                  connection="AzureWebJobsStorage")
-def servicebus_triggered(msg: func.ServiceBusMessage) -> str:
-    return str(msg)
+def servicebus_trigger_batch(msg: func.ServiceBusMessage) -> str:
+    msg = msg[0]
+    print(f"Message ============> {msg}")
+    result = json.dumps({
+        'body': msg.get_body().decode('utf-8'),
+        'content_type': msg.content_type,
+        'delivery_count': msg.delivery_count,
+        'expiration_time': (msg.expiration_time.isoformat() if
+                            msg.expiration_time else None),
+        'label': msg.label,
+        'partition_key': msg.partition_key,
+        'reply_to': msg.reply_to,
+        'reply_to_session_id': msg.reply_to_session_id,
+        'scheduled_enqueue_time': (msg.scheduled_enqueue_time.isoformat() if
+                                   msg.scheduled_enqueue_time else None),
+        'session_id': msg.session_id,
+        'time_to_live': msg.time_to_live,
+        'to': msg.to,
+        'user_properties': msg.user_properties,
+    })
+
+    return result
 
 
-@app.route(route="servicebus_triggered")
+@app.route(route="get_servicebus_triggered_batch")
 @app.blob_input(arg_name="file",
-                path="python-worker-tests/servicebus-triggered_batch.txt",
+                path="python-worker-tests/test-servicebus-batch.txt",
                 connection="AzureWebJobsStorage")
-def get_servicebus_triggered(req: func.HttpRequest,
-                             file: func.InputStream) -> str:
+def get_servicebus_triggered_batch(req: func.HttpRequest,
+                                   file: func.InputStream) -> str:
     return func.HttpResponse(
-        file.read().decode('utf-8'), mimetype='application/json')
+        file.read().decode('utf-8'), mimetype='application/json')
+

tests/endtoend/test_servicebus_functions.py

@@ -47,27 +47,18 @@ def get_script_dir(cls):
 
     @testutils.retryable_test(3, 5)
     def test_servicebus_batch(self):
-        data = str({"value": "2024-01-19T12:50:41.250940Z"})
-        r = self.webhost.request('POST', 'put_message',
+        data = '{"value": "2024-01-19T12:50:41.250940Z"}'
+        r = self.webhost.request('POST', 'put_message_batch',
                                  data=data)
         self.assertEqual(r.status_code, 200)
         self.assertEqual(r.text, 'OK')
 
-        max_retries = 10
-        for try_no in range(max_retries):
-            # wait for trigger to process the queue item
-            time.sleep(1)
+        time.sleep(2)
 
-            try:
-                r = self.webhost.request('GET', 'servicebus_triggered')
-                self.assertEqual(r.status_code, 200)
-                msg = r.json()
-                self.assertEqual(msg['body'], data)
-            except (AssertionError, json.JSONDecodeError):
-                if try_no == max_retries - 1:
-                    raise
-            else:
-                break
+        r = self.webhost.request('GET', 'get_servicebus_triggered_batch')
+        self.assertEqual(r.status_code, 200)
+        msg = r.json()
+        self.assertEqual(msg["body"], data)
 
 
 class TestServiceBusFunctionsSteinGeneric(TestServiceBusFunctions):