[TT-14446] Add an integration test for Input (http_server) → Output(amqp_1) scenario (#7026)

### **User description**
<details open>
<summary><a href="https://tyktech.atlassian.net/browse/TT-14446"
title="TT-14446" target="_blank">TT-14446</a></summary>
  <br />
  <table>
    <tr>
      <th>Summary</th>
      <td>Investigate flaky amqp integration tests</td>
    </tr>
    <tr>
      <th>Type</th>
      <td>
<img alt="Task"
src="https://tyktech.atlassian.net/rest/api/2/universal_avatar/view/type/issuetype/avatar/10318?size=medium"
/>
        Task
      </td>
    </tr>
    <tr>
      <th>Status</th>
      <td>In Code Review</td>
    </tr>
    <tr>
      <th>Points</th>
      <td>N/A</td>
    </tr>
    <tr>
      <th>Labels</th>
      <td>-</td>
    </tr>
  </table>
</details>
<!--
  do not remove this marker as it will break jira-lint's functionality.
  added_by_jira_lint
-->

---

This PR adds an integration test for `Input (http_server) →
Output(amqp_1)` scenario. See
warpstreamlabs/bento#302 for details.


___

### **PR Type**
Tests, Enhancement


___

### **Description**
- Add integration test for HTTP input to AMQP 1.0 output scenario

- Implement AMQP 1.0 output test logic in test context

- Extend test runner to support AMQP 1.0 output validation

- Ensure message mapping and pipeline processor are included for AMQP
1.0


___



### **Changes walkthrough** 📝
<table><thead><tr><th></th><th align="left">Relevant
files</th></tr></thead><tbody><tr><td><strong>Tests</strong></td><td><table>
<tr>
  <td>
    <details>
<summary><strong>mw_streaming_amqp_test.go</strong><dd><code>Add and
integrate AMQP 1.0 output test in streaming integration
tests</code></dd></summary>
<hr>

gateway/mw_streaming_amqp_test.go

<li>Added <code>testAMQP1Output</code> method for AMQP 1.0 message
validation<br> <li> Extended <code>testTykStreamAMQPIntegration</code>
to handle AMQP 1.0 output<br> <li> Introduced a new integration test for
HTTP input to AMQP 1.0 output<br> <li> Ensured proper session, receiver,
and message handling for AMQP 1.0


</details>


  </td>
<td><a
href="https://github.com/TykTechnologies/tyk/pull/7026/files#diff-3599fefd35a5eddd4f464208edce54054fc3e5045d1694473367e834c331f9dc">+81/-0</a>&nbsp;
&nbsp; </td>

</tr>
</table></td></tr></tr></tbody></table>

___

> <details> <summary> Need help?</summary><li>Type <code>/help how to
...</code> in the comments thread for any questions about PR-Agent
usage.</li><li>Check out the <a
href="https://qodo-merge-docs.qodo.ai/usage-guide/">documentation</a>
for more information.</li></details>

Author: buraksezer

gateway/mw_streaming_amqp_test.go

@@ -310,6 +310,7 @@ func (t *amqpTestContext) testAMQP09Output(expectedMessages [][]byte) {
 			result = append(result, d.Body)
 			if len(result) == len(expectedMessages) {
 				close(done)
+				break
 			}
 		}
 	}()
@@ -322,6 +323,48 @@ func (t *amqpTestContext) testAMQP09Output(expectedMessages [][]byte) {
 	assert.Equal(t.t, expectedMessages, result)
 }
 
+func (t *amqpTestContext) testAMQP1Output(expectedMessages [][]byte) {
+	baseCtx := context.Background()
+	conn, err := amqp1.Dial(baseCtx, t.amqpURL, nil)
+	require.NoErrorf(t.t, err, "Failed to connect to RabbitMQ")
+
+	session, err := conn.NewSession(baseCtx, nil)
+	require.NoErrorf(t.t, err, "Failed to create a session")
+
+	// create a new receiver
+	receiver, err := session.NewReceiver(baseCtx, t.queueName, nil)
+	require.NoErrorf(t.t, err, "Failed to create a receiver")
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	done := make(chan struct{})
+	var result [][]byte
+	go func() {
+		for {
+			msg, receiverErr := receiver.Receive(ctx, nil)
+			if receiverErr != nil {
+				t.t.Logf("Failed to receive message: %v", receiverErr)
+				break
+			}
+
+			result = append(result, msg.GetData())
+			if len(result) == len(expectedMessages) {
+				close(done)
+				break
+			}
+			require.NoError(t.t, receiver.AcceptMessage(ctx, msg))
+		}
+	}()
+
+	select {
+	case <-baseCtx.Done():
+	case <-done:
+	}
+
+	assert.Equal(t.t, expectedMessages, result)
+}
+
 func (t *amqpTestContext) testTykStreamAMQPIntegration() {
 	t.t.Helper()
 
@@ -359,6 +402,8 @@ func (t *amqpTestContext) testTykStreamAMQPIntegration() {
 		testWebsocketOutput(t.t, wsClients, messageToSend, numMessages)
 	} else if t.output == "amqp_0_9" {
 		t.testAMQP09Output(messages)
+	} else if t.output == "amqp_1" {
+		t.testAMQP1Output(messages)
 	} else {
 		require.Fail(t.t, "Invalid output type")
 	}
@@ -527,4 +572,40 @@ streams:
 		testCtx.initializeAMQP09Environment()
 		testCtx.testTykStreamAMQPIntegration()
 	})
+
+	t.Run("Publish messages to http input then consume messages from amqp_1 output", func(t *testing.T) {
+		// We need to add pipeline processor for amqp_1 output,
+		// see https://github.com/warpstreamlabs/bento/issues/302
+		queueName := "test-queue-input-http-amqp-1-output"
+		streamingConfig := fmt.Sprintf(`
+streams:
+  test:
+    input:
+      http_server:
+        path: /post
+        timeout: 1s
+    pipeline:
+      processors:
+      - mapping: |
+          meta = deleted()
+    output:
+      amqp_1:
+        urls: 
+          - %s
+        target_address: %s
+`, amqpURL, queueName)
+		tykStreamingOAS, oasErr := setupOASForStreamAPI(streamingConfig)
+		require.NoError(t, oasErr)
+		apiName := setupStreamingAPIForOAS(t, ts, &tykStreamingOAS)
+		testCtx := &amqpTestContext{
+			t:         t,
+			ts:        ts,
+			apiName:   apiName,
+			queueName: queueName,
+			amqpURL:   amqpURL,
+			input:     "http_server",
+			output:    "amqp_1",
+		}
+		testCtx.testTykStreamAMQPIntegration()
+	})
 }