I have a function which get item from a dynamodb table, add 1, put it back in the table and then return the value
import json
import boto3
dynamodb = boto3.resource('dynamodb', 'us-east-1')
table = dynamodb.Table('cloud-resume-challenge')
def visitors_count(event, context):
response = table.get_item(
Key={
'ID': 'visitors'
}
)
visit_count = response['Item']['visitors']
visit_count = str(int(visit_count) + 1)
response = table.put_item(
Item={
'ID': 'visitors',
'visitors': visit_count
}
)
return {
'statusCode': 200,
'headers': {
'Access-Control-Allow-Headers': '*',
'Access-Control-Allow-Origin': '*',
'Access-Control-Allow-Methods': '*'
},
'body': json.dumps({'visit_count': visit_count})
I am testing this function using moto to mock a dynamodb table then calling my function to use it against the mocked table.
import json
import boto3
import pytest
from moto import mock_dynamodb
@pytest.fixture
def apigw_event():
""" Generates API GW Event"""
return {
"body": '{ "test": "body"}',
"resource": "/{proxy+}",
"requestContext": {
"resourceId": "123456",
"apiId": "1234567890",
"resourcePath": "/{proxy+}",
"httpMethod": "POST",
"requestId": "c6af9ac6-7b61-11e6-9a41-93e8deadbeef",
"accountId": "123456789012",
"identity": {
"apiKey": "",
"userArn": "",
"cognitoAuthenticationType": "",
"caller": "",
"userAgent": "Custom User Agent String",
"user": "",
"cognitoIdentityPoolId": "",
"cognitoIdentityId": "",
"cognitoAuthenticationProvider": "",
"sourceIp": "127.0.0.1",
"accountId": "",
},
"stage": "prod",
},
"queryStringParameters": {"foo": "bar"},
"headers": {
"Via": "1.1 08f323deadbeefa7af34d5feb414ce27.cloudfront.net (CloudFront)",
"Accept-Language": "en-US,en;q=0.8",
"CloudFront-Is-Desktop-Viewer": "true",
"CloudFront-Is-SmartTV-Viewer": "false",
"CloudFront-Is-Mobile-Viewer": "false",
"X-Forwarded-For": "127.0.0.1, 127.0.0.2",
"CloudFront-Viewer-Country": "US",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
"Upgrade-Insecure-Requests": "1",
"X-Forwarded-Port": "443",
"Host": "1234567890.execute-api.us-east-1.amazonaws.com",
"X-Forwarded-Proto": "https",
"X-Amz-Cf-Id": "aaaaaaaaaae3VYQb9jd-nvCd-de396Uhbp027Y2JvkCPNLmGJHqlaA==",
"CloudFront-Is-Tablet-Viewer": "false",
"Cache-Control": "max-age=0",
"User-Agent": "Custom User Agent String",
"CloudFront-Forwarded-Proto": "https",
"Accept-Encoding": "gzip, deflate, sdch",
},
"pathParameters": {"proxy": "/examplepath"},
"httpMethod": "POST",
"stageVariables": {"baz": "qux"},
"path": "/examplepath",
}
@mock_dynamodb
def test_visitors_count(apigw_event, mocker):
from visitors_count import app
table_name = 'test'
# Create mock DynamoDB table
dynamodb = boto3.resource('dynamodb', 'us-east-1')
table = dynamodb.create_table(
TableName=table_name,
KeySchema=[{'AttributeName': 'visitors', 'KeyType': 'HASH'}],
AttributeDefinitions=[{'AttributeName': 'visitors', 'AttributeType': 'S'}],
ProvisionedThroughput={'ReadCapacityUnits': 1, 'WriteCapacityUnits': 1}
)
table = dynamodb.Table(table_name)
ret = app.visitors_count(apigw_event, "")
data = json.loads(ret["body"])
assert ret["statusCode"] == 200
assert "visit_count" in ret["body"]
assert data["visit_count"] == '1'
I get the following error :
apigw_event = {'body': '{ "test": "body"}', 'headers': {'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,...ept-Language': 'en-US,en;q=0.8', 'Cache-Control': 'max-age=0', ...}, 'httpMethod': 'POST', 'path': '/examplepath', ...}
mocker = <pytest_mock.plugin.MockerFixture object at 0x00000237DC498310>
@mock_dynamodb
def test_visitors_count(apigw_event, mocker):
from visitors_count import app
table_name = 'test'
# Create mock DynamoDB table
dynamodb = boto3.resource('dynamodb', 'us-east-1')
table = dynamodb.create_table(
TableName=table_name,
KeySchema=[{'AttributeName': 'visitors', 'KeyType': 'HASH'}],
AttributeDefinitions=[{'AttributeName': 'visitors', 'AttributeType': 'S'}],
ProvisionedThroughput={'ReadCapacityUnits': 1, 'WriteCapacityUnits': 1}
)
table = dynamodb.Table(table_name)
> ret = app.visitors_count(apigw_event, "")
tests\unit\test_count.py:78:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
visitors_count\app.py:9: in visitors_count
response = table.get_item(
..\AppData\Local\Programs\Python\Python310\lib\site-packages\boto3\resources\factory.py:580: in do_action
response = action(self, *args, **kwargs)
..\AppData\Local\Programs\Python\Python310\lib\site-packages\boto3\resources\action.py:88: in __call__
response = getattr(parent.meta.client, operation_name)(*args, **params)
..\AppData\Local\Programs\Python\Python310\lib\site-packages\botocore\client.py:508: in _api_call
return self._make_api_call(operation_name, kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <botocore.client.DynamoDB object at 0x00000237DC9C6860>, operation_name = 'GetItem'
api_params = {'Key': {'ID': 'visitors'}, 'TableName': 'cloud-resume-challenge'}
def _make_api_call(self, operation_name, api_params):
operation_model = self._service_model.operation_model(operation_name)
service_name = self._service_model.service_name
history_recorder.record(
'API_CALL',
{
'service': service_name,
'operation': operation_name,
'params': api_params,
},
)
if operation_model.deprecated:
logger.debug(
'Warning: %s.%s() is deprecated', service_name, operation_name
)
request_context = {
'client_region': self.meta.region_name,
'client_config': self.meta.config,
'has_streaming_input': operation_model.has_streaming_input,
'auth_type': operation_model.auth_type,
}
request_dict = self._convert_to_request_dict(
api_params, operation_model, context=request_context
)
resolve_checksum_context(request_dict, operation_model, api_params)
service_id = self._service_model.service_id.hyphenize()
handler, event_response = self.meta.events.emit_until_response(
'before-call.{service_id}.{operation_name}'.format(
service_id=service_id, operation_name=operation_name
),
model=operation_model,
params=request_dict,
request_signer=self._request_signer,
context=request_context,
)
if event_response is not None:
http, parsed_response = event_response
else:
apply_request_checksum(request_dict)
http, parsed_response = self._make_request(
operation_model, request_dict, request_context
)
self.meta.events.emit(
'after-call.{service_id}.{operation_name}'.format(
service_id=service_id, operation_name=operation_name
),
http_response=http,
parsed=parsed_response,
model=operation_model,
context=request_context,
)
if http.status_code >= 300:
error_code = parsed_response.get("Error", {}).get("Code")
error_class = self.exceptions.from_code(error_code)
> raise error_class(parsed_response, operation_name)
E botocore.errorfactory.ResourceNotFoundException: An error occurred (ResourceNotFoundException) when calling the GetItem operation: Requested resource not found
..\AppData\Local\Programs\Python\Python310\lib\site-packages\botocore\client.py:915: ResourceNotFoundException
As you can see in the error description, my function tries to find my real dynamodb table.
When my function is called, I want it to use my mocked table, how can I fix that ?
Your test creates a table called test - and then tries to find a specific item in a table called cloud-resume-challenge.
The visitors_count-method assumes that two preconditions are met:
cloud-resume-challenge existsKey={ 'ID': 'visitors' } existsIn your test, only a table called test is created - so Moto will rightfully throw an error that the correct table cannot be found, exactly like AWS would do.
The test should create a table called cloud-resume-challenge first, and then create the required item. When those preconditions are met, you should be able to call the visitors_count-method.