integration-testing

Enterprise Integration Testing: EDI, Middleware, and API Gateway Testing Strategies

26 Sep 2026 — 11 min read

Enterprise integration layers — EDI, ESBs, message queues, API gateways, and service meshes — are where the most critical business transactions flow, and they are systematically undertested. This guide covers EDI X12/EDIFACT validation, MuleSoft MUnit testing, IBM MQ and RabbitMQ message flow testing, contract testing with Pact, API gateway policy testing with Apigee and Kong, and chaos testing for integration resilience.

Key Takeaways

EDI testing is not optional for supply chain reliability. A malformed 850 Purchase Order or a missing ISA envelope segment can silently fail and delay thousands of shipments. Message queue tests must validate both happy paths and failure modes. Dead letter queue behavior, message TTL expiry, and poison message handling are where integration bugs hide. Contract testing prevents integration regressions without end-to-end test environments. Pact consumer-driven contracts give you confidence that provider changes don't break consumers before deployment.

Enterprise integration is the nervous system of modern business operations. An order placed in an e-commerce platform triggers an 850 EDI transaction to a supplier, a message to an inventory management system via IBM MQ, a REST call through an API gateway to a logistics provider, and a financial transaction through a payment processor. Each of these hops is a potential failure point, and failures are often silent — the order looks fine in the source system while the downstream integration has silently dropped or corrupted the transaction.

Testing these integration layers requires different strategies for each technology in the chain.

Enterprise Integration Patterns and Their Test Surfaces

The Integration Testing Map

Integration Type	Primary Test Surface	Key Failure Modes
EDI (X12, EDIFACT)	Schema validation, acknowledgment flow	Malformed segments, missing mandatory elements, wrong qualifier values
ESB / Middleware	Transformation correctness, routing rules	Wrong mapping, routing to wrong endpoint, header loss
Message queues (IBM MQ, RabbitMQ)	Message delivery, ordering, DLQ behavior	Lost messages, duplicate delivery, poison messages
API gateways	Policy enforcement, rate limiting, auth	Policy bypass, wrong backend routing, auth token passthrough
Service mesh (Istio)	mTLS, retry policies, circuit breaking	Certificate rotation, retry storms, circuit breaker tuning
ETL pipelines	Data transformation, row counts, data quality	Truncation, type coercion errors, encoding issues

EDI Testing

EDI X12 Structure

EDI X12 documents have a hierarchical envelope structure:

ISA*00*          *00*          *ZZ*SENDER123      *ZZ*RECEIVER456    *260528*1200*^*00501*000000001*0*P*>~
GS*PO*SENDER123*RECEIVER456*20260528*1200*1*X*005010~
ST*850*0001~
BEG*00*SA*PO-2026-00001**20260528~
REF*DP*DEPT-001~
ITD*01*3*2**10~
DTM*002*20260615~
N1*BY*ACME CORPORATION*92*ACME001~
N1*SE*SUPPLIER INC*92*SUPP001~
PO1*1*100*EA*9.50**VP*WIDGET-A-001*BP*WA001~
CTT*1~
SE*11*0001~
GE*1*1~
IEA*1*000000001~

EDI Validation with Python edi-tools

from pyx12 import x12n_document, error_handler
from pyx12.params import params

def validate_edi_x12_850(edi_content: str) -> dict:
    """
    Validate an EDI X12 850 Purchase Order.
    Returns validation results including errors and warnings.
    """
    errors = []
    warnings = []

    param = params()
    err_handler = error_handler.errh_null()

    try:
        x12_doc = x12n_document(
            param=param,
            src_file=None,
            src_string=edi_content,
            errh=err_handler
        )

        for seg in x12_doc.iterate_segments():
            # Check ISA envelope
            if seg.id == 'ISA':
                if len(seg.elements) != 16:
                    errors.append("ISA segment must have exactly 16 elements")
                if seg.elements[11].value not in ('00401', '00501'):
                    warnings.append(f"Unusual ISA12 version: {seg.elements[11].value}")

            # Check mandatory BEG segment in 850
            if seg.id == 'BEG':
                if seg.elements[1].value not in ('00', 'SA', 'CN', 'RO'):
                    errors.append(
                        f"BEG02 purpose code '{seg.elements[1].value}' not recognized"
                    )

            # Check PO1 line items
            if seg.id == 'PO1':
                try:
                    qty = float(seg.elements[1].value)
                    if qty <= 0:
                        errors.append(
                            f"PO1 quantity must be positive, got {qty}"
                        )
                    price = float(seg.elements[3].value)
                    if price <= 0:
                        errors.append(
                            f"PO1 unit price must be positive, got {price}"
                        )
                except (ValueError, IndexError) as e:
                    errors.append(f"PO1 parsing error: {e}")

    except Exception as e:
        errors.append(f"EDI parse failure: {e}")

    return {
        'valid': len(errors) == 0,
        'errors': errors,
        'warnings': warnings
    }

