test_calls.py 2.8 KB
Newer Older
1 2
from __future__ import print_function

Nathan Graule's avatar
Nathan Graule committed
3 4
import json
import os
Nathan Graule's avatar
Nathan Graule committed
5
import random
Nathan Graule's avatar
Nathan Graule committed
6 7
import time
import unittest
8

Nathan Graule's avatar
Nathan Graule committed
9 10 11 12 13
from call import Call


# noinspection PyMethodMayBeStatic
class TestCall(unittest.TestCase):
14
    def callback_succeeds(self, resolve, _):
Nathan Graule's avatar
Nathan Graule committed
15 16
        resolve('Data')

17
    def callback_rejects(self, _, reject):
Nathan Graule's avatar
Nathan Graule committed
18 19
        reject('Fail')

20
    def callback_resolves_delay(self, resolve, _):
Nathan Graule's avatar
Nathan Graule committed
21 22 23
        time.sleep(2)
        resolve('Data')

24
    def callback_rejects_delay(self, _, reject):
Nathan Graule's avatar
Nathan Graule committed
25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68
        time.sleep(2)
        reject('Fail')

    def test_callback(self):
        call = Call(self.callback_resolves_delay)
        result = call.wait()
        self.assertEqual(result, 'Data')

    def test_rejects(self):
        with self.assertRaises(Exception):
            result = Call(self.callback_rejects_delay).wait()
            self.fail('Call should fail')

    def test_sets_status(self):
        call = Call(self.callback_succeeds)
        call.join()
        self.assertEqual(call.status, Call.RESOLVED)

        call = Call(self.callback_rejects)
        call.join()
        self.assertEqual(call.status, Call.REJECTED)

    def test_chains(self):
        call = Call(self.callback_resolves_delay).then(lambda val: val + ' appended')
        result = call.wait()

        self.assertEqual(result, 'Data appended')

    def test_chains_fail(self):
        call = Call(self.callback_rejects_delay).catch(lambda err: err)
        res = call.wait()
        self.assertEqual(res, 'Fail')

    def test_immediate_resolve(self):
        call = Call.resolve('Data')
        value = call.wait()
        self.assertEqual(value, 'Data')

    def test_immediate_reject(self):
        call = Call.reject(Exception('Something went wrong!'))
        with self.assertRaises(Exception):
            call.wait()
            self.fail('Call should fail')

69
    def test_realworld_example(self):
Nathan Graule's avatar
Nathan Graule committed
70 71 72 73
        def cb(res, rej):
            with open(os.path.join(os.path.dirname(__file__), 'data.json')) as f:
                res(f.read())

74 75 76
        call = Call(cb) \
            .then(lambda data: json.loads(data)) \
            .then(lambda data: data['app-id']) \
Nathan Graule's avatar
Nathan Graule committed
77 78 79 80 81 82 83
            .catch(lambda err: err)
        call.then(print)
        try:
            value = call.wait()
        except Exception as e:
            self.fail('Call.wait shouldn\'t throw as it is being caught')

84
        self.assertIsNone(value)
85
        self.assertTrue(isinstance(value, Exception))
Nathan Graule's avatar
Nathan Graule committed
86 87 88 89 90

    def test_all(self):
        calls = []
        for i in range(10):
            def func():
91
                _i = i
Nathan Graule's avatar
Nathan Graule committed
92
                time.sleep(2 + random.random())
93
                return _i
Nathan Graule's avatar
Nathan Graule committed
94 95

            calls.append(Call.from_function(func))
96
            time.sleep(.2)  # Wait before going to next iteration
Nathan Graule's avatar
Nathan Graule committed
97 98 99 100

        results = Call.all(calls).wait()
        for i in range(10):
            self.assertEqual(results[i], i)