Viewing file: test_dispatching.py (15.26 KB) -rw-r--r-- Select action/file-type: (+) | (+) | (+) | Code (+) | Session (+) | (+) | SDB (+) | (+) | (+) | (+) | (+) | (+) |
from twisted.trial import unittest from twisted.internet import defer from coherence.dispatcher import Dispatcher, UnknownSignal, Receiver, \ SignalingProperty, ChangedSignalingProperty, CustomSignalingProperty
class TestDispatcher(Dispatcher): __signals__ = {'test': 'Test signal'}
class SimpleTarget(object):
def __init__(self): self.called = 0 self.called_a = 0 self.called_b = 0 self.called_c = 0 self.called_d = 0
def callback(self): self.called += 1
def updater(self, arg1, arg2, value, arg4, key_a='p', variable=None): setattr(self, variable, value) setattr(self, "%s_%s" % (variable, arg2), key_a)
def plus(self, plus, variable=False): setattr(self, variable, getattr(self, variable) + plus)
def fail_before(self, plus, variable=False): raise TypeError(':(') self.update(plus, variable=variable)
class TestDispatching(unittest.TestCase):
def setUp(self): self.called_counter = 0 self.dispatcher = TestDispatcher() self.target = SimpleTarget()
def test_simple_emit(self):
receiver = self.dispatcher.connect('test', self.target.callback) self.dispatcher.emit('test') self.assertEquals(self.target.called, 1)
self.dispatcher.emit('test') self.assertEquals(self.target.called, 2)
self.dispatcher.disconnect(receiver) self.dispatcher.emit('test') self.assertEquals(self.target.called, 2)
def test_simple_deferred_emit(self):
receiver = self.dispatcher.connect('test', self.target.callback) self.dispatcher.deferred_emit('test') self.assertEquals(self.target.called, 1)
self.dispatcher.deferred_emit('test') self.assertEquals(self.target.called, 2)
self.dispatcher.disconnect(receiver) self.dispatcher.deferred_emit('test') self.assertEquals(self.target.called, 2)
def test_simple_save_emit(self):
def call(res): return self.dispatcher.save_emit('test')
def test(res, val): self.assertEquals(self.target.called, val)
receiver = self.dispatcher.connect('test', self.target.callback)
dfr = defer.succeed(None) dfr.addCallback(call) dfr.addCallback(test, 1) dfr.addCallback(call) dfr.addCallback(test, 2) dfr.addCallback(lambda x: self.dispatcher.disconnect(receiver)) dfr.addCallback(call) dfr.addCallback(test, 2) return dfr
def test_connect_typo(self): self.assertRaises(UnknownSignal, self.dispatcher.connect, 'Test', None)
def test_disconnect_none_receiver(self): """ trying to disconnect with None shouldn't fail, it is a valid use case """ self.dispatcher.disconnect(None)
def test_disconnect_false_receiver(self): """ this receiver isn't coming from this dispatcher """ # this is REALLY constructed. you may *not* instantiate a Receiver yourself anyway rec = Receiver('test', None, None, None) self.dispatcher.disconnect(rec)
def test_disconnect_wrong_signal_receiver(self): rec = Receiver('Test', None, None, None) self.assertRaises(UnknownSignal, self.dispatcher.disconnect, rec)
def test_disconnect_not_receiver(self): self.assertRaises(TypeError, self.dispatcher.disconnect, 'test')
def test_emit_false_signal(self): self.assertRaises(UnknownSignal, self.dispatcher.emit, False)
def test_emit_without_receivers(self): self.dispatcher.emit('test') self.assertEquals(self.target.called, 0)
def test_emit_with_multiple_receiver(self): rc1 = self.dispatcher.connect('test', self.target.updater, 1, 2, variable='va1') rc2 = self.dispatcher.connect('test', self.target.updater, 'value', 2, variable='variable') rc3 = self.dispatcher.connect('test', self.target.updater, 'other', 2, variable='one')
self.dispatcher.emit('test', self, 'other', key_a='q') # check rc1 self.assertEquals(self.target.va1, 1) self.assertEquals(self.target.va1_other, 'q') #check rc2 self.assertEquals(self.target.variable, 'value') self.assertEquals(self.target.variable_other, 'q') # check rc3 self.assertEquals(self.target.one, 'other') self.assertEquals(self.target.one_other, 'q')
# now removing the one in the middel self.dispatcher.disconnect(rc2)
# and try again with other data self.dispatcher.emit('test', self, 'other', key_a='thistime') # check rc1 self.assertEquals(self.target.va1, 1) self.assertEquals(self.target.va1_other, 'thistime') #check rc2 self.assertEquals(self.target.variable, 'value') self.assertEquals(self.target.variable_other, 'q') # check rc3 self.assertEquals(self.target.one, 'other') self.assertEquals(self.target.one_other, 'thistime')
# no keyword self.dispatcher.emit('test', self, 'a') # worked for rc1 and rc3 with the default value self.assertEquals(self.target.va1_a, 'p') self.assertEquals(self.target.one_a, 'p') # but not on rc2 self.assertFalse(hasattr(self.target, 'variable_a'))
self.dispatcher.disconnect(rc1) self.dispatcher.disconnect(rc3)
def test_emit_multiple_with_failing_in_between(self):
rc1 = self.dispatcher.connect('test', self.target.plus, 1, variable='called_a') rc2 = self.dispatcher.connect('test', self.target.plus, 2, variable='called_b') rc3 = self.dispatcher.connect('test', self.target.fail_before, 3, variable='called_c') rc4 = self.dispatcher.connect('test', self.target.plus, 4, variable='called_d')
self.dispatcher.emit('test') self.assertEquals(self.target.called_a, 1) self.assertEquals(self.target.called_b, 2) self.assertEquals(self.target.called_c, 0) self.assertEquals(self.target.called_d, 4)
self.dispatcher.emit('test') self.assertEquals(self.target.called_a, 2) self.assertEquals(self.target.called_b, 4) self.assertEquals(self.target.called_c, 0) self.assertEquals(self.target.called_d, 8)
self.dispatcher.disconnect(rc1) self.dispatcher.disconnect(rc2) self.dispatcher.disconnect(rc3) self.dispatcher.disconnect(rc4)
# Receiver tests
class TestReceiver(unittest.TestCase):
def setUp(self): self.called = 0
def _callback(self, *args, **kw): self.called += 1 self.args = args self.kw = kw
def test_simple_calling(self): rec = Receiver('test', self._callback, (), {}) self.assertEquals(rec.signal, 'test') rec() self.assertEquals(self.called, 1) self.assertEquals(self.args, ()) self.assertEquals(self.kw, {})
rec() self.assertEquals(self.called, 2) self.assertEquals(self.args, ()) self.assertEquals(self.kw, {})
rec() self.assertEquals(self.called, 3) self.assertEquals(self.args, ()) self.assertEquals(self.kw, {})
def test_calling_with_args(self): rec = Receiver('test', self._callback, (1, 2, 3), {'test': 'a'}) self.assertEquals(rec.signal, 'test') rec(0) self.assertEquals(self.called, 1) self.assertEquals(self.args, (0, 1, 2, 3)) self.assertEquals(self.kw, {'test': 'a'})
rec(-1) self.assertEquals(self.called, 2) self.assertEquals(self.args, (-1, 1, 2, 3)) self.assertEquals(self.kw, {'test': 'a'})
rec(-2) self.assertEquals(self.called, 3) self.assertEquals(self.args, (-2, 1, 2, 3)) self.assertEquals(self.kw, {'test': 'a'})
def test_calling_with_kw(self): rec = Receiver('test', self._callback, (1, 2, 3), {'test': 'a'}) self.assertEquals(rec.signal, 'test') rec(p='q') self.assertEquals(self.called, 1) self.assertEquals(self.args, (1, 2, 3)) self.assertEquals(self.kw, {'test': 'a', 'p': 'q'})
rec(other='wise') self.assertEquals(self.called, 2) self.assertEquals(self.args, (1, 2, 3)) self.assertEquals(self.kw, {'test': 'a', 'other': 'wise'})
rec(and_one='more') self.assertEquals(self.called, 3) self.assertEquals(self.args, (1, 2, 3)) self.assertEquals(self.kw, {'test': 'a', 'and_one': 'more'})
def test_calling_with_clashing_kw(self): rec = Receiver('test', self._callback, (1, 2, 3), {'test': 'a', 'p': 'a'}) self.assertEquals(rec.signal, 'test') rec(p='q') self.assertEquals(self.called, 1) self.assertEquals(self.args, (1, 2, 3)) self.assertEquals(self.kw, {'test': 'a', 'p': 'q'})
rec(other='wise') self.assertEquals(self.called, 2) self.assertEquals(self.args, (1, 2, 3)) self.assertEquals(self.kw, {'test': 'a', 'other': 'wise', 'p': 'a'})
def test_calling_with_clashing_kw_and_args(self): rec = Receiver('test', self._callback, (1, 2, 3), {'test': 'a', 'p': 'a'}) self.assertEquals(rec.signal, 'test') # without rec() self.assertEquals(self.called, 1) self.assertEquals(self.args, (1, 2, 3)) self.assertEquals(self.kw, {'test': 'a', 'p': 'a'})
rec(1, 2, 7, test='True', o='p') self.assertEquals(self.called, 2) self.assertEquals(self.args, (1, 2, 7, 1, 2, 3)) self.assertEquals(self.kw, {'test': 'True', 'o': 'p', 'p': 'a'})
def test_repr(self): rec = Receiver('test', 'callback', (0, 1, 2), {}) self.assertIn('%s' % id(rec), '%r' % rec) self.assertIn('test', '%r' % rec) self.assertIn('callback', '%r' % rec) self.assertIn('0, 1, 2', '%r' % rec)
# Signal Descriptor test
class SimpleSignaler(object): simple = SignalingProperty('simple')
def __init__(self): self.emitted = []
def emit(self, signal, *values, **kw): self.emitted.append((signal, values, kw))
class DummySignaler(SimpleSignaler):
simple_with_default = SignalingProperty('simple2', default=0)
double_a = SignalingProperty('same-signal') double_b = SignalingProperty('same-signal')
double_c = SignalingProperty('dif-var', var_name='_a') double_d = SignalingProperty('dif-var', var_name='_b')
changer = ChangedSignalingProperty('state') changer_with_default = ChangedSignalingProperty('state2', default='off')
def __init__(self): self.emitted = [] self._x = 0 self.x_get = 0 self.x_set = 0
def xget(self): self.x_get += 1 return self._x
def xset(self, value): self.x_set += 1 self._x = value
def xsq(self, value): self.x_set += 1 self._x = value * value
x = CustomSignalingProperty('x-changed', xget, xset) x_square = CustomSignalingProperty('x-square', xget, xsq)
class TestSignalingDescriptors(unittest.TestCase):
def setUp(self): self.signaler = DummySignaler()
def test_simple(self): self.signaler.simple = 'A' self._check(values=[('simple', ('A',), {})])
# empty self.signaler.emitted = [] self.signaler.simple = 'A' # stays empty self._check()
def test_simple_with_default(self): self.signaler.simple_with_default = 'B' self._check(values=[('simple2', ('B',), {})])
# empty self.signaler.emitted = [] self.signaler.simple_with_default = 'B' # stays empty self._check()
def test_changer(self): self.signaler.changer = 'Yes' self._check(values=[('state', ('Yes', None), {})])
# empty self.signaler.emitted = [] self.signaler.changer = 'Yes' # stays empty self._check()
def test_changer_with_default(self): self.signaler.changer_with_default = 'another' self._check(values=[('state2', ('another', 'off'), {})])
# empty self.signaler.emitted = [] self.signaler.changer_with_default = 'another' # stays empty self._check()
def test_double_same_var(self): self.signaler.double_a = 'A1' self.signaler.double_b = 'B2' self._check(values=[('same-signal', ('A1',), {}), ('same-signal', ('B2',), {})])
# empty self.signaler.emitted = [] # sending B2 over double a even thought it was changed by b self.signaler.double_a = 'B2' self.signaler.double_b = 'B2' # stays empty self._check()
# but changing them different works self.signaler.double_a = 'B1' self.signaler.double_b = 'A2' self._check(values=[('same-signal', ('B1',), {}), ('same-signal', ('A2',), {})])
def test_double_differnt_var(self): self.signaler.double_c = 'A1' self.signaler.double_d = 'B2' self._check(values=[('dif-var', ('A1',), {}), ('dif-var', ('B2',), {})])
# empty self.signaler.emitted = [] self.signaler.double_c = 'A1' self.signaler.double_d = 'B2' # stays empty self._check()
# but they still allow changes self.signaler.double_c = 'B1' self.signaler.double_d = 'A2' self._check(values=[('dif-var', ('B1',), {}), ('dif-var', ('A2',), {})])
def test_custom(self): self.signaler.x = 'Pocahontas' self._check(values=[('x-changed', ('Pocahontas',), {})], x='Pocahontas', x_get=2, x_set=1) self.assertEquals(self.signaler.x, 'Pocahontas')
# settings again to the same value is boring me self.signaler.emitted = [] self.signaler.x_get = 0 self.signaler.x_set = 0
self.signaler.x = 'Pocahontas' self.assertEquals(self.signaler.emitted, []) self.assertEquals(self.signaler.x, 'Pocahontas')
def test_custom_square(self): self.signaler.x_square = 10 self._check(values=[('x-square', (100,), {})], x=100, x_get=2, x_set=1) self.assertEquals(self.signaler.x, 100)
def test_custom_square_nearly_the_same(self): self.signaler._x = 10 self.signaler.x_square = 10 self._check(values=[('x-square', (100,), {})], x=100, x_get=2, x_set=1) self.assertEquals(self.signaler.x, 100)
def _check(self, values=[], x=0, x_set=0, x_get=0): self.assertEquals(self.signaler._x, x) self.assertEquals(self.signaler.x_set, x_set) self.assertEquals(self.signaler.x_get, x_get) self.assertEquals(self.signaler.emitted, values)
class TestStayInObjectSignaling(unittest.TestCase):
def setUp(self): self.foo = SimpleSignaler() self.bar = SimpleSignaler()
def test_double_different_values(self): self.foo.simple = 'A' self.bar.simple = 'B' self.assertEquals(self.foo.simple, 'A') self.assertEquals(self.bar.simple, 'B') self.assertEquals(len(self.foo.emitted), 1) self.assertEquals(len(self.bar.emitted), 1)
self.assertEquals(self.foo.emitted[0][1][0], 'A') self.assertEquals(self.bar.emitted[0][1][0], 'B')
|