Moto TNT Presentation

README.md

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

dynamo.py

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'
    }

moto.rb

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

test_dynamo.py

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