Testing

Azure Blob Storage Testing: BlobServiceClient, Azurite Emulator, and Integration Patterns

HelpMeTest

HelpMeTest

— 6 min read
Azure Blob Storage Testing: BlobServiceClient, Azurite Emulator, and Integration Patterns

Azure Blob Storage has two testing paths: Azurite for integration tests against a real local emulator, and azure-storage-blob mocks for fast unit tests. Both have sharp edges — Azurite doesn't support all Blob Storage features, and the official SDK's mock story is weaker than AWS's moto. This guide shows both approaches with Python and TypeScript examples.

Key Takeaways

Azurite is Microsoft's official local emulator. Run it via Docker and point your BlobServiceClient at http://localhost:10000/devstoreaccount1 with the well-known development credentials.

The azure-storage-blob SDK doesn't have a built-in mock. Unit test by injecting a fake BlobServiceClient or using unittest.mock to patch at the method level — there's no moto equivalent for Azure.

SAS token testing must verify both structure and expiry. Generate a SAS token with known parameters and assert the token string contains the expected signed fields.

Azurite limitations matter. Blob inventory policies, geo-redundancy, and some diagnostic settings don't work in Azurite — stick to integration tests for features it does support.

TypeScript tests with Azurite need the connection string format. Use UseDevelopmentStorage=true or the explicit Azurite connection string; the Azure SDK resolves the rest.

Azure Blob Storage is Azure's equivalent of S3, with its own SDK, terminology (containers instead of buckets, blobs instead of objects), and quirks. Testing it effectively means picking the right tool for each layer: mocks for unit tests of your business logic, Azurite for integration tests of your storage code.

Azurite: The Local Azure Emulator

Start Azurite with Docker:

docker run -d \
  --name azurite \
  -p 10000:10000 \
  -p 10001:10001 \
  -p 10002:10002 \
  mcr.microsoft.com/azure-storage/azurite

# Azurite ports:
<span class="hljs-comment"># 10000 = Blob service
<span class="hljs-comment"># 10001 = Queue service  
<span class="hljs-comment"># 10002 = Table service

The development account credentials are hardcoded and well-known:

  • Account name: devstoreaccount1
  • Account key: Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==

Python Integration Tests with Azurite

pip install azure-storage-blob pytest
import pytest
from azure.storage.blob import BlobServiceClient, ContentSettings

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

@pytest.fixture(scope="session")
def blob_service_client():
    return BlobServiceClient.from_connection_string(AZURITE_CONNECTION_STRING)

@pytest.fixture
def test_container(blob_service_client):
    container_name = "test-container"
    container_client = blob_service_client.get_container_client(container_name)
    
    # Create if not exists
    if not container_client.exists():
        container_client.create_container()
    
    yield container_name
    
    # Cleanup: delete all blobs
    for blob in container_client.list_blobs():
        container_client.delete_blob(blob.name)

def test_upload_and_download_blob(blob_service_client, test_container):
    container_client = blob_service_client.get_container_client(test_container)
    
    # Upload
    container_client.upload_blob(
        name="documents/report.txt",
        data=b"Q1 financial report data",
        content_settings=ContentSettings(content_type="text/plain"),
        overwrite=True
    )
    
    # Download
    blob_client = container_client.get_blob_client("documents/report.txt")
    download = blob_client.download_blob()
    content = download.readall()
    
    assert content == b"Q1 financial report data"

def test_list_blobs_with_prefix(blob_service_client, test_container):
    container_client = blob_service_client.get_container_client(test_container)
    
    # Upload a few blobs under different prefixes
    container_client.upload_blob("images/photo1.jpg", b"jpg data 1", overwrite=True)
    container_client.upload_blob("images/photo2.jpg", b"jpg data 2", overwrite=True)
    container_client.upload_blob("videos/clip.mp4", b"mp4 data", overwrite=True)
    
    # List only images/
    images = list(container_client.list_blobs(name_starts_with="images/"))
    
    assert len(images) == 2
    names = [b.name for b in images]
    assert "images/photo1.jpg" in names
    assert "images/photo2.jpg" in names

def test_blob_properties_and_metadata(blob_service_client, test_container):
    container_client = blob_service_client.get_container_client(test_container)
    
    # Upload with metadata
    container_client.upload_blob(
        name="tagged/file.bin",
        data=b"binary content",
        metadata={"uploaded_by": "pipeline", "version": "2.0"},
        content_settings=ContentSettings(content_type="application/octet-stream"),
        overwrite=True
    )
    
    blob_client = container_client.get_blob_client("tagged/file.bin")
    props = blob_client.get_blob_properties()
    
    assert props.metadata["uploaded_by"] == "pipeline"
    assert props.metadata["version"] == "2.0"
    assert props.content_settings.content_type == "application/octet-stream"
    assert props.size == 14

