Part 7: Testing, Performance, and Integration Patterns

Testing SOAP Services

Testing SOAP services is not fundamentally different from testing any other service — you need unit tests for your service logic, integration tests that use real SOAP calls, and a way to mock the SOAP layer when testing code that depends on an external SOAP service.

What is different with SOAP is that the XML format creates additional things to verify: are the types correct, does the fault structure match the contract, does the WSDL accurately reflect the service?

This part covers all of these and adds performance and integration patterns for using SOAP in a modern Python architecture.

Unit Testing spyne Services

spyne service methods are plain Python — they receive typed arguments and return typed objects. Test them directly without any SOAP or HTTP overhead.

Project Structure for Tests

soap_inventory/
├── models.py
├── service.py
├── faults.py
├── app.py
└── server.py
tests/
├── conftest.py
├── unit/
│   ├── test_service.py
│   └── test_faults.py
├── integration/
│   └── test_soap_client.py
└── fixtures/
    ├── soap_fault_product_not_found.xml
    └── soap_response_get_product.xml

Unit Tests for Service Methods

# tests/unit/test_service.py
import pytest
from decimal import Decimal
from unittest.mock import patch

from spyne.error import ResourceNotFoundError
from soap_inventory.service import InventoryService, _PRODUCTS
from soap_inventory.models import Product, ProductInput
from soap_inventory.faults import ProductNotFoundFault, InvalidStockCountFault


@pytest.fixture(autouse=True)
def reset_products():
    """Reset the in-memory product store before each test."""
    original = dict(_PRODUCTS)
    yield
    _PRODUCTS.clear()
    _PRODUCTS.update(original)


class FakeContext:
    """Minimal mock for spyne's method context (ctx)."""
    pass


class TestGetProduct:
    def test_returns_product_when_found(self):
        ctx = FakeContext()
        result = InventoryService.GetProduct(ctx, "SKU-001")

        assert result.ProductId == "SKU-001"
        assert result.Name == "Widget Pro"
        assert result.Price == Decimal("29.99")
        assert result.InStock is True

    def test_raises_fault_when_not_found(self):
        ctx = FakeContext()
        with pytest.raises(ProductNotFoundFault) as exc_info:
            InventoryService.GetProduct(ctx, "SKU-NONEXISTENT")

        fault = exc_info.value
        assert fault.faultcode == "Client.ProductNotFound"
        assert "SKU-NONEXISTENT" in fault.faultstring


class TestCreateProduct:
    def test_creates_new_product(self):
        ctx = FakeContext()
        product_input = ProductInput(
            ProductId="SKU-NEW",
            Name="New Product",
            Price=Decimal("19.99"),
            InStock=True,
        )
        result = InventoryService.CreateProduct(ctx, product_input)

        assert result.ProductId == "SKU-NEW"
        assert result.Name == "New Product"
        assert result.Price == Decimal("19.99")
        assert result.StockCount == 0

    def test_raises_when_product_already_exists(self):
        ctx = FakeContext()
        product_input = ProductInput(
            ProductId="SKU-001",  # Already exists
            Name="Duplicate",
            Price=Decimal("1.00"),
        )
        with pytest.raises(Exception):
            InventoryService.CreateProduct(ctx, product_input)


class TestUpdateStock:
    def test_updates_stock_count(self):
        ctx = FakeContext()
        result = InventoryService.UpdateStock(ctx, "SKU-001", 200)

        assert result.ProductId == "SKU-001"
        assert result.OldStockCount == 150
        assert result.NewStockCount == 200

    def test_sets_in_stock_false_when_zero(self):
        ctx = FakeContext()
        InventoryService.UpdateStock(ctx, "SKU-001", 0)
        assert _PRODUCTS["SKU-001"].InStock is False

    def test_raises_on_negative_count(self):
        ctx = FakeContext()
        with pytest.raises(InvalidStockCountFault):
            InventoryService.UpdateStock(ctx, "SKU-001", -10)

    def test_raises_when_product_not_found(self):
        ctx = FakeContext()
        with pytest.raises(ProductNotFoundFault):
            InventoryService.UpdateStock(ctx, "SKU-MISSING", 100)

Testing the Generated WSDL

Check that the WSDL generated by spyne contains the expected operations:

# tests/unit/test_wsdl.py
import pytest
from lxml import etree
from wsgiref.simple_server import make_server
from threading import Thread
import urllib.request

from soap_inventory.app import wsgi_app


@pytest.fixture(scope="module")
def soap_server():
    """Start a local SOAP server on a random port for WSDL tests."""
    server = make_server("127.0.0.1", 0, wsgi_app)
    port = server.server_address[1]
    thread = Thread(target=server.serve_forever, daemon=True)
    thread.start()
    yield f"http://127.0.0.1:{port}"
    server.shutdown()