def test_valid_850_passes_validation():
    with open('tests/fixtures/edi/850-valid.edi') as f:
        content = f.read()
    result = validate_edi_x12_850(content)
    assert result['valid'], f"Valid 850 failed validation: {result['errors']}"

def test_missing_ben_segment_fails():
    # 850 with BEG segment removed
    with open('tests/fixtures/edi/850-missing-beg.edi') as f:
        content = f.read()
    result = validate_edi_x12_850(content)
    assert not result['valid']
    assert any('BEG' in e for e in result['errors'])

def test_997_acknowledgment_matches_850():
    """Validate that a 997 FA acknowledges the correct ISA control number."""
    isa_control = '000000001'
    with open(f'tests/fixtures/edi/997-{isa_control}.edi') as f:
        ack_content = f.read()

    # Parse 997 AK1 segment for referenced group control
    for line in ack_content.split('~'):
        if line.startswith('AK1'):
            elements = line.split('*')
            referenced_control = elements[2]
            assert referenced_control == isa_control, \
                f"997 references control {referenced_control}, expected {isa_control}"
            return

    raise AssertionError("No AK1 segment found in 997 acknowledgment")

EDIFACT Testing

For European supply chains using EDIFACT, the structure and validation approach are similar but use different segment identifiers:

def validate_edifact_orders(edifact_content: str) -> dict:
    """Validate EDIFACT ORDERS D96A message."""
    errors = []
    segments = edifact_content.split("'")

    unb_found = False
    unh_found = False
    bgm_found = False

    for seg in segments:
        seg = seg.strip()
        if not seg:
            continue

        tag = seg[:3]

        if tag == 'UNB':
            unb_found = True
            parts = seg.split('+')
            if len(parts) < 5:
                errors.append("UNB segment has insufficient elements")

        elif tag == 'UNH':
            unh_found = True
            parts = seg.split('+')
            # UNH+1+ORDERS:D:96A:UN:EAN008
            if len(parts) >= 2:
                msg_ref = parts[2] if len(parts) > 2 else ''
                if not msg_ref.startswith('ORDERS'):
                    errors.append(f"Expected ORDERS message type, got: {msg_ref}")

        elif tag == 'BGM':
            bgm_found = True

    if not unb_found:
        errors.append("Missing UNB interchange header")
    if not unh_found:
        errors.append("Missing UNH message header")
    if not bgm_found:
        errors.append("Missing BGM beginning of message segment")

    return {'valid': len(errors) == 0, 'errors': errors}

MuleSoft Integration Testing with MUnit

MUnit is MuleSoft's native testing framework for Mule 4 applications. Tests are defined in XML and can mock connectors, stub HTTP calls, and assert on message payloads.

MUnit Test Structure