Unit Tests with Mocks (Python)

For unit tests of code that uses BlobServiceClient, mock at the method level with unittest.mock:

from unittest.mock import MagicMock, patch, PropertyMock
import pytest
from azure.storage.blob import BlobServiceClient
from azure.core.exceptions import ResourceNotFoundError

# Your application code to test
class BlobStorageService:
    def __init__(self, connection_string: str, container_name: str):
        self.client = BlobServiceClient.from_connection_string(connection_string)
        self.container_name = container_name
    
    def upload_file(self, blob_name: str, data: bytes, content_type: str = "application/octet-stream") -> str:
        container = self.client.get_container_client(self.container_name)
        container.upload_blob(
            name=blob_name,
            data=data,
            content_settings=ContentSettings(content_type=content_type),
            overwrite=True
        )
        return blob_name
    
    def file_exists(self, blob_name: str) -> bool:
        blob = self.client.get_blob_client(container=self.container_name, blob=blob_name)
        return blob.exists()
    
    def get_file_size(self, blob_name: str) -> int:
        blob = self.client.get_blob_client(container=self.container_name, blob=blob_name)
        props = blob.get_blob_properties()
        return props.size

# Unit tests using mocks
@pytest.fixture
def mock_blob_service():
    with patch("azure.storage.blob.BlobServiceClient.from_connection_string") as mock_factory:
        mock_client = MagicMock()
        mock_factory.return_value = mock_client
        yield mock_client

def test_upload_file_calls_upload_blob(mock_blob_service):
    service = BlobStorageService("fake-conn-string", "my-container")
    
    mock_container = MagicMock()
    mock_blob_service.get_container_client.return_value = mock_container
    
    result = service.upload_file("path/to/file.txt", b"content", "text/plain")
    
    assert result == "path/to/file.txt"
    mock_container.upload_blob.assert_called_once_with(
        name="path/to/file.txt",
        data=b"content",
        content_settings=ContentSettings(content_type="text/plain"),
        overwrite=True
    )

def test_file_exists_returns_true_when_blob_present(mock_blob_service):
    service = BlobStorageService("fake-conn-string", "my-container")
    
    mock_blob = MagicMock()
    mock_blob.exists.return_value = True
    mock_blob_service.get_blob_client.return_value = mock_blob
    
    assert service.file_exists("existing-file.txt") is True

def test_file_exists_returns_false_when_missing(mock_blob_service):
    service = BlobStorageService("fake-conn-string", "my-container")
    
    mock_blob = MagicMock()
    mock_blob.exists.return_value = False
    mock_blob_service.get_blob_client.return_value = mock_blob
    
    assert service.file_exists("missing-file.txt") is False

TypeScript Integration Tests with Azurite

npm install @azure/storage-blob jest ts-jest @types/jest
// blob-storage.ts — your application code
import {
  BlobServiceClient,
  ContainerClient,
  StorageSharedKeyCredential,
  generateBlobSASQueryParameters,
  BlobSASPermissions,
} from "@azure/storage-blob";

export class BlobStorage {
  private serviceClient: BlobServiceClient;
  private containerName: string;

  constructor(connectionString: string, containerName: string) {
    this.serviceClient = BlobServiceClient.fromConnectionString(connectionString);
    this.containerName = containerName;
  }

  async upload(blobName: string, content: Buffer, contentType: string): Promise<void> {
    const containerClient = this.serviceClient.getContainerClient(this.containerName);
    const blobClient = containerClient.getBlockBlobClient(blobName);
    await blobClient.upload(content, content.length, {
      blobHTTPHeaders: { blobContentType: contentType },
    });
  }

  async download(blobName: string): Promise<Buffer> {
    const containerClient = this.serviceClient.getContainerClient(this.containerName);
    const blobClient = containerClient.getBlobClient(blobName);
    const downloadResponse = await blobClient.download();
    const chunks: Buffer[] = [];
    for await (const chunk of downloadResponse.readableStreamBody!) {
      chunks.push(Buffer.from(chunk));
    }
    return Buffer.concat(chunks);
  }
}
// blob-storage.test.ts
import { BlobStorage } from "./blob-storage";
import { BlobServiceClient } from "@azure/storage-blob";

