"Fossies" - the Fresh Open Source Software Archive

Member "zim-0.71.1/tests/signals.py" (4 Apr 2019, 3674 Bytes) of package /linux/privat/zim-0.71.1.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. For more information about "signals.py" see the Fossies "Dox" file reference documentation and the last Fossies "Diffs" side-by-side code changes report: 0.69.1_vs_0.70.

    1 
    2 # Copyright 2013 Jaap Karssenberg <jaap.karssenberg@gmail.com>
    3 
    4 
    5 
    6 
    7 import tests
    8 
    9 import time
   10 
   11 from zim.signals import *
   12 
   13 
   14 # TODO test Connector
   15 
   16 
   17 class TestEmitter(tests.TestCase):
   18 
   19     def testSignal(self):
   20         emitter = Emitter()
   21         self.assertIsNone(emitter.state)
   22 
   23         emitter.emit('bar', 'test1')
   24         self.assertEqual(emitter.state, 'DO bar test1')
   25 
   26         data = []
   27         def check(o, a):
   28             self.assertIs(o, emitter)
   29             data.append(a)
   30 
   31         i = emitter.connect('bar', check)
   32         emitter.emit('bar', 'test2')
   33         self.assertEqual(emitter.state, 'DO bar test2')
   34         self.assertEqual(data, ['test2'])
   35         emitter.disconnect(i)
   36 
   37         emitter.emit('bar', 'test3')
   38         self.assertEqual(emitter.state, 'DO bar test3')
   39         self.assertEqual(data, ['test2']) # check stopped listening after disconnect
   40 
   41 
   42     def testHook(self):
   43         emitter = Emitter()
   44         self.assertIsNone(emitter.emit_return_first('foo', 'x'))
   45 
   46         emitter.connect('foo', lambda o, a: a * 3)
   47         emitter.connect('foo', lambda o, a: a * 5)
   48         self.assertEqual(emitter.emit_return_first('foo', 'x'), 'xxx')
   49             # pick first result
   50 
   51     def testSignalSetup(self):
   52         emitter = FancyEmitter()
   53         self.assertIsNone(emitter.state)
   54 
   55         emitter.connect('foo', lambda o, a: None)
   56         self.assertEqual(emitter.state, 'SETUP foo')
   57 
   58     def testInheritance(self):
   59         emitter = ChildEmitter()
   60         emitter.connect('bar', lambda o: 'foo') # no error
   61         self.assertRaises(AssertionError, emitter.connect, 'none_existing', lambda o: 'foo')
   62             # assert non existing raises --> thus previous non-error was really OK
   63 
   64     def testRunSequence(self):
   65         emitter = ChildEmitter()
   66 
   67         emitter.connect('last', lambda o, l: l.append('NORMAL'))
   68         emitter.connect_after('last', lambda o, l: l.append('AFTER'))
   69         seq = []
   70         emitter.emit('last', seq)
   71         self.assertEqual(seq, ['NORMAL', 'CLOSURE', 'AFTER'])
   72 
   73         emitter.connect('first', lambda o, l: l.append('NORMAL'))
   74         emitter.connect_after('first', lambda o, l: l.append('AFTER'))
   75         seq = []
   76         emitter.emit('first', seq)
   77         self.assertEqual(seq, ['CLOSURE', 'NORMAL', 'AFTER'])
   78 
   79 
   80 class Emitter(SignalEmitter):
   81 
   82     __signals__ = {
   83         'foo': (None, object, (str,)),
   84         'bar': (None, None, (str,)),
   85     }
   86 
   87     def __init__(self):
   88         self.state = None
   89 
   90     def do_bar(self, arg):
   91         self.state = 'DO bar %s' % arg
   92 
   93 
   94 class FancyEmitter(SignalEmitter):
   95 
   96     __signals__ = {
   97         'foo': (None, None, ()),
   98     }
   99 
  100     def __init__(self):
  101         self.state = None
  102 
  103     def _setup_signal(self, signal):
  104         self.state = 'SETUP %s' % signal
  105 
  106     def _teardown_signal(self, signal):
  107         self.state = 'TEARDOWN %s' % signal
  108 
  109 
  110 class ChildEmitter(Emitter):
  111 
  112     __signals__ = {
  113         'first': (SIGNAL_RUN_FIRST, None, (object,)),
  114         'last': (SIGNAL_RUN_LAST, None, (object,)),
  115     }
  116 
  117     def do_first(self, list):
  118         list.append('CLOSURE')
  119 
  120     def do_last(self, list):
  121         list.append('CLOSURE')
  122 
  123 
  124 class TestSignalHandler(tests.TestCase):
  125 
  126     def runTest(self):
  127         obj = ClassWithHandler()
  128         self.assertEqual(obj.count, 0)
  129         self.assertEqual(id(obj.add_one), id(obj.add_one)) # unique instance object
  130 
  131         obj.add_one()
  132         self.assertEqual(obj.count, 1)
  133 
  134         with obj.add_one.blocked():
  135             obj.add_one()
  136             obj.add_one()
  137             obj.add_one()
  138         self.assertEqual(obj.count, 1)
  139 
  140         obj.add_one()
  141         obj.add_one()
  142         obj.add_one()
  143         self.assertEqual(obj.count, 4)
  144 
  145 
  146 class ClassWithHandler(object):
  147 
  148     def __init__(self):
  149         self.count = 0
  150 
  151     @SignalHandler
  152     def add_one(self):
  153         self.count += 1
  154 
  155 
  156 
  157 
  158 @tests.slowTest
  159 class TestDelayedCallback(tests.TestCase):
  160 
  161     def runTest(self):
  162         counter = tests.Counter()
  163 
  164         callback = DelayedCallback(500, lambda o: counter())
  165         for i in range(3):
  166             callback('foo')
  167 
  168         for i in range(10):
  169             time.sleep(1)
  170             tests.gtk_process_events()
  171             if callback.timer_id is None:
  172                 break
  173 
  174         self.assertEqual(counter.count, 1)