def test_wsdl_contains_get_product_operation(soap_server):
    wsdl_url = f"{soap_server}/?wsdl"
    with urllib.request.urlopen(wsdl_url) as response:
        wsdl_content = response.read()

    root = etree.fromstring(wsdl_content)
    ns = {
        "wsdl": "http://schemas.xmlsoap.org/wsdl/",
        "soap": "http://schemas.xmlsoap.org/wsdl/soap/",
    }

    operations = root.xpath("//wsdl:operation/@name", namespaces=ns)
    assert "GetProduct" in operations
    assert "ListProducts" in operations
    assert "UpdateStock" in operations
    assert "CreateProduct" in operations


def test_wsdl_uses_document_literal_style(soap_server):
    wsdl_url = f"{soap_server}/?wsdl"
    with urllib.request.urlopen(wsdl_url) as response:
        wsdl_content = response.read()

    root = etree.fromstring(wsdl_content)
    ns = {"soap": "http://schemas.xmlsoap.org/wsdl/soap/"}

    bindings = root.xpath("//soap:binding", namespaces=ns)
    assert len(bindings) > 0
    assert bindings[0].get("style") == "document"

Integration Testing with a Live zeep Client

Integration tests run against the real SOAP server and verify the full request/response cycle including XML serialisation.

# tests/integration/test_soap_client.py
import pytest
from decimal import Decimal
from threading import Thread
from wsgiref.simple_server import make_server

from zeep import Client
from zeep.exceptions import Fault

from soap_inventory.app import wsgi_app


@pytest.fixture(scope="module")
def client():
    """Start the SOAP server and return a configured zeep client."""
    server = make_server("127.0.0.1", 0, wsgi_app)
    port = server.server_address[1]
    thread = Thread(target=server.serve_forever, daemon=True)
    thread.start()

    wsdl_url = f"http://127.0.0.1:{port}/?wsdl"
    soap_client = Client(wsdl_url)

    yield soap_client

    server.shutdown()


class TestGetProductIntegration:
    def test_returns_existing_product(self, client):
        result = client.service.GetProduct(ProductId="SKU-001")
        assert result.ProductId == "SKU-001"
        assert result.Name == "Widget Pro"

    def test_raises_fault_for_unknown_product(self, client):
        with pytest.raises(Fault) as exc_info:
            client.service.GetProduct(ProductId="SKU-UNKNOWN")

        fault = exc_info.value
        assert "not found" in fault.message.lower()

    def test_get_product_response_has_all_fields(self, client):
        result = client.service.GetProduct(ProductId="SKU-001")
        assert result.ProductId is not None
        assert result.Name is not None
        assert result.Price is not None
        assert result.InStock is not None
        assert result.StockCount is not None


class TestListProductsIntegration:
    def test_pagination(self, client):
        result = client.service.ListProducts(page=1, page_size=1)
        assert result.TotalCount >= 2
        assert result.PageSize == 1
        assert len(result.Products) == 1

    def test_second_page(self, client):
        page1 = client.service.ListProducts(page=1, page_size=1)
        page2 = client.service.ListProducts(page=2, page_size=1)

        if page1.Products and page2.Products:
            assert page1.Products[0].ProductId != page2.Products[0].ProductId

Mocking SOAP Responses in Application Tests

When testing application code that calls a SOAP service, you do not want to hit the real service. Mock the zeep client with pytest-mock or unittest.mock.

Mocking at the Service Method Level

# tests/unit/test_inventory_api.py
from decimal import Decimal
from unittest.mock import MagicMock, patch
import pytest

# Assume there is an application layer that wraps the SOAP client
from myapp.inventory_service import InventoryService as AppService


@pytest.fixture
def mock_soap_client():
    """Mock the zeep client used by the application service."""
    with patch("myapp.inventory_service.Client") as mock_client_class:
        mock_client = MagicMock()
        mock_client_class.return_value = mock_client
        yield mock_client


def test_get_product_returns_formatted_data(mock_soap_client):
    # Set up a fake zeep response object
    mock_product = MagicMock()
    mock_product.ProductId = "SKU-001"
    mock_product.Name = "Widget Pro"
    mock_product.Price = Decimal("29.99")
    mock_product.InStock = True
    mock_product.StockCount = 150

    mock_soap_client.service.GetProduct.return_value = mock_product

    service = AppService(wsdl_url="http://mock/wsdl")
    result = service.get_product("SKU-001")

    assert result["name"] == "Widget Pro"
    assert result["price"] == 29.99
    mock_soap_client.service.GetProduct.assert_called_once_with(ProductId="SKU-001")