<!-- src/test/munit/order-processing-test.xml -->
<mule xmlns:munit="http://www.mulesoft.org/schema/mule/munit"
      xmlns:munit-tools="http://www.mulesoft.org/schema/mule/munit-tools"
      xmlns:http="http://www.mulesoft.org/schema/mule/http"
      xmlns="http://www.mulesoft.org/schema/mule/core">

    <munit:config name="order-processing-test"/>

    <!-- Test: Valid order routes to fulfillment system -->
    <munit:test name="valid-order-routes-to-fulfillment"
                description="POST /orders with valid payload triggers fulfillment flow">

        <munit:behavior>
            <!-- Mock the fulfillment HTTP call -->
            <munit-tools:mock-when messageProcessor="http:request"
                                   doc:name="Mock Fulfillment API">
                <munit-tools:with-attributes>
                    <munit-tools:with-attribute attributeName="config-ref"
                                               whereValue="Fulfillment_HTTP_Config"/>
                </munit-tools:with-attributes>
                <munit-tools:then-return>
                    <munit-tools:payload value='{"status": "accepted", "fulfillment_id": "FUL-001"}'
                                        mediaType="application/json"/>
                    <munit-tools:attributes>
                        <munit-tools:attribute key="statusCode" value="202"/>
                    </munit-tools:attributes>
                </munit-tools:then-return>
            </munit-tools:mock-when>
        </munit:behavior>

        <munit:execution>
            <munit-tools:http-requester
                path="/api/orders"
                method="POST"
                body='{"order_id": "ORD-001", "customer_id": "CUST-123", "total": 150.00}'
                contentType="application/json"/>
        </munit:execution>

        <munit:validation>
            <munit-tools:assert-that
                expression="#[attributes.statusCode]"
                is="#[MunitTools::equalTo(202)]"/>
            <munit-tools:assert-that
                expression="#[payload.fulfillment_id]"
                is="#[MunitTools::notNullValue()]"/>
        </munit:validation>
    </munit:test>

    <!-- Test: Invalid order rejected with 400 -->
    <munit:test name="invalid-order-missing-customer-returns-400">
        <munit:execution>
            <munit-tools:http-requester
                path="/api/orders"
                method="POST"
                body='{"order_id": "ORD-002", "total": 150.00}'
                contentType="application/json"/>
        </munit:execution>

        <munit:validation>
            <munit-tools:assert-that
                expression="#[attributes.statusCode]"
                is="#[MunitTools::equalTo(400)]"/>
            <munit-tools:assert-that
                expression="#[payload.error]"
                is="#[MunitTools::containsString('customer_id')]"/>
        </munit:validation>
    </munit:test>

    <!-- Test: Transformation correctness -->
    <munit:test name="order-transforms-to-correct-edi-850">
        <munit:execution>
            <flow-ref name="transform-to-edi-850-flow"/>
        </munit:execution>

        <munit:validation>
            <!-- Verify ISA segment present -->
            <munit-tools:assert-that
                expression="#[payload]"
                is="#[MunitTools::containsString('ISA')]"/>
            <!-- Verify BEG segment with SA purpose code -->
            <munit-tools:assert-that
                expression="#[payload]"
                is="#[MunitTools::containsString('BEG*00*SA')]"/>
        </munit:validation>
    </munit:test>
</mule>

Run MUnit tests:

mvn munit:test -Dmunit.test=order-processing-test
# Produces: target/surefire-reports/TEST-order-processing-test.xml

IBM MQ Message Flow Testing

Testing Message Routing and Transformation

import pymqi
import json
import time
import pytest

class MQTestClient:
    def __init__(self, queue_manager, channel, host, port, user, password):
        self.conn_info = f"{host}({port})"
        cd = pymqi.CD()
        cd.ChannelName = channel.encode()
        cd.ConnectionName = self.conn_info.encode()
        cd.ChannelType = pymqi.CMQC.MQCHT_CLNTCONN
        cd.TransportType = pymqi.CMQC.MQXPT_TCP

        sco = pymqi.SCO()
        sco.UserIdentifier = user.encode()
        sco.Password = password.encode()

        self.qmgr = pymqi.QueueManager(None)
        self.qmgr.connectTCPClient(queue_manager, cd, channel, sco=sco)

    def put_message(self, queue_name: str, payload: dict, headers: dict = None):
        q = pymqi.Queue(self.qmgr, queue_name)
        md = pymqi.MD()
        if headers:
            md.CorrelId = headers.get('correlation_id', b' ' * 24)
            md.ReplyToQ = headers.get('reply_to', b' ' * 48).encode() \
                if isinstance(headers.get('reply_to', ''), str) else b' ' * 48
        message = json.dumps(payload).encode('utf-8')
        q.put(message, md)
        q.close()
        return md.MsgId

    def get_message(self, queue_name: str, timeout_ms: int = 5000) -> dict:
        q = pymqi.Queue(self.qmgr, queue_name)
        gmo = pymqi.GMO()
        gmo.Options = pymqi.CMQC.MQGMO_WAIT
        gmo.WaitInterval = timeout_ms
        try:
            message = q.get(None, pymqi.MD(), gmo)
            q.close()
            return json.loads(message.decode('utf-8'))
        except pymqi.MQMIError as e:
            if e.reason == pymqi.CMQC.MQRC_NO_MSG_AVAILABLE:
                return None
            raise

    def get_queue_depth(self, queue_name: str) -> int:
        q = pymqi.Queue(self.qmgr, queue_name, pymqi.CMQC.MQOO_INQUIRE)
        depth = q.inquire(pymqi.CMQC.MQIA_CURRENT_Q_DEPTH)
        q.close()
        return depth

