1############################################################################### 2# 3# The MIT License (MIT) 4# 5# Copyright (c) Crossbar.io Technologies GmbH 6# 7# Permission is hereby granted, free of charge, to any person obtaining a copy 8# of this software and associated documentation files (the "Software"), to deal 9# in the Software without restriction, including without limitation the rights 10# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell 11# copies of the Software, and to permit persons to whom the Software is 12# furnished to do so, subject to the following conditions: 13# 14# The above copyright notice and this permission notice shall be included in 15# all copies or substantial portions of the Software. 16# 17# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR 18# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, 19# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE 20# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER 21# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 22# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN 23# THE SOFTWARE. 24# 25############################################################################### 26 27import pytest 28import txaio 29 30from util import run_once 31 32 33def test_as_future_immediate(framework): 34 ''' 35 Returning an immediate value from as_future 36 ''' 37 errors = [] 38 results = [] 39 calls = [] 40 41 def method(*args, **kw): 42 calls.append((args, kw)) 43 return 42 44 f = txaio.as_future(method, 1, 2, 3, key='word') 45 46 def cb(x): 47 results.append(x) 48 49 def errback(f): 50 errors.append(f) 51 52 txaio.add_callbacks(f, cb, errback) 53 54 run_once() 55 56 assert len(results) == 1 57 assert len(errors) == 0 58 assert results[0] == 42 59 assert calls[0] == ((1, 2, 3), dict(key='word')) 60 61 62def test_as_future_immediate_none(framework): 63 ''' 64 Returning None immediately from as_future 65 ''' 66 errors = [] 67 results = [] 68 calls = [] 69 70 def method(*args, **kw): 71 calls.append((args, kw)) 72 return None 73 f = txaio.as_future(method, 1, 2, 3, key='word') 74 75 def cb(x): 76 results.append(x) 77 78 def errback(f): 79 errors.append(f) 80 81 txaio.add_callbacks(f, cb, errback) 82 83 run_once() 84 85 assert len(results) == 1 86 assert len(errors) == 0 87 assert results[0] is None 88 assert calls[0] == ((1, 2, 3), dict(key='word')) 89 90 91def test_as_future_coroutine(framework): 92 ''' 93 call a coroutine (asyncio) 94 ''' 95 pytest.importorskip('asyncio') 96 # can import asyncio on python3.4, but might still be using 97 # twisted 98 if not txaio.using_asyncio: 99 return 100 101 errors = [] 102 results = [] 103 calls = [] 104 105 from asyncio import coroutine 106 107 @coroutine 108 def method(*args, **kw): 109 calls.append((args, kw)) 110 return 42 111 f = txaio.as_future(method, 1, 2, 3, key='word') 112 113 def cb(x): 114 results.append(x) 115 116 def errback(f): 117 errors.append(f) 118 119 txaio.add_callbacks(f, cb, errback) 120 121 run_once() 122 run_once() 123 124 assert len(results) == 1 125 assert len(errors) == 0 126 assert results[0] == 42 127 assert calls[0] == ((1, 2, 3), dict(key='word')) 128 129 130def test_as_future_exception(framework): 131 ''' 132 Raises an exception from as_future 133 ''' 134 errors = [] 135 results = [] 136 calls = [] 137 exception = RuntimeError("sadness") 138 139 def method(*args, **kw): 140 calls.append((args, kw)) 141 raise exception 142 f = txaio.as_future(method, 1, 2, 3, key='word') 143 144 def cb(x): 145 results.append(x) 146 147 def errback(f): 148 errors.append(f) 149 150 txaio.add_callbacks(f, cb, errback) 151 152 run_once() 153 154 assert len(results) == 0 155 assert len(errors) == 1 156 assert errors[0].value == exception 157 assert calls[0] == ((1, 2, 3), dict(key='word')) 158 159 160def test_as_future_recursive(framework): 161 ''' 162 Returns another Future from as_future 163 ''' 164 errors = [] 165 results = [] 166 calls = [] 167 f1 = txaio.create_future_success(42) 168 169 def method(*args, **kw): 170 calls.append((args, kw)) 171 return f1 172 f0 = txaio.as_future(method, 1, 2, 3, key='word') 173 174 def cb(x): 175 results.append(x) 176 177 def errback(f): 178 errors.append(f) 179 180 txaio.add_callbacks(f0, cb, errback) 181 182 run_once() 183 184 assert len(results) == 1 185 assert len(errors) == 0 186 assert results[0] == 42 187 assert calls[0] == ((1, 2, 3), dict(key='word')) 188