def test_get_product_raises_on_not_found(mock_soap_client):
    from zeep.exceptions import Fault

    mock_soap_client.service.GetProduct.side_effect = Fault(
        message="Product not found",
        code="Client.ProductNotFound",
    )

    service = AppService(wsdl_url="http://mock/wsdl")

    with pytest.raises(KeyError, match="SKU-MISSING"):
        service.get_product("SKU-MISSING")

Fixture XML Files for Response Mocking

For complex response structures, store expected XML in fixture files rather than constructing them in code:

<!-- tests/fixtures/soap_response_get_product.xml -->
<?xml version="1.0" encoding="utf-8"?>
<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
  <soap:Body>
    <GetProductResponse xmlns="http://example.com/inventory">
      <Product>
        <ProductId>SKU-001</ProductId>
        <Name>Widget Pro</Name>
        <Price>29.99</Price>
        <InStock>true</InStock>
        <StockCount>150</StockCount>
      </Product>
    </GetProductResponse>
  </soap:Body>
</soap:Envelope>

# tests/unit/test_raw_response_parsing.py
from pathlib import Path
from lxml import etree

FIXTURES = Path(__file__).parent.parent / "fixtures"


def test_parse_get_product_response():
    xml_content = (FIXTURES / "soap_response_get_product.xml").read_bytes()
    root = etree.fromstring(xml_content)

    ns = {
        "soap": "http://schemas.xmlsoap.org/soap/envelope/",
        "inv": "http://example.com/inventory",
    }

    product_id = root.xpath("//inv:ProductId", namespaces=ns)
    assert product_id[0].text == "SKU-001"

Performance Considerations

1. WSDL Loading Cost

WSDL parsing is the biggest startup cost in zeep. On a slow network, loading a large WSDL with multiple imported schemas can take 2–5 seconds.

Mitigation: Singleton client per process

# soap_inventory/client_pool.py
from functools import lru_cache
from zeep import Client
from zeep.cache import SqliteCache
from zeep.transports import Transport


@lru_cache(maxsize=1)
def get_soap_client(wsdl_url: str) -> Client:
    """Return a cached singleton SOAP client. Thread-safe after first call."""
    cache = SqliteCache(timeout=3600)
    transport = Transport(cache=cache)
    return Client(wsdl=wsdl_url, transport=transport)

Use get_soap_client(wsdl_url) throughout the application — the WSDL and client are loaded once and reused.

2. XML Parsing Overhead

SOAP messages are XML and XML parsing is CPU-intensive compared to JSON. For high-throughput applications, a few guidelines:

• Avoid re-parsing the WSDL on every request (use the singleton pattern above)

• Use connection pooling — the requests.Session with a requests.adapters.HTTPAdapter and pool_connections / pool_maxsize parameters

• Compress responses — if the service supports Accept-Encoding: gzip, enabling it can reduce payload sizes by 60–80%

# connection_pooling.py
import requests
from requests.adapters import HTTPAdapter
from zeep import Client
from zeep.transports import Transport

session = requests.Session()
adapter = HTTPAdapter(
    pool_connections=5,
    pool_maxsize=20,
    max_retries=3,
)
session.mount("https://", adapter)
session.mount("http://", adapter)

transport = Transport(session=session, timeout=(5, 30))
client = Client(wsdl="https://api.example.com/service?wsdl", transport=transport)

3. Async SOAP Clients with httpx

For high-concurrency workloads (e.g., a FastAPI endpoint calling a SOAP service), use httpx as the transport to enable async:

# async_soap_client.py
import asyncio
from zeep import AsyncClient
from zeep.transports import AsyncTransport

async def get_products():
    async with AsyncTransport() as transport:
        client = AsyncClient(
            wsdl="http://localhost:8000/?wsdl",
            transport=transport,
        )
        # Non-blocking — does not hold a thread
        result = await client.service.ListProducts(page=1, page_size=10)
        return result

# Usage in FastAPI
from fastapi import FastAPI
from zeep import AsyncClient
from zeep.transports import AsyncTransport

app = FastAPI()

@app.get("/products")
async def list_products():
    async with AsyncTransport() as transport:
        client = AsyncClient(wsdl="http://localhost:8000/?wsdl", transport=transport)
        result = await client.service.ListProducts(page=1, page_size=10)
        return {"total": result.TotalCount}

SOAP-to-REST Proxy Pattern

A common integration pattern is exposing a SOAP backend through a REST API. This lets modern frontends and microservices consume SOAP services without knowing about XML.

# soap_rest_proxy.py
from decimal import Decimal
from flask import Flask, jsonify, request, abort
from zeep import Client
from zeep.exceptions import Fault
from zeep.helpers import serialize_object
import json