@pytest.fixture
def mq_client():
    client = MQTestClient(
        queue_manager='QM1',
        channel='DEV.APP.SVRCONN',
        host=os.environ['MQ_HOST'],
        port=1414,
        user=os.environ['MQ_USER'],
        password=os.environ['MQ_PASS']
    )
    yield client
    client.qmgr.disconnect()

def test_order_message_routes_to_fulfillment(mq_client):
    """Put an order on INPUT queue, verify it arrives on FULFILLMENT queue."""
    order = {
        'order_id': 'ORD-TEST-001',
        'customer_id': 'CUST-123',
        'items': [{'sku': 'WIDGET-A', 'qty': 10}],
        'total': 95.00
    }

    msg_id = mq_client.put_message('ORDER.INPUT', order)
    time.sleep(2)  # Allow routing to process

    fulfillment_msg = mq_client.get_message('FULFILLMENT.QUEUE', timeout_ms=10000)
    assert fulfillment_msg is not None, "Message did not arrive on FULFILLMENT.QUEUE"
    assert fulfillment_msg['order_id'] == 'ORD-TEST-001'
    assert fulfillment_msg['customer_id'] == 'CUST-123'

def test_malformed_message_goes_to_dlq(mq_client):
    """Malformed JSON should be routed to dead letter queue."""
    q = pymqi.Queue(mq_client.qmgr, 'ORDER.INPUT')
    q.put(b'not valid json {broken')
    q.close()
    time.sleep(3)

    dlq_depth = mq_client.get_queue_depth('SYSTEM.DEAD.LETTER.QUEUE')
    assert dlq_depth > 0, "Malformed message should have gone to DLQ"

def test_message_ttl_expiry(mq_client):
    """Messages that exceed TTL should expire and not be delivered."""
    q = pymqi.Queue(mq_client.qmgr, 'ORDER.INPUT')
    md = pymqi.MD()
    md.Expiry = 1  # 100ms TTL (Expiry is in tenths of a second)
    q.put(json.dumps({'order_id': 'EXPIRED-001'}).encode(), md)
    q.close()

    time.sleep(1)  # Wait for expiry

    msg = mq_client.get_message('FULFILLMENT.QUEUE', timeout_ms=2000)
    assert msg is None or msg.get('order_id') != 'EXPIRED-001', \
        "Expired message should not have been delivered"

Contract Testing with Pact

Pact enables consumer-driven contract testing — the consumer defines what it expects from the provider, and the provider verifies it can fulfill those expectations independently.

Consumer Test (Python)

import pytest
from pact import Consumer, Provider

@pytest.fixture(scope='session')
def pact(request):
    pact = Consumer('OrderService').has_pact_with(
        Provider('InventoryService'),
        host_name='localhost',
        port=8080,
        pact_dir='./pacts'
    )
    pact.start_service()
    yield pact
    pact.stop_service()

