Library to easily mock/stub Azure Python SDK methods?

[jinkang23]

Are there any library to easily mock and stub Azure Python SDK methods for writing unit tests?

I noticed that azure-devtools project may offer this capability but it has been archived since 2020.
What I'm looking for is similar to Moto library for AWS boto3 SDK but for Azure Python SDK.

Thank you.

[xiangyan99]

Thanks for reaching out.

Usually we use modk.patch to mock a method.

Does this work for you?

[github-actions]

Hi @jinkang23. Thank you for opening this issue and giving us the opportunity to assist. To help our team better understand your issue and the details of your scenario please provide a response to the question asked above or the information requested above. This will help us more accurately address your issue.

[annatisch]

Hi @jinkang23,
Depending on your needs for the mock, I find that the Azure SDK transport is a good way to achieve this.
All Azure SDKs accept a replacement transport at the constructor, which allows you to swap out the actual service request for your own logic, so you can inspect the outgoing payload or inject your own response or errors.

Here's an example I made for some unit tests in the Tables SDK:
https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/tables/azure-data-tables/tests/test_encoder.py#L47-L77

As mentioned, all our SDKs support this, so you should be able to share the implementation between any combination of SDKs you might be using.
The protocol is pretty straightforward:

from azure.core.pipeline.transport import HttpTransport

class MockTransport(HttpTransport):
    def send(request: HttpRequest, **kwargs) -> HttpResponse:
        # This is where you add any test logic, for example, testing a connection failure:
        # raise azure.core.exception.ServiceResponseError("something went wrong!")

# Then we can plug this into any SDK via kwargs:
storage_client = BlobServiceClient(account_url, credential, transport=MockTransport())

[github-actions]

Hi @jinkang23, we're sending this friendly reminder because we haven't heard back from you in 7 days. We need more information about this issue to help address it. Please be sure to give us your input. If we don't hear back from you within 14 days of this comment the issue will be automatically closed. Thank you!

[pamelafox]

@annatisch Could you possibly advise on mocking AsyncHttpTransport? I'm having a harder time with that, starting from your code. Current status:

import os
from collections import namedtuple

import aiohttp
from azure.core.exceptions import ResourceNotFoundError
from azure.core.pipeline.transport import HttpRequest, HttpResponse, AioHttpTransportResponse
from azure.storage.blob.aio import BlobServiceClient
from azure.core.pipeline.transport import AsyncHttpTransport
import pytest

import app


MockToken = namedtuple("MockToken", ["token", "expires_on"])


class MockAzureCredential:
    async def get_token(self, uri):
        return MockToken("mock_token", 9999999999)


@pytest.mark.asyncio
async def test_content_file(monkeypatch, mock_env):

    class MockAiohttpClientResponse404(aiohttp.ClientResponse):
        def __init__(self, url, body_bytes, headers=None):
            self._body = body_bytes
            self._headers = headers
            self._cache = {}
            self.status = 404
            self.reason = "Not Found"
            self._url = url

    class MockAiohttpClientResponse(aiohttp.ClientResponse):
        def __init__(self, url, body_bytes, headers=None):
            self._body = body_bytes
            self._headers = headers
            self._cache = {}
            self.status = 200
            self.reason = "OK"
            self._url = url

    class MockTransport(AsyncHttpTransport):
        async def send(self, request: HttpRequest, **kwargs) -> AioHttpTransportResponse:
            if request.url.endswith("notfound.pdf"):
                raise ResourceNotFoundError(MockAiohttpClientResponse404(request.url, b""))
            else:
                return AioHttpTransportResponse(request,
                    MockAiohttpClientResponse(request.url, b"test content", {"Content-Type": "application/octet-stream", "Content-Range": "bytes 0-27/28", "Content-Length": "28"}))

        async def __aenter__(self):
            return self

        async def __aexit__(self, *args):
            pass

        async def open(self):
            pass

        async def close(self):
            pass

    # Then we can plug this into any SDK via kwargs:
    blob_client = BlobServiceClient(f"https://{os.environ['AZURE_STORAGE_ACCOUNT']}.blob.core.windows.net",
        credential=MockAzureCredential(), transport=MockTransport())
    blob_container_client = blob_client.get_container_client(os.environ["AZURE_STORAGE_CONTAINER"])

    quart_app = app.create_app() 
    async with quart_app.test_app() as test_app:

        quart_app.config.update({"blob_container_client": blob_container_client})

        client = test_app.test_client()
        response = await client.get("/content/notfound.pdf")
        assert response.status_code == 404

        response = await client.get("/content/role_library.pdf")
        assert response.status_code == 200

[pamelafox]

Update: My test is passing now, but is very slow (20 seconds), so I'm wondering if the SDK code is waiting/sleeping somewhere. Typically I expected tests with mocked out network calls to be quite fast.