Generate and record boto3 errors for use with Cloud Custodian

README.md

Generate and record boto3 errors

The testing for developers section of the Cloud Custodian documentation walks through recording and playing back API responses for use in testing. Some cases can be tricky to reproduce in isolation though, because they occur due to concurrency, timing or caching issues that are difficult to replicate.

If you know that a particular error is possible but it's complex to reproduce inside a test, one alternative is forcing the error outside the Custodian test suite. You can record the error response in a way that simulates Custodian's own placebo-based flight recording, and then use it as an input to your tests.

This (extremely niche and low budget) example generates a couple error cases referenced in cloud-custodian/cloud-custodian#7227. Given an existing Directory Services directory with an ID of d-12345abcde, activate a Cloud Custodian development environment and run:

python /path/to/workspaces_deregister_errorgenerator.py d-12345abcde

This should generate recorded JSON files in the current directory, similar to the ones in this gist.

workspaces.DeregisterWorkspaceDirectory_1.json

{
    "status_code": 400,
    "data": {
        "Error": {
            "Message": "The specified directory is not in a valid state. Confirm that the directory has a status of Active, and try again.",
            "Code": "InvalidResourceStateException"
        },
        "ResponseMetadata": {},
        "message": "The specified directory is not in a valid state. Confirm that the directory has a status of Active, and try again."
    }
}

workspaces.DeregisterWorkspaceDirectory_2.json

{
    "status_code": 400,
    "data": {
        "Error": {
            "Message": "The specified directory could not be found. Confirm that the directory exists in your AWS account and that you own the directory.",
            "Code": "ResourceNotFoundException"
        },
        "ResponseMetadata": {},
        "message": "The specified directory could not be found. Confirm that the directory exists in your AWS account and that you own the directory."
    }
}

workspaces_deregister_errorgenerator.py

import boto3
import sys
import placebo


class RedPill(placebo.pill.Pill):
    """Scrub responses before saving to disk

    Stripped down version of c7n's `RedPill`:

    https://github.com/cloud-custodian/cloud-custodian/blob/9d1e0d6a7035a185eed361b6012c75db4bb3dc4f/tools/c7n_awscc/tests/zpill.py#L247-L270
    """

    def save_response(self, service, operation, response_data, http_response=200):
        """
        `ResponseMetadata` is not useful for replaying data into tests.
        """
        if "ResponseMetadata" in response_data:
            response_data["ResponseMetadata"] = {}
        super().save_response(service, operation, response_data, http_response)


sess = boto3.Session()
pill = RedPill()
pill.attach(sess, data_path="")
pill.record("workspaces", "DeregisterWorkspaceDirectory")
client = sess.client("workspaces")

if len(sys.argv) != 2:
    print(
        """
        Provide a directory ID (d-abcde12345) as an argument. Make sure it contains no WorkSpaces,
        and you're comfortable registering and deregistering it with the WorkSpaces service.
    """
    )
    sys.exit(1)

REAL_DIRECTORY = sys.argv[1]
FAKE_DIRECTORY = "d-a1b2c3d4e5"

errors = []
try:
    # Try to register/deregister/deregister a workspaces directory
    # in rapid succession.
    #
    # If the directory isn't currently registered, it will start
    # registering and then immediately fail on the first deregister
    # attempt.
    #
    # If the directory _is_ currently registered, it will skip the
    # first register call, start deregistering, and fail on the second
    # deregister attempt.
    client.register_workspace_directory(
        DirectoryId=REAL_DIRECTORY, EnableWorkDocs=False
    )
    client.deregister_workspace_directory(DirectoryId=REAL_DIRECTORY)
    client.deregister_workspace_directory(DirectoryId=REAL_DIRECTORY)
except client.exceptions.InvalidResourceStateException as err:
    errors.append(err)

try:
    # Deregister a directory that doesn't exist
    client.deregister_workspace_directory(DirectoryId=FAKE_DIRECTORY)
except client.exceptions.ResourceNotFoundException as err:
    errors.append(err)

print(f"Yay, we broke it! Error count: {len(errors)}", *errors, sep="\n")