def test_get_product_availability(pact):
    """Consumer expects InventoryService to return availability for a SKU."""
    expected_response = {
        'sku': 'WIDGET-A',
        'available_qty': 500,
        'warehouse': 'US-EAST-1',
        'lead_time_days': 2
    }

    (pact
     .given('WIDGET-A is in stock with 500 units')
     .upon_receiving('a request for WIDGET-A availability')
     .with_request(
         method='GET',
         path='/inventory/WIDGET-A',
         headers={'Authorization': 'Bearer token123'}
     )
     .will_respond_with(
         status=200,
         headers={'Content-Type': 'application/json'},
         body=expected_response
     ))

    with pact:
        result = inventory_client.get_availability('WIDGET-A')
        assert result['available_qty'] == 500
        assert result['warehouse'] == 'US-EAST-1'

def test_insufficient_stock_returns_409(pact):
    (pact
     .given('WIDGET-B is out of stock')
     .upon_receiving('a request to reserve 100 units of WIDGET-B')
     .with_request(
         method='POST',
         path='/inventory/reserve',
         body={'sku': 'WIDGET-B', 'qty': 100}
     )
     .will_respond_with(
         status=409,
         body={'error': 'insufficient_stock', 'available': 0}
     ))

    with pact:
        with pytest.raises(InsufficientStockError):
            inventory_client.reserve('WIDGET-B', qty=100)

Provider Verification

import pytest
from pact import Verifier

def test_inventory_service_fulfills_consumer_contracts():
    verifier = Verifier(
        provider='InventoryService',
        provider_base_url='http://localhost:8081'
    )

    output, _ = verifier.verify_with_broker(
        broker_url='https://pact-broker.company.com',
        broker_username=os.environ['PACT_BROKER_USER'],
        broker_password=os.environ['PACT_BROKER_PASS'],
        provider_states_setup_url='http://localhost:8081/pact/provider-states',
        publish_verification_results=True,
        provider_version=os.environ['GIT_COMMIT']
    )

    assert output == 0, "Pact provider verification failed"

API Gateway Policy Testing

Kong Gateway Policy Tests

import requests
import pytest

KONG_ADMIN = os.environ.get('KONG_ADMIN_URL', 'http://localhost:8001')
KONG_PROXY = os.environ.get('KONG_PROXY_URL', 'http://localhost:8000')

def test_rate_limiting_enforced():
    """Verify rate limiting plugin rejects requests exceeding limit."""
    headers = {'apikey': os.environ['TEST_API_KEY']}

    # Make requests up to the limit (assuming 5/minute for test key)
    responses = []
    for i in range(7):
        resp = requests.get(f"{KONG_PROXY}/api/products", headers=headers)
        responses.append(resp.status_code)

    assert 429 in responses, \
        "Rate limiting should have returned 429 before request 7"

def test_jwt_auth_policy_rejects_invalid_token():
    """Requests with invalid JWT should return 401."""
    resp = requests.get(
        f"{KONG_PROXY}/api/orders",
        headers={'Authorization': 'Bearer invalid.jwt.token'}
    )
    assert resp.status_code == 401
    assert 'Unauthorized' in resp.text or 'invalid' in resp.text.lower()

def test_jwt_auth_policy_accepts_valid_token():
    token = generate_test_jwt(
        secret=os.environ['JWT_SECRET'],
        claims={'sub': 'test-user', 'iss': 'test-issuer'}
    )
    resp = requests.get(
        f"{KONG_PROXY}/api/orders",
        headers={'Authorization': f'Bearer {token}'}
    )
    assert resp.status_code in (200, 404), \
        f"Valid JWT should not return auth error, got {resp.status_code}"

def test_request_transformer_adds_required_headers():
    """Verify Kong request transformer adds X-Forwarded-For and correlation ID."""
    resp = requests.get(
        f"{KONG_PROXY}/api/products/WIDGET-A",
        headers={'apikey': os.environ['TEST_API_KEY']}
    )
    # Check that the upstream received correlation ID
    # (assumes upstream echoes headers back in test mode)
    assert 'X-Correlation-ID' in resp.headers or resp.status_code == 200

def test_cors_policy_allows_configured_origins():
    resp = requests.options(
        f"{KONG_PROXY}/api/orders",
        headers={
            'Origin': 'https://app.company.com',
            'Access-Control-Request-Method': 'POST'
        }
    )
    assert resp.status_code == 200
    allowed = resp.headers.get('Access-Control-Allow-Origin', '')
    assert 'app.company.com' in allowed or allowed == '*'

