!C99Shell v. 2.0 [PHP 7 Update] [25.02.2019]!

Software: Apache/2.2.16 (Debian). PHP/5.3.3-7+squeeze19 

uname -a: Linux mail.tri-specialutilitydistrict.com 2.6.32-5-amd64 #1 SMP Tue May 13 16:34:35 UTC
2014 x86_64
 

uid=33(www-data) gid=33(www-data) groups=33(www-data) 

Safe-mode: OFF (not secure)

/usr/share/pyshared/coherence/test/   drwxr-xr-x
Free 129.89 GB of 142.11 GB (91.4%)
Home    Back    Forward    UPDIR    Refresh    Search    Buffer    Encoder    Tools    Proc.    FTP brute    Sec.    SQL    PHP-code    Update    Feedback    Self remove    Logout    


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')

:: Command execute ::

Enter:
 
Select:
 

:: Search ::
  - regexp 

:: Upload ::
 
[ Read-Only ]

:: Make Dir ::
 
[ Read-Only ]
:: Make File ::
 
[ Read-Only ]

:: Go Dir ::
 
:: Go File ::
 

--[ c99shell v. 2.0 [PHP 7 Update] [25.02.2019] maintained by KaizenLouie | C99Shell Github | Generation time: 0.0092 ]--