Azure Functions Testing: Local Testing and CI/CD Integration

Azure Functions Testing: Local Testing and CI/CD Integration

Testing Azure Functions has three layers: unit tests for business logic (no Azure SDK needed), local integration tests with Azure Functions Core Tools (func) and Azurite for storage emulation, and CI/CD pipeline tests. The key is to extract your business logic from the function trigger, test it in pure Python or C#, and only use the Azure SDK for integration tests against Azurite or real Azure.

Key Takeaways

Azure Functions Core Tools (func) runs functions locally. Install the func CLI, run func start, and your functions respond on localhost with the same trigger bindings as in Azure.

Azurite emulates Azure Storage locally. Azurite provides local Blob Storage, Queue Storage, and Table Storage. Use it for integration tests without Azure costs.

Extract business logic from the function trigger. The function entry point should only parse the trigger input and call your business logic. Test business logic independently.

Test binding behavior with unit tests using mocked context. For HTTP triggers, pass a mock HttpRequest. For queue triggers, pass a mock QueueMessage. Don't deploy to Azure just to test binding parsing.

Use the Azure SDK's dev/emulator connection strings. For Azurite: "UseDevelopmentStorage=true" or the explicit emulator connection string.

Azure Functions Testing Architecture

An Azure Function has three components, each tested differently:

  1. Function trigger binding — how the function receives input (HTTP request, queue message, blob event)
  2. Business logic — what the function does with the input
  3. Output binding — how the function writes results (HTTP response, queue message, blob)

Test business logic with pure unit tests. Test the trigger/binding layer with mocks. Test the full function with Core Tools running locally.

Python Azure Functions

Project Structure

order-processor/
  function_app.py           # App entry point (v2 programming model)
  order_processor/
    __init__.py
    service.py              # Business logic
    repository.py           # Storage operations
  tests/
    conftest.py
    test_service.py
    test_http_trigger.py
    test_queue_trigger.py
  host.json
  local.settings.json
  requirements.txt
  requirements-test.txt

Local Settings for Testing

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "ORDERS_TABLE_CONNECTION": "UseDevelopmentStorage=true",
    "ORDER_PROCESSING_QUEUE": "order-processing-queue"
  }
}

Function Code (v2 Programming Model)

# function_app.py
import azure.functions as func
import json
import logging
from order_processor.service import process_order

app = func.FunctionApp()

@app.function_name("ProcessOrder")
@app.route(route="orders", methods=["POST"])
def process_order_http(req: func.HttpRequest) -> func.HttpResponse:
    try:
        body = req.get_json()
        result = process_order(body)
        return func.HttpResponse(
            json.dumps(result),
            mimetype="application/json",
            status_code=201
        )
    except ValueError as e:
        return func.HttpResponse(str(e), status_code=400)
    except Exception as e:
        logging.error(f"Error processing order: {e}")
        return func.HttpResponse("Internal server error", status_code=500)

@app.function_name("ProcessOrderQueue")
@app.queue_trigger(arg_name="msg", queue_name="order-processing-queue",
                   connection="ORDERS_TABLE_CONNECTION")
def process_order_queue(msg: func.QueueMessage) -> None:
    order_data = msg.get_json()
    process_order(order_data)
    logging.info(f"Processed order {order_data.get('orderId')} from queue")

Unit Tests for Business Logic

# tests/test_service.py
import pytest
from unittest.mock import patch, MagicMock
from order_processor.service import process_order

