How to mock failed operations with moto, @mock_dynamodb2?

Question

I am currently trying to write unit tests for my python code using Moto & @mock_dynamodb2 . So far it's been working for me to test my "successful operation" test cases. But I'm having trouble getting it to work for my "failure cases".

In my test code I have:

@mock_dynamodb2
class TestClassUnderTestExample(unittest.TestCase):
    def setUp(self):
        ddb = boto3.resource("dynamodb", "us-east-1")
        self.table = ddb.create_table(<the table definition)
        self.example_under_test = ClassUnderTestExample(ddb)

    def test_some_thing_success(self):
        expected_response = {<some value>}
        assert expected_response = self.example_under_test.write_entry(<some value>)

    def test_some_thing_success(self):
        response = self.example_under_test.write_entry(<some value>)
        # How to assert exception is thrown by forcing put item to fail?

The TestClassUnderTestExample would look something like this:

class ClassUnderTestExample:
   def __init__(self, ddb_resource=None):
        if not ddb_resource:
            ddb_resource = boto3.resource('dynamodb')
        self.table = ddb_resource.Table(.....)

   def write_entry(some_value)
        ddb_item = <do stuff with some_value to create sanitized item>        

        response = self.table.put_item(
            Item=ddb_item
        )

        if pydash.get(response, "ResponseMetadata.HTTPStatusCode") != 200:
            raise SomeCustomErrorType("Unexpected response from DynamoDB when attempting to PutItem")

        return ddb_item

I've been completely stuck when it comes to actually mocking the .put_item operation to return a non-success value so that I can test that the ClassUnderTestExample will handle it as expected and throw the custom error. I've tried things like deleting the table before running the test, but that just throws an exception when getting the table rather than an executed PutItem with an error code.

I've also tried putting a patch for pydash or for the table above the test but I must be doing something wrong. I can't find anything in moto's documentation. Any help would be appreciated!

score 2 · Answer 1 · answered Oct 15 '21 at 20:01

The goal of Moto is to completely mimick AWS' behaviour, including how to behave when the user supplies erroneous inputs. In other words, a call to put_item() that fails against AWS, would/should also fail against Moto.

There is no build-in way to force an error response on a valid input.

It's difficult to tell from your example how this can be forced, but it looks like it's worth playing around with this line to create an invalid input:
ddb_item = <do stuff with some_value to create sanitized item>

funnydman · Answer 2 · 2021-10-29T12:49:39.830

Yes, you can. Use mocking for this. Simple and runnable example:

from unittest import TestCase
from unittest.mock import Mock
from uuid import uuid4

import boto3
from moto import mock_dynamodb2


def create_user_table(table_name: str) -> dict:
    return dict(
        TableName=table_name,
        KeySchema=[
            {
                'AttributeName': 'id',
                'KeyType': 'HASH'
            },
        ],
        AttributeDefinitions=[
            {
                'AttributeName': 'id',
                'AttributeType': 'S'
            },
        ],
        BillingMode='PAY_PER_REQUEST'
    )


class UserRepository:
    table_name = 'users'

    def __init__(self, ddb_resource):
        if not ddb_resource:
            ddb_resource = boto3.resource('dynamodb')
        self.table = ddb_resource.Table(self.table_name)

    def create_user(self, username):
        return self.table.put_item(Item={'id': str(uuid4), 'username': username})


@mock_dynamodb2
class TestUserRepository(TestCase):
    def setUp(self):
        ddb = boto3.resource("dynamodb", "us-east-1")
        self.table = ddb.create_table(**create_user_table('users'))
        self.test_user_repo = UserRepository(ddb)

    def tearDown(self):
        self.table.delete()

    def test_some_thing_success(self):
        user = self.test_user_repo.create_user(username='John')
        assert len(self.table.scan()['Items']) == 1

    def test_some_thing_failure(self):
        self.test_user_repo.table = table = Mock()
        table.put_item.side_effect = Exception('Boto3 Exception')

        with self.assertRaises(Exception) as exc:
            self.test_user_repo.create_user(username='John')
            self.assertTrue('Boto3 Exception' in exc.exception)

How to mock failed operations with moto, @mock_dynamodb2?

2 Answers2