如何使用moto@mok_dynamodb2模拟失败的操作

Question

我目前正在尝试使用Moto&mock_dynamodb2。到目前为止，它一直在为我测试我的"；成功操作"；测试用例。但我很难让它为我的"；失败案例"；。

在我的测试代码中，我有：

@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? 

TestClassUnderTestExample看起来像这样：

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

当我真正嘲笑.put_item操作以返回一个不成功的值时，我已经完全陷入了困境，这样我就可以测试ClassUnderTestExample是否会按预期处理它并抛出自定义错误。我尝试过在运行测试之前删除表之类的操作，但这只是在获取表时抛出异常，而不是执行带有错误代码的PutItem。

我也试过为pydash或测试上方的表格添加补丁，但我一定做错了什么。我在莫的文件里找不到任何东西。任何帮助都将不胜感激！

Answer 1

Moto的目标是完全模仿AWS的行为，包括用户提供错误输入时的行为。换句话说，对put_item()的调用如果对AWS失败，那么对Moto也会失败。

没有强制对有效输入进行错误响应的内置方式。

从你的例子中很难判断这是如何强制的，但似乎值得一试这一行来创建一个无效的输入：
ddb_item = <do stuff with some_value to create sanitized item>

Answer 2

可以。对此使用嘲讽。简单且可运行的示例：

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)