pydux-test/calculate.py

import asyncio

initial_state = {
    'running': False,
    'result': None,
    'error': None
}


class ActionTypes(object):
    START = 'calculate/START'
    FAILURE = 'calculate/FAILURE'
    SUCCESS = 'calculate/SUCCESS'


class ActionCreators(object):
    @staticmethod
    def calculate(value1, value2):
        return thunk(value1, value2)

    @staticmethod
    def start():
        return dict(
            type=ActionTypes.START
        )

    @staticmethod
    def success(result):
        return dict(
            type=ActionTypes.SUCCESS,
            result=result
        )

    @staticmethod
    def failure(error):
        return dict(
            type=ActionTypes.FAILURE,
            error=error
        )


def reducer(state=None, action=None):
    if state is None:
        state = initial_state
    if not (isinstance(action, dict) and 'type' in action):
        return state
    if action['type'] == ActionTypes.START:
        state['running'] = True
    elif action['type'] == ActionTypes.SUCCESS:
        state['running'] = False
        state['result'] = action['result']
        state['error'] = None
    elif action['type'] == ActionTypes.FAILURE:
        state['running'] = False
        state['result'] = None
        state['error'] = action['error']
    return state


async def calculate(value1, value2):
    print('dividing', value1, value2)
    await asyncio.sleep(3)
    return value1 / value2


def thunk(value1, value2):
    async def wrapper(dispatch, get_state):
        dispatch(ActionCreators.start())
        try:
            result = await calculate(value1, value2)
            dispatch(ActionCreators.success(result))
        except Exception as error:
            dispatch(ActionCreators.failure(dict(
                type=error.__class__.__name__,
                description=str(error)
            )))

    return wrapper


if __name__ == '__main__':
    import unittest
    import unittest.mock as mock
    import inspect
    loop = asyncio.get_event_loop()

    class TestSuite(unittest.TestCase):
        def test_action_creators(self):
            received = ActionCreators.calculate(1, 2)
            expected = thunk(1, 2)
            self.assertEqual(expected.__code__, received.__code__)
            self.assertEqual(ActionCreators.start(), {'type': 'calculate/START'})
            self.assertEqual(ActionCreators.success(4), {'type': 'calculate/SUCCESS', 'result': 4})
            self.assertEqual(ActionCreators.failure(ZeroDivisionError().__class__), {'type': 'calculate/FAILURE', 'error': ZeroDivisionError().__class__})

        def test_reducer(self):
            self.assertEqual(reducer(None, None), initial_state)
            self.assertEqual(reducer(initial_state, None), initial_state)
            self.assertEqual(reducer(initial_state, {'type': ActionCreators.calculate(1, 2)}), initial_state)

            self.assertEqual(reducer(initial_state, ActionCreators.start()), {'running': True, 'result': None, 'error': None})
            self.assertEqual(reducer(initial_state, ActionCreators.success(7)), {'running': False, 'result': 7, 'error': None})
            self.assertEqual(reducer(initial_state, ActionCreators.failure(ZeroDivisionError.__class__)), {'running': False, 'result': None, 'error': ZeroDivisionError.__class__})

        def test_thunk(self):
            dispatched_actions = []

            def dispatch(action):
                dispatched_actions.append(action)

            def get_state():
                pass

            wrapper = thunk(3, 7)
            self.assertTrue(inspect.iscoroutinefunction(wrapper))
            task = wrapper(dispatch, get_state)
            loop.run_until_complete(task)
            self.assertEqual(dispatched_actions, [{'type': 'calculate/START'}, {'type': 'calculate/SUCCESS', 'result': 3/7}])


    unittest.main()