app = Flask(__name__)
soap_client = Client("http://localhost:8000/?wsdl")

FAULT_TO_STATUS = {
    "Client.ProductNotFound": 404,
    "Client.InvalidStockCount": 400,
}


def _soap_to_json(obj) -> dict:
    """Convert zeep response object to JSON-serialisable dict."""
    raw = serialize_object(obj)

    def convert_types(data):
        if isinstance(data, dict):
            return {k: convert_types(v) for k, v in data.items()}
        elif isinstance(data, list):
            return [convert_types(i) for i in data]
        elif isinstance(data, Decimal):
            return float(data)
        return data

    return convert_types(raw)


def _handle_fault(fault: Fault):
    code = fault.code or ""
    status = FAULT_TO_STATUS.get(code, 500)
    abort(status, description=fault.message)


@app.get("/api/v1/products/<product_id>")
def get_product(product_id: str):
    try:
        product = soap_client.service.GetProduct(ProductId=product_id)
        return jsonify(_soap_to_json(product))
    except Fault as fault:
        _handle_fault(fault)


@app.get("/api/v1/products")
def list_products():
    page = request.args.get("page", default=1, type=int)
    page_size = request.args.get("pageSize", default=10, type=int)

    try:
        result = soap_client.service.ListProducts(page=page, page_size=page_size)
        return jsonify({
            "items": [_soap_to_json(p) for p in result.Products],
            "total": result.TotalCount,
            "page": result.Page,
            "pageSize": result.PageSize,
        })
    except Fault as fault:
        _handle_fault(fault)


@app.put("/api/v1/products/<product_id>/stock")
def update_stock(product_id: str):
    body = request.get_json()

    if not body or "quantity" not in body:
        abort(400, description="Request body must include 'quantity'")

    quantity = body["quantity"]
    if not isinstance(quantity, int) or quantity < 0:
        abort(400, description="'quantity' must be a non-negative integer")

    try:
        result = soap_client.service.UpdateStock(
            product_id=product_id,
            new_stock_count=quantity,
        )
        return jsonify(_soap_to_json(result))
    except Fault as fault:
        _handle_fault(fault)


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Observability: Logging and Tracing SOAP Traffic

For production systems, structured logging of SOAP calls is essential for debugging integration issues:

# observability.py
import logging
import time
import uuid
from contextlib import contextmanager
from zeep import Client
from zeep.exceptions import Fault

logger = logging.getLogger("soap.client")


@contextmanager
def soap_span(operation: str, **context):
    """Log SOAP call duration, success/failure with structured context."""
    trace_id = str(uuid.uuid4())[:8]
    start = time.perf_counter()

    logger.info(
        "soap_call_start | trace=%s | operation=%s | ctx=%s",
        trace_id, operation, context,
    )

    try:
        yield
        elapsed_ms = (time.perf_counter() - start) * 1000
        logger.info(
            "soap_call_success | trace=%s | operation=%s | elapsed_ms=%.1f",
            trace_id, operation, elapsed_ms,
        )
    except Fault as fault:
        elapsed_ms = (time.perf_counter() - start) * 1000
        logger.warning(
            "soap_call_fault | trace=%s | operation=%s | elapsed_ms=%.1f | code=%s | message=%s",
            trace_id, operation, elapsed_ms, fault.code, fault.message,
        )
        raise
    except Exception as exc:
        elapsed_ms = (time.perf_counter() - start) * 1000
        logger.error(
            "soap_call_error | trace=%s | operation=%s | elapsed_ms=%.1f | error=%s",
            trace_id, operation, elapsed_ms, exc,
        )
        raise


# Usage
client = Client("http://localhost:8000/?wsdl")

with soap_span("GetProduct", product_id="SKU-001"):
    product = client.service.GetProduct(ProductId="SKU-001")

Series Recap

This series covered the full lifecycle of working with SOAP web services in Python:

Part	Topic
Part 1	SOAP message structure, Envelope/Header/Body/Fault, SOAP 1.1 vs 1.2, raw Python requests
Part 2	WSDL sections, XSD types, reading WSDLs with zeep, writing minimal WSDL
Part 3	spyne service definition, ComplexModel, @rpc, Flask hosting, event handlers
Part 4	zeep client setup, complex types, sessions, WSDL caching, debugging
Part 5	WS-Security UsernameToken, PasswordDigest, timestamps, X.509 signatures
Part 6	SOAP Fault structure, typed faults in spyne, fault handling in zeep
Part 7	Unit/integration testing, mocking, performance, SOAP-to-REST proxy

SOAP is not the future of APIs, but it is a present reality for a large portion of enterprise software. Understanding it deeply makes integration work faster, debugging easier, and the frustration manageable.