Demo of Ports and Adapters Architecture + Testing for cloud functions in Python

conftest.py

import os

class InMemoryUserRepository(UserRepository):
    def __init__(self):
        self._items = []

    def add(self, user: User) -> User:
        self._items.append(user)

    def get(self, id: str) -> User:
        for user in self._items:
            if user.name == id:
                return user

              
class InMemoryCognitoClient:
    def __init__(self, )

    def sign_up(self, ClientId: str) -> Dict[str, Any]:
        if not ClientId:
            raise CognitoHardToFindError('Give me a ClientId!')

        return {
            'UserConfirmed': True|False,
            'CodeDeliveryDetails': {
                'Destination': 'string',
                'DeliveryMedium': 'SMS'|'EMAIL',
                'AttributeName': 'string'
            },
            'UserSub': 'string'
        }

    def get_user(self, AccessToken: str) -> Dict[str, Any]:
        return {
            'Username': 'string',
            'UserAttributes': [
                {
                    'Name': 'string',
                    'Value': 'string'
                },
            ],
            'MFAOptions': [
                {
                    'DeliveryMedium': 'SMS'|'EMAIL',
                    'AttributeName': 'string'
                },
            ],
            'PreferredMfaSetting': 'string',
            'UserMFASettingList': [
                'string',
            ]
        }

      
@pytest.fixture
def user_repository():
    if os.environ.get('STAGE') == 'CI':
        return InMemoryUserRepository()
    else:
        return CognitoUserRepository()


@pytest.fixture
def cognito_client():
    if os.environ.get('STAGE') == 'CI':
        return InMemoryCognitoClient()
    else:
        return boto3.client('cognito')

handler.py

from abc import ABCMeta, abstractmethod
from dataclasses import dataclass
from typing import Dict

import boto3
from aws_lambda import lambda_http_response


Payload = Dict[str, str]


def parse_http_event(event) -> Payload:
    return {'user_name': event['bodyparams']['user_name']}


@dataclass
class User:
    name: str


class UserCreateService:

    def __init__(self, repository):
        self._repository = repository

    def create(self, payload: Dict[str, str]) -> User:
        user = User(name=payload['name'])
        return self._repository.add(user=user)


class UserRepository(metaclass=ABCMeta):
    def __init__(self):
        pass

    @abstractmethod
    def add(self, user: User) -> User:
        pass

    @abstractmethod
    def get(self, id: str) -> User:
        pass


class CognitoUserRepository(UserRepository):
    def __init__(self, client=None):
        self._client = client if client else boto3.client('cognito-dentity-provider')

    def add(self, user: User) -> User:
        response = self._client.sign_up(ClientId=user.name)
        return self._from_sign_up_response(response)

    def _from_get_user_response(self, response: Dict[str, Any]) -> User:
        pass

    def get(self, id: str) -> User:
        response = self._client.get_user(AccessToken=id)
        return self._from_get_user_response(response)

    def _from_sign_up_response(self, response: Dict[str, Any]) -> User:
        pass



@lambda_http_response
def handler(event, context):
    repository = CognitoUserRepository()
    service = UserCreateService(repository)
    return service.create(parse_http_event(event))

test_handler.py

import handler


class TestHandler:
    
    def test_handler(self):
        event = dict(PathParams=dict(user_name='oswald'))
        response = handler(event=event)
        assert response

        repository = CognitoUserRepository()
        user = repository.get(id=response['user_id'])
        assert user
    
    def test_handler_missing_repository(self):
        event = dict(PathParams=dict(user_name='oswald'))
        response = handler(event=event)
        assert response
        assert response['statusCode'] == http.BAD_REQUEST
        assert response['body']['message'] == 'Give me the name of your user'

test_repository.py

import pytest


class TestCognitoUserRepository:

    @pytest.mark.use_fixture('cognito_client')
    def test_add(self, cognito_client):
        repository = CognitoUserRepository(client=cognito_client)
        user = repository.add(user=User('oswald'))
        assert user
        assert user.name == 'oswald'

    def test_add_raises():
        pass

    @pytest.mark.use_fixture('cognito_client')
    def test_get(self, cognito_client):

    def test_get_raises():
        pass

def test_parse_http_event():
    event = dict(PathParams=dict(user_name='oswald'))
    payload = parse_http_event(event)

    assert payload

test_service.py

import pytest


class TestUserCreateService:

    @pytest.mark.use_fixture('user_repository')
    def test_create(self, user_repository):
        test_user = User('oswald')
        subject = UserCreateService(user_repository)
        response = service.create(payload=test_user)
        assert response

        expected_user = user_repository.get(id=test_user.name)

        assert expected_user
        assert expected_user.name == test_user.name