Apigee Policy Testing with Node.js

// test/apigee-policies.test.js
const apigeeint = require('apigee-lint');
const { expect } = require('chai');

describe('Apigee Policy Validation', () => {
    it('should have OAuthV2 VerifyAccessToken on all proxies', async () => {
        const proxies = await apigeeint.listProxies();

        for (const proxy of proxies) {
            const preflowPolicies = await apigeeint.getProxyPreflowPolicies(proxy);
            const hasOAuth = preflowPolicies.some(p =>
                p.type === 'OAuthV2' && p.operation === 'VerifyAccessToken'
            );
            expect(hasOAuth, `Proxy ${proxy} missing OAuthV2 VerifyAccessToken`).to.be.true;
        }
    });

    it('should enforce spike arrest on all public-facing APIs', async () => {
        const proxies = await apigeeint.listProxies({ tag: 'public' });

        for (const proxy of proxies) {
            const policies = await apigeeint.getAllPolicies(proxy);
            const hasSpikeArrest = policies.some(p => p.type === 'SpikeArrest');
            expect(hasSpikeArrest, `Proxy ${proxy} missing SpikeArrest policy`).to.be.true;
        }
    });
});

Chaos Testing for Integration Layers

Integration chaos tests verify your system behaves correctly when dependencies fail:

import requests
import pytest
from chaos_proxy import ChaosProxy  # hypothetical chaos injection library

@pytest.fixture
def chaos_proxy():
    """Start a chaos proxy in front of the inventory service."""
    proxy = ChaosProxy(
        upstream='http://inventory-service:8080',
        proxy_port=9090
    )
    proxy.start()
    yield proxy
    proxy.stop()

def test_order_service_handles_inventory_timeout(chaos_proxy):
    """Order service should queue orders when inventory is slow, not fail."""
    chaos_proxy.inject_latency(delay_ms=5000, probability=1.0)

    # Update order service to use chaos proxy
    resp = requests.post(
        'http://order-service/api/orders',
        json={'customer_id': 'CUST-001', 'items': [{'sku': 'WIDGET-A', 'qty': 1}]},
        headers={'X-Inventory-URL': 'http://localhost:9090'}
    )

    # Order should be accepted and queued, not failed
    assert resp.status_code in (200, 202), \
        f"Order service should accept orders during inventory slowness, got {resp.status_code}"
    assert resp.json().get('status') in ('queued', 'pending', 'accepted')

def test_circuit_breaker_opens_after_repeated_failures(chaos_proxy):
    """After N failures, circuit breaker should open and fail fast."""
    chaos_proxy.inject_error(status_code=503, probability=1.0)

    # Trigger enough failures to open circuit breaker
    response_times = []
    for i in range(15):
        start = time.time()
        requests.post(
            'http://order-service/api/orders',
            json={'customer_id': 'CUST-001', 'items': [{'sku': 'WIDGET-A', 'qty': 1}]},
            headers={'X-Inventory-URL': 'http://localhost:9090'}
        )
        response_times.append(time.time() - start)

    # After circuit opens, responses should be fast (fail immediately, not wait for timeout)
    last_5_avg = sum(response_times[-5:]) / 5
    first_5_avg = sum(response_times[:5]) / 5

    assert last_5_avg < first_5_avg * 0.5, \
        f"Circuit breaker should reduce response time after opening. " \
        f"First 5 avg: {first_5_avg:.2f}s, Last 5 avg: {last_5_avg:.2f}s"

Integration Testing Environment Strategy

Environment	Purpose	Refresh Cadence
Developer sandbox	Fast feedback, unit + component tests	On-demand
Integration environment	Cross-system tests, contract verification	Per PR
Staging	Full end-to-end, performance, chaos	Nightly
Production shadow	Shadow mode comparison (strangler fig only)	Continuous

Try HelpMeTest

Enterprise integration bugs are often silent — a message sits in a dead letter queue for hours before anyone notices. HelpMeTest can monitor your integration health continuously: message queue depths, API gateway error rates, EDI acknowledgment latency. You define the thresholds; HelpMeTest alerts you the moment something drifts. Visit https://helpmetest.com to set up integration monitoring without writing custom alerting code.