@patch("order_processor.service.save_order")
class TestProcessOrder:
    def test_creates_order_with_calculated_total(self, mock_save):
        mock_save.return_value = None
        
        result = process_order({
            "customerId": "cust-123",
            "items": [
                {"productId": "p1", "quantity": 2, "price": 50.00},
                {"productId": "p2", "quantity": 1, "price": 30.00},
            ]
        })
        
        assert result["customerId"] == "cust-123"
        assert result["total"] == 130.00
        assert result["status"] == "pending"
        assert "orderId" in result
    
    def test_raises_for_missing_customer_id(self, mock_save):
        with pytest.raises(ValueError, match="customerId is required"):
            process_order({"items": [{"productId": "p1", "quantity": 1, "price": 10}]})
    
    def test_raises_for_empty_items(self, mock_save):
        with pytest.raises(ValueError, match="items cannot be empty"):
            process_order({"customerId": "cust-123", "items": []})
    
    def test_persists_order(self, mock_save):
        process_order({
            "customerId": "cust-123",
            "items": [{"productId": "p1", "quantity": 1, "price": 100}]
        })
        mock_save.assert_called_once()

HTTP Trigger Unit Tests

# tests/test_http_trigger.py
import json
import pytest
from unittest.mock import patch, MagicMock
import azure.functions as func
from function_app import process_order_http

def make_http_request(body: dict, method: str = "POST") -> func.HttpRequest:
    return func.HttpRequest(
        method=method,
        url="http://localhost:7071/api/orders",
        headers={"Content-Type": "application/json"},
        params={},
        route_params={},
        body=json.dumps(body).encode()
    )

@patch("function_app.process_order")
class TestProcessOrderHttp:
    def test_post_returns_201(self, mock_process):
        mock_process.return_value = {"orderId": "ord-123", "total": 100.0}
        
        req = make_http_request({"customerId": "cust-1", "items": []})
        response = process_order_http(req)
        
        assert response.status_code == 201
        body = json.loads(response.get_body())
        assert body["orderId"] == "ord-123"
    
    def test_returns_400_for_invalid_input(self, mock_process):
        mock_process.side_effect = ValueError("customerId is required")
        
        req = make_http_request({"items": []})
        response = process_order_http(req)
        
        assert response.status_code == 400
    
    def test_returns_500_for_unexpected_error(self, mock_process):
        mock_process.side_effect = Exception("Database unavailable")
        
        req = make_http_request({"customerId": "c1", "items": [{"p": "x", "q": 1, "price": 10}]})
        response = process_order_http(req)
        
        assert response.status_code == 500

Queue Trigger Unit Tests

# tests/test_queue_trigger.py
import json
from unittest.mock import patch, MagicMock, PropertyMock
import azure.functions as func
from function_app import process_order_queue

def make_queue_message(body: dict) -> MagicMock:
    msg = MagicMock(spec=func.QueueMessage)
    msg.get_json.return_value = body
    return msg

@patch("function_app.process_order")
class TestProcessOrderQueue:
    def test_processes_order_from_queue(self, mock_process):
        mock_process.return_value = {"orderId": "ord-123", "status": "completed"}
        
        msg = make_queue_message({
            "orderId": "ord-123",
            "customerId": "cust-1",
            "items": [{"productId": "p1", "quantity": 1, "price": 100}]
        })
        
        process_order_queue(msg)
        
        mock_process.assert_called_once()
        call_arg = mock_process.call_args[0][0]
        assert call_arg["orderId"] == "ord-123"

Integration Tests with Azurite

Azurite is Microsoft's local Azure Storage emulator. Run it with Docker:

# docker-compose.test.yml
services:
  azurite:
    image: mcr.microsoft.com/azure-storage/azurite
    command: "azurite --blobHost 0.0.0.0 --queueHost 0.0.0.0 --tableHost 0.0.0.0"
    ports:
      - "10000:10000"  # Blob
      - "10001:10001"  # Queue
      - "10002:10002"  # Table
# tests/conftest.py
import pytest
from azure.storage.queue import QueueServiceClient
from azure.storage.blob import BlobServiceClient

AZURITE_CONNECTION = (
    "DefaultEndpointsProtocol=http;"
    "AccountName=devstoreaccount1;"
    "AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;"
    "BlobEndpoint=http://127.0.0.1:10000/devstoreaccount1;"
    "QueueEndpoint=http://127.0.0.1:10001/devstoreaccount1;"
)