const AZURITE_CONNECTION_STRING =
  "DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;" +
  "AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;" +
  "BlobEndpoint=http://127.0.0.1:10000/devstoreaccount1;";

const CONTAINER = "ts-test-container";

describe("BlobStorage integration tests (Azurite)", () => {
  let serviceClient: BlobServiceClient;

  beforeAll(async () => {
    serviceClient = BlobServiceClient.fromConnectionString(AZURITE_CONNECTION_STRING);
    const containerClient = serviceClient.getContainerClient(CONTAINER);
    await containerClient.createIfNotExists();
  });

  afterEach(async () => {
    const containerClient = serviceClient.getContainerClient(CONTAINER);
    for await (const blob of containerClient.listBlobsFlat()) {
      await containerClient.deleteBlob(blob.name);
    }
  });

  it("uploads and downloads a file correctly", async () => {
    const storage = new BlobStorage(AZURITE_CONNECTION_STRING, CONTAINER);
    const content = Buffer.from("TypeScript test content");

    await storage.upload("test/file.txt", content, "text/plain");
    const downloaded = await storage.download("test/file.txt");

    expect(downloaded).toEqual(content);
  });

  it("lists blobs with a prefix filter", async () => {
    const containerClient = serviceClient.getContainerClient(CONTAINER);
    const storage = new BlobStorage(AZURITE_CONNECTION_STRING, CONTAINER);

    await storage.upload("prefix/a.txt", Buffer.from("a"), "text/plain");
    await storage.upload("prefix/b.txt", Buffer.from("b"), "text/plain");
    await storage.upload("other/c.txt", Buffer.from("c"), "text/plain");

    const blobs: string[] = [];
    for await (const blob of containerClient.listBlobsFlat({ prefix: "prefix/" })) {
      blobs.push(blob.name);
    }

    expect(blobs).toHaveLength(2);
    expect(blobs).toContain("prefix/a.txt");
    expect(blobs).toContain("prefix/b.txt");
  });
});

Testing SAS Tokens

SAS (Shared Access Signature) tokens are Azure's equivalent of presigned URLs. Test that your generation code produces tokens with the right permissions and expiry:

from azure.storage.blob import (
    BlobServiceClient,
    generate_blob_sas,
    BlobSasPermissions,
    generate_container_sas,
    ContainerSasPermissions,
)
from datetime import datetime, timezone, timedelta
from urllib.parse import urlparse, parse_qs
import requests

def test_sas_token_blob_access(blob_service_client, test_container):
    container_client = blob_service_client.get_container_client(test_container)
    container_client.upload_blob("sas-test/document.txt", b"SAS protected", overwrite=True)
    
    # Generate a SAS token
    expiry = datetime.now(timezone.utc) + timedelta(hours=1)
    sas_token = generate_blob_sas(
        account_name="devstoreaccount1",
        container_name=test_container,
        blob_name="sas-test/document.txt",
        account_key="Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==",
        permission=BlobSasPermissions(read=True),
        expiry=expiry,
    )
    
    # Build the full URL
    sas_url = (
        f"http://127.0.0.1:10000/devstoreaccount1/{test_container}/sas-test/document.txt?{sas_token}"
    )
    
    # Verify the SAS token contains expected fields
    params = parse_qs(sas_token)
    assert "se" in params  # signed expiry
    assert "sp" in params  # signed permissions
    assert "sig" in params  # signature
    assert "r" in params["sp"][0]  # read permission
    
    # Verify it works (Azurite validates SAS tokens)
    response = requests.get(sas_url)
    assert response.status_code == 200
    assert response.content == b"SAS protected"

Running Tests in CI

# .github/workflows/blob-storage-tests.yml
name: Azure Blob Storage Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    
    services:
      azurite:
        image: mcr.microsoft.com/azure-storage/azurite
        ports:
          - 10000:10000
          - 10001:10001
          - 10002:10002
        options: >-
          --health-cmd "nc -z localhost 10000"
          --health-interval 5s
          --health-timeout 3s
          --health-retries 10

    steps:
      - uses: actions/checkout@v4
      
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      
      - name: Install dependencies
        run: pip install azure-storage-blob pytest requests
      
      - name: Run tests
        run: pytest tests/ -v

Using GitHub Actions' services block means Azurite starts before any test steps and is available at localhost:10000 for the entire job. No manual Docker setup required.

Read more