If you spend much time in the Cloud Engineering space, the moto testing module has probably come up.
It’s a reliable option for writing robust tests with AWS mocks. The only problem with the library, if you can call it that, is that it establishes unfulfilled expectations for what it is like to write tests for Azure and GCP, which have no comparable project. This gap can be so impactful that I have seen several projects with robust coverage for AWS operations and zero for the other services – with some teams I’ve encountered insisting that these services are untestable.
Challenge accepted. To implement the basics of a mock service, a test suite would need to:
- Convince the SDK client it is authenticated
- Create the basic primitives of the service for the mock (e.g., an Azure subscription)
- Produce an expected return according to the spec
One bright side to approaching these SDKs is the availability of firm typing in the requests and responses, unlike AWS, where yet another community-driven project has had to make typing available for boto3 interactions since AWS chooses not to. As a result, with some reflection and code reading, the mocking implementations come together once you get the hang of how the library’s internals work.
Writing an example Azure Mock
Authentication
Let’s start easy. An Azure SDK client needs a token. The SDK never validates it, so it doesn’t need to be genuine. Fortunately, mocking Azure authentication is as simple as making an invalid credential object.
from azure.identity import ClientSecretCredential
MOCK_CLIENT_ID = "mock-client-id"
MOCK_CLIENT_SECRET = "mock-client-secret"
MOCK_TENANT_ID = "11111111-1111-1111-1111-111111111111"
def mock_azure_token() -> ClientSecretCredential:
token = ClientSecretCredential(
client_id=MOCK_CLIENT_ID,
client_secret=MOCK_CLIENT_SECRET,
tenant_id=MOCK_TENANT_ID,
)
Creating a subscription
Now that there is some ability to create tokens, it is possible to start playing with generating resources. Azure clients have a _deserialize method to return a specified type defined in the models. A basic implementation that vends subscriptions for a test suite is below.
from azure.mgmt.subscription import SubscriptionClient
from azure.mgmt.subscription.models import Subscription
def mock_azure_subscription_object(
subscription_id="00000000-0000-0000-0000-000000000000", state="Enabled"
) -> Subscription:
token = mock_azure_token()
sub_client = SubscriptionClient(token)
mock_subscription: Subscription = sub_client.subscription._deserialize(
"Subscription",
{
"id": f"/subscriptions/{subscription_id}",
"subscriptionId": subscription_id,
"displayName": "Mock Subscription",
"state": state, # Possible values are Enabled, Warned, PastDue, Disabled, and Deleted.
},
)
return mock_subscription
Mocking tenant-level behavior
With subscription vending out of the way, you gain the flexibility to implement higher-order tenant management functions. For instance, a test could now have a deterministic list of subscriptions to work from in the testing suite. Again, the test relies on the _deserialize method, but for this example, it uses SubscriptionListResult instead of simply a Subscription.
def mock_azure_subscription_list_result(_):
token = mock_azure_token()
sub_client = SubscriptionClient(token)
mock_subscription: SubscriptionListResult = sub_client.subscription._deserialize(
"SubscriptionListResult",
{
"value": [
mock_azure_subscription_object(),
mock_azure_subscription_object(
subscription_id="00000000-1111-0000-1111-000000000000"
),
mock_azure_subscription_object(
subscription_id="00000000-2222-0000-2222-000000000000",
state="Disabled",
),
mock_azure_subscription_object(
subscription_id="00000000-3333-0000-3333-000000000000",
state="Deleted",
),
mock_azure_subscription_object(
subscription_id="00000000-4444-0000-4444-000000000000"
),
]
},
)
return mock_subscription.value
Writing an example GCP Mock
Authentication
Faking GCP authentication required a bit more investigation into the SDK internals, but ultimately, it still supports a reasonably clean implementation. Essentially, the auth flow has to be both patched and also assigned a return_value that includes the credentials. Fortunately, using unittest‘s sentinel provides sufficient fakes to satisfy the SDK.
from unittest.mock import patch, sentinel
import pytest
@pytest.fixture
def patch_gcp_auth():
with patch("google.auth.default") as google_auth_default:
google_auth_default.return_value = (
sentinel.credentials,
sentinel.project,
)
yield
Creating a project
Unlike Azure, creating a method to vend a GCP project object requires no external knowledge or dependencies.
from datetime import datetime
from google.cloud.resourcemanager_v3.types import Project
from google.protobuf.timestamp_pb2 import Timestamp
def mock_gcp_project_object():
# https://stackoverflow.com/a/66960304/1886901
t = datetime.now().timestamp()
seconds = int(t)
nanos = int(t % 1 * 1e9)
project = Project(
name="projects/123456789012",
project_id="mock-project-123",
state="ACTIVE", # Possible values are STATE_UNSPECIFIED, ACTIVE, DELETE_REQUESTED.
create_time=Timestamp(seconds=seconds, nanos=nanos),
)
return project
Mocking organization-level behavior
Due to the nature of the GCP SDK structure, more of the project needs to be patched – as opposed to Azure’s, which is more independent class-driven. Some naive return values can be implemented to list projects and folders.
from unittest.mock import patch, sentinel
import pytest
from google.cloud.resourcemanager_v3 import ListFoldersResponse, ListProjectsResponse
from google.cloud.resourcemanager_v3.services.folders.pagers import ListFoldersPager
from google.cloud.resourcemanager_v3.services.projects.pagers import ListProjectsPager
from tests.test_functions.test_account_monitor.gcp_utils import mock_gcp_project_object
@pytest.fixture
def patch_gcp_auth():
with patch("google.auth.default") as google_auth_default, patch(
"google.cloud.resourcemanager_v3.ProjectsClient.list_projects"
) as list_projects, patch(
"google.cloud.resourcemanager_v3.FoldersClient.list_folders"
) as list_folders:
google_auth_default.return_value = (
sentinel.credentials,
sentinel.project,
)
list_projects.return_value = ListProjectsPager(
None,
None,
ListProjectsResponse(projects=[mock_gcp_project_object()]),
)
list_folders.return_value = ListFoldersPager(
None, None, ListFoldersResponse(folders=[])
)
yield
Building from here
These examples are meant to serve as getting-started guides. The implementation lacks state, does not return handled errors, and doesn’t carry any nuance about how the services work. I expect a properly motivated team could extrapolate pretty far from here. Unfortunately, my current day-to-day does not position me to invest much more than my team’s essentials into a test suite like this, but I would be happy to partner with others to create a battle-tested version.
Leave a Reply