@pytest.fixture(scope="session")
def blob_client():
    return BlobServiceClient.from_connection_string(AZURITE_CONNECTION)

@pytest.fixture(scope="session")
def queue_client():
    return QueueServiceClient.from_connection_string(AZURITE_CONNECTION)

@pytest.fixture
def test_container(blob_client):
    container_name = "test-orders"
    blob_client.create_container(container_name)
    yield blob_client.get_container_client(container_name)
    blob_client.delete_container(container_name)

@pytest.fixture
def test_queue(queue_client):
    queue_name = "test-order-queue"
    queue_client.create_queue(queue_name)
    yield queue_client.get_queue_client(queue_name)
    queue_client.delete_queue(queue_name)

Blob Storage Integration Tests

# tests/integration/test_blob_operations.py
import json
from order_processor.blob_storage import upload_order_json, download_order_json

class TestBlobOperations:
    def test_upload_and_download_order(self, test_container):
        order = {"orderId": "ord-123", "total": 150.0}
        
        upload_order_json(test_container, "orders/ord-123.json", order)
        retrieved = download_order_json(test_container, "orders/ord-123.json")
        
        assert retrieved["orderId"] == "ord-123"
        assert retrieved["total"] == 150.0
    
    def test_blob_exists_after_upload(self, test_container):
        upload_order_json(test_container, "orders/ord-456.json", {"orderId": "ord-456"})
        
        blob_client = test_container.get_blob_client("orders/ord-456.json")
        assert blob_client.exists()

Local Testing with Azure Functions Core Tools

Install Core Tools:

# macOS
brew tap azure/functions
brew install azure-functions-core-tools@4

<span class="hljs-comment"># Ubuntu
curl https://packages.microsoft.com/keys/microsoft.asc <span class="hljs-pipe">| gpg --dearmor > microsoft.gpg
<span class="hljs-built_in">sudo <span class="hljs-built_in">mv microsoft.gpg /etc/apt/trusted.gpg.d/microsoft.gpg
<span class="hljs-built_in">sudo sh -c <span class="hljs-string">'echo "deb [arch=amd64] https://packages.microsoft.com/repos/microsoft-ubuntu-$(lsb_release -cs)-prod $(lsb_release -cs) main" > /etc/apt/sources.list.d/dotnetdev.list'
<span class="hljs-built_in">sudo apt-get install azure-functions-core-tools-4

Run functions locally:

func start

Test with curl:

curl -X POST http://localhost:7071/api/orders \
  -H "Content-Type: application/json" \
  -d <span class="hljs-string">'{"customerId": "cust-123", "items": [{"productId": "p1", "quantity": 2, "price": 50}]}'

CI/CD Pipeline Configuration

# .github/workflows/azure-functions-tests.yml
name: Azure Functions Tests
on: [push, pull_request]

jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install -r requirements-test.txt
      - run: pytest tests/unit/ -v

  integration-tests:
    runs-on: ubuntu-latest
    services:
      azurite:
        image: mcr.microsoft.com/azure-storage/azurite
        ports:
          - 10000:10000
          - 10001:10001
          - 10002:10002
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install -r requirements-test.txt
      - run: pytest tests/integration/ -v
        env:
          AZURE_STORAGE_CONNECTION: "UseDevelopmentStorage=true"

Summary

Azure Functions testing follows the same pattern as other serverless testing:

  1. Unit tests — extract business logic, test it without Azure bindings
  2. Binding unit tests — mock HttpRequest, QueueMessage, and verify your handler parses them correctly
  3. Integration tests — Azurite for local storage testing, Cosmos DB emulator for database testing
  4. Local end-to-endfunc start with Core Tools for full function execution

The most common mistake is testing business logic through the function binding—it makes tests slow and requires Azure infrastructure. Extract the logic, test it directly, and keep binding tests focused on parsing and response formatting.

Read more