Skip to content

Instantly share code, notes, and snippets.

@ASRagab
Last active June 20, 2019 17:19
Show Gist options
  • Save ASRagab/1628ba07a7293fbc955d87a5ecacf911 to your computer and use it in GitHub Desktop.
Save ASRagab/1628ba07a7293fbc955d87a5ecacf911 to your computer and use it in GitHub Desktop.
Moto TNT Presentation

Instructions

pip install boto3 moto 'moto[server]'

Usage

To start moto server for sqs (ONLY FOR NON BOTO TESTING):

moto_server sqs
  • Replace sqs with whatever other service you would want to test (i.e. s3 or ec2)
  • see README.md for more information about the services you can mock moto
import boto3
import os
from datetime import datetime
keyField = 'keyField'
purchase = 'purchase'
updated_at = 'updated_at'
def get_dynamo_resource():
return boto3.resource('dynamodb', region_name=os.environ['REGION'])
def get_table(resource, name):
return resource.Table(name)
def get_item(table, key):
try:
resp = table.get_item(Key=key)
return resp
except Exception as e:
raise Exception("something bad happened")
def put_item(table, item):
try:
result = table.update_item(
Key={keyField: item[keyField]},
UpdateExpression=f"SET {updated_at}=:updatedAt ADD {purchase} :purchase",
ExpressionAttributeValues={
':purchase': {item[purchase]},
':updatedAt': datetime.utcnow().isoformat() + 'Z'
}
)
status_code = result['ResponseMetadata']['HTTPStatusCode']
return status_code
except Exception as e:
raise Exception(f"something bad happened: {str(e)}")
def sample_item():
return {
keyField: 1,
purchase: 'purchase2'
}
module Moto
require 'aws-sdk-sqs'
sqs = Aws::SQS::Resource.new(region: 'us-west-2', endpoint: 'http://localhost:5000') # use us-west-2
sqs.create_queue(queue_name: 'tnt-test-queue')
url = sqs.client.get_queue_url(queue_name: 'tnt-test-queue')[:queue_url]
puts url
resp = sqs.client.send_message(queue_url: url, message_body: '{"key": "test_message"}')
puts resp
end
import pytest
import os
import boto3
from botocore.exceptions import ClientError
from moto import mock_dynamodb2
from service import dynamo as test_module
from service.dynamo import keyField, purchase
table_name = 'purchases-table'
OK = 200
class TestDynamo(object):
def _setup(self):
os.environ['REGION'] = 'us-east-1'
return self._create_table()
@staticmethod
@mock_dynamodb2
def _create_table():
dynamodb = boto3.resource('dynamodb', region_name=os.environ['REGION'])
table = dynamodb.create_table(
TableName=table_name,
KeySchema=[
{
'AttributeName': keyField,
'KeyType': 'HASH'
}
],
AttributeDefinitions=[
{
'AttributeName': keyField,
'AttributeType': 'N'
}
],
ProvisionedThroughput={
'ReadCapacityUnits': 10,
'WriteCapacityUnits': 10
}
)
print("Table status:", table.table_status)
return dynamodb, table
def test_harness(self):
assert 1 == 1
@mock_dynamodb2
def test_get_table(self):
dynamodb, table = self._setup()
table = test_module.get_table(dynamodb, table_name)
assert table.table_status == 'ACTIVE'
assert table.table_name == table_name
@mock_dynamodb2
def test_get_table_should_fail_if_given_wrong_table(self):
dynamodb, table = self._setup()
bad = "Tabley McTableson"
table = test_module.get_table(dynamodb, bad)
with pytest.raises(ClientError):
assert table.table_status == 'ACTIVE'
@mock_dynamodb2
def test_get_item(self):
dynamodb, table = self._setup()
table.put_item(Item=test_module.sample_item())
resp = test_module.get_item(table, {keyField: 1})
assert resp == test_module.sample_item()
@mock_dynamodb2
def test_put_item(self):
dynamodb, table = self._setup()
res = test_module.put_item(table, test_module.sample_item())
assert res == OK
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment