"Fossies" - the Fresh Open Source Software Archive

Member "zim-0.71.1/tests/operations.py" (30 Apr 2019, 4550 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 "operations.py" see the Fossies "Dox" file reference documentation and the last Fossies "Diffs" side-by-side code changes report: 0.70_vs_0.71.0.

    1 
    2 # Copyright 2017 Jaap Karssenberg <jaap.karssenberg@gmail.com>
    3 
    4 
    5 
    6 import tests
    7 
    8 from gi.repository import Gtk
    9 
   10 from zim.notebook.operations import *
   11 
   12 
   13 class MockProgressDialog(object):
   14 
   15     def __init__(self, parent, operation):
   16         # self.heading.set_text(operation.msg)
   17         operation.connect('step', self.on_iteration_step)
   18         operation.connect('finished', self.on_iteration_finished)
   19         self.steps = []
   20         self.finished = False
   21 
   22     def on_iteration_step(self, o, progress):
   23         self.steps.append(progress)
   24         if isinstance(progress, tuple) and len(progress) == 3:
   25             i, total, msg = progress
   26             # self.bar.set_fraction(i/total)
   27             # self.bar.set_text('%i / %i' % (i, total))
   28         else:
   29             msg = progress
   30             # self.bar.pulse()
   31 
   32         # if isinstance(msg, basestring):
   33             # self.label.set_text(msg)
   34 
   35     def on_iteration_finished(self, o):
   36         self.finished = True
   37         # self.response(...) # break run() loop
   38 
   39 
   40 class MockNotebook(object):
   41 
   42     def __init__(self):
   43         self._operation_check = NOOP
   44         self.value = None
   45 
   46     @notebook_state
   47     def test(self, value):
   48         self.value = value
   49 
   50     def raw(self, value):
   51         self.value = value
   52 
   53 
   54 def mock_iterator(notebook):
   55     for i in 0, 1, 2:
   56         notebook.test('Test %i' % i)
   57         yield i
   58 
   59 def mock_gtk_iter(notebook):
   60     for i in 0, 1, 2:
   61         notebook.test('Test %i' % i)
   62         yield i
   63     Gtk.main_quit()
   64 
   65 
   66 class TestNotebookOperation(tests.TestCase):
   67 
   68     ## TODO add signal monitor to check step and finished emitted
   69 
   70     def testIterator(self):
   71         nb = MockNotebook()
   72         nb.test('Foo')
   73         self.assertEqual(nb.value, 'Foo')
   74 
   75         # Test iterator
   76         op = NotebookOperation(nb, 'My Op', mock_iterator(nb))
   77 
   78         self.assertFalse(op.is_running())
   79         nb.test('Bar')
   80         self.assertEqual(nb.value, 'Bar')
   81 
   82         i = None
   83         for i, x in enumerate(op):
   84             self.assertTrue(op.is_running())
   85             self.assertEqual(nb.value, 'Test %i' % i)
   86             self.assertRaises(NotebookOperationOngoing, nb.test, 'Foo')
   87 
   88         self.assertEqual(i, 2)
   89         self.assertFalse(op.is_running())
   90         self.assertFalse(op.cancelled)
   91 
   92         nb.test('Baz')
   93         self.assertEqual(nb.value, 'Baz')
   94 
   95         # Test cancel
   96         op = NotebookOperation(nb, 'My Op', mock_iterator(nb))
   97 
   98         i = None
   99         for i, x in enumerate(op):
  100             self.assertTrue(op.is_running())
  101             op.cancel()
  102 
  103         self.assertEqual(i, 0)
  104         self.assertFalse(op.is_running())
  105         self.assertTrue(op.cancelled)
  106 
  107     def testIdle(self):
  108         nb = MockNotebook()
  109         op = NotebookOperation(nb, 'My Op', mock_gtk_iter(nb))
  110         op.run_on_idle()
  111         Gtk.main()
  112         self.assertFalse(op.is_running())
  113         self.assertEqual(nb.value, 'Test %i' % 2)
  114 
  115     def testContext(self):
  116         nb = MockNotebook()
  117 
  118         def test():
  119             with NotebookState(nb):
  120                 nb.raw('Foo')
  121 
  122         test()
  123         self.assertEqual(nb.value, 'Foo')
  124 
  125         op = NotebookOperation(nb, 'My Op', mock_iterator(nb))
  126         for i, x in enumerate(op):
  127             self.assertRaises(NotebookOperationOngoing, test)
  128 
  129     def testSignals(self):
  130         nb = MockNotebook()
  131 
  132         op = NotebookOperation(nb, 'My Op', mock_iterator(nb))
  133         dialog = MockProgressDialog(None, op)
  134 
  135         i = None
  136         for i, x in enumerate(op):
  137             pass
  138 
  139         self.assertEqual(i, 2)
  140         self.assertEqual(len(dialog.steps), 3)
  141         self.assertTrue(dialog.finished)
  142 
  143         op = NotebookOperation(nb, 'My Op', mock_iterator(nb))
  144         dialog = MockProgressDialog(None, op)
  145 
  146         i = None
  147         for i, x in enumerate(op):
  148             self.assertTrue(op.is_running())
  149             op.cancel()
  150 
  151         self.assertEqual(i, 0)
  152         self.assertEqual(len(dialog.steps), 1)
  153         self.assertTrue(dialog.finished)
  154 
  155 
  156 import threading
  157 
  158 def mock_thread_main(notebook, lock):
  159     with lock:
  160         for i in 0, 1, 2:
  161             notebook.test('Test %i' % i)
  162 
  163 
  164 class TestSimpleAsyncOperation(tests.TestCase):
  165 
  166     def runTest(self):
  167         nb = MockNotebook()
  168         lock = threading.Lock()
  169         lock.acquire()
  170         thread = threading.Thread(target=mock_thread_main, args=(nb, lock))
  171         thread.start()
  172         # using lock to ensure thread doesn't finish before iteration seen
  173 
  174         result = []
  175         def post():
  176             result.append('foo')
  177 
  178         op = SimpleAsyncOperation(nb, 'my op', thread, post)
  179         for i, x in enumerate(op):
  180             self.assertTrue(op.is_running())
  181             if i == 1:
  182                 lock.release()
  183         self.assertTrue(i >= 1)
  184 
  185         self.assertFalse(op.is_running())
  186         self.assertFalse(op.cancelled)
  187 
  188         self.assertEqual(nb.value, 'Test 2')
  189         self.assertEqual(result, ['foo'])
  190 
  191         # now with cancel - result can vary depending on who goes first
  192         lock = threading.Lock()
  193         lock.acquire()
  194         thread = threading.Thread(target=mock_thread_main, args=(nb, lock))
  195         thread.start()
  196 
  197         op = SimpleAsyncOperation(nb, 'my op', thread, post)
  198         for i, x in enumerate(op):
  199             if i == 1:
  200                 lock.release()
  201                 op.cancel()
  202 
  203         self.assertFalse(op.is_running())
  204         self.assertTrue(op.cancelled)