faaspact-mocker.py

import json
from contextlib import contextmanager
from typing import Any, Callable, Dict, Generator, List, NamedTuple, Optional, Set, Tuple
from urllib.parse import urlparse, parse_qs

import requests
import responses


class Request(NamedTuple):
    method: str
    path: str
    json: Dict
    query: Dict
    headers: Dict


class Response(NamedTuple):
    status_code: int
    headers: Dict
    json: Dict


class Interaction(NamedTuple):
    description: str
    provider_states: Tuple[str]
    request: Request
    response: Response


class Pact:

    def __init__(self, consumer_name: str, provider_name: str, provider_url: str) -> None:
        self.consumer_name = consumer_name
        self.provider_name = provider_name
        self.provider_url = provider_url
        self.pact_specification = '3.0.0'
        self.interactions: List[Interaction] = []
        self.calls: List[Tuple[Interaction, Any]] = []

    def add_interaction(self, interaction: Interaction) -> None:
        self.interactions.append(interaction)

    @contextmanager
    def mocker(self, outer: Optional[responses.RequestsMock] = None) -> Generator:
        with responses.RequestsMock() as responses_mock:
            for interaction in self.interactions:
                responses_mock.add_callback(
                    interaction.request.method,
                    self.provider_url + interaction.request.path,
                    callback=self._make_callback(interaction, register_call=lambda call: self.calls.append(call))
                )

            yield

            for interaction, request in self.calls:
                assert set(interaction.request.headers.items()) <= set(request.headers.items())
                assert self._query_param_set(interaction.request.query) <= self._query_param_set(self._pluck_query_params(request.url))
                self._pluck_query_params(request.url)

                if interaction.request.json:
                    assert interaction.request.json == json.loads(request.body)

    @staticmethod
    def _make_callback(interaction: Interaction, register_call: Callable[[Tuple[Interaction, requests.models.PreparedRequest]], None]) -> Callable[[requests.models.PreparedRequest], None]:
        def callback(request: requests.models.PreparedRequest):
            register_call((interaction, request))
            return (
                interaction.response.status_code,
                interaction.response.headers,
                json.dumps(interaction.response.json)
            )

        return callback

    @staticmethod
    def _pluck_query_params(url: str) -> Dict:
        qs = urlparse(url).query
        return parse_qs(qs)

    @staticmethod
    def _query_param_set(params: Dict) -> Set:
        return {(k, tuple(v) if isinstance(v, list) else v) for k, v in params.items()}


pact = Pact('Zach', 'Gabe', 'https://zachgabechat.com')

pact.add_interaction(Interaction(
    description='Zach messages gabe',
    provider_states=('Gabe is online',),
    request=Request(
        method='POST',
        path='/gabe',
        json={'message': 'Hey gabe'},
        query={},
        headers={}
    ),
    response=Response(
        json={'message': 'Ayee whatsup'},
        headers={},
        status_code=200
    )
))

pact.add_interaction(Interaction(
    description='Zach checks friends online',
    provider_states=('I have one friend online',),
    request=Request(
        method='GET',
        path='/friends',
        query={'status': ['online']},
        json={},
        headers={}
    ),
    response=Response(
        json={'number': 1},
        headers={},
        status_code=200
    )
))

with pact.mocker():
    requests.post('https://zachgabechat.com/gabe', json={'message': 'Hey gabe'})
    requests.get('https://zachgabechat.com/friends', params={'status': 'online'})