pydux-test/calculate.py

import pydux_thunk
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


    class TestSuite(unittest.TestCase):
        def test_action_creators(self):
            # self.assertEqual(ActionCreators.calculate(1, 2), thunk(1, 2))
            self.assertEqual(ActionCreators.start(), {'type': 'calculate/START'})
            self.assertEqual(ActionCreators.success(4), {'type': 'calculate/SUCCESS', 'result': 4})

        # self.assertEqual(ActionCreators.failure(ZeroDivisionError()), {'type': 'calculate/FAILURE', 'error': ZeroDivisionError()})

        def test_reducer(self):
            self.assertEqual(reducer(None, ActionCreators.calculate(2)),
							 {'running': False, 'result': None, 'error': None})

        # self.assertEqual(reducer(452, ActionCreators.set_state(13)), 13)
        # self.assertEqual(reducer(None, ActionCreators.increment()), 1)
        # self.assertEqual(reducer(341, ActionCreators.increment()), 342)
        # self.assertEqual(reducer(None, ActionCreators.decrement()), -1)
        # self.assertEqual(reducer(251, ActionCreators.decrement()), 250)


    unittest.main()