summaryrefslogtreecommitdiff
path: root/src/leap/gui
diff options
context:
space:
mode:
Diffstat (limited to 'src/leap/gui')
-rw-r--r--src/leap/gui/progress.py95
-rw-r--r--src/leap/gui/tests/__init__.py0
-rw-r--r--src/leap/gui/tests/test_mainwindow_rc.py (renamed from src/leap/gui/test_mainwindow_rc.py)3
-rw-r--r--src/leap/gui/tests/test_progress.py284
-rw-r--r--src/leap/gui/tests/test_threads.py27
5 files changed, 380 insertions, 29 deletions
diff --git a/src/leap/gui/progress.py b/src/leap/gui/progress.py
index 64b87b2c..e68c35d2 100644
--- a/src/leap/gui/progress.py
+++ b/src/leap/gui/progress.py
@@ -4,7 +4,7 @@ from first run wizard
"""
try:
from collections import OrderedDict
-except ImportError:
+except ImportError: # pragma: no cover
# We must be in 2.6
from leap.util.dicts import OrderedDict
@@ -73,15 +73,16 @@ class ProgressStepContainer(object):
self.steps = {}
def step(self, identity):
- return self.step.get(identity)
+ return self.steps.get(identity, None)
def addStep(self, step):
self.steps[step.index] = step
def removeStep(self, step):
- del self.steps[step.index]
- del step
- self.dirty = True
+ if step and self.steps.get(step.index, None):
+ del self.steps[step.index]
+ del step
+ self.dirty = True
def removeAllSteps(self):
for item in iter(self):
@@ -107,7 +108,7 @@ class StepsTableWidget(QtGui.QTableWidget):
"""
def __init__(self, parent=None):
- super(StepsTableWidget, self).__init__(parent)
+ super(StepsTableWidget, self).__init__(parent=parent)
# remove headers and all edit/select behavior
self.horizontalHeader().hide()
@@ -149,18 +150,39 @@ class StepsTableWidget(QtGui.QTableWidget):
class WithStepsMixIn(object):
+ """
+ This Class is a mixin that can be inherited
+ by InlineValidation pages (which will display
+ a progress steps widget in the same page as the form)
+ or by Validation Pages (which will only display
+ the progress steps in the page, below a progress bar widget)
+ """
+ STEPS_TIMER_MS = 100
- # worker threads for checks
+ #
+ # methods related to worker threads
+ # launched for individual checks
+ #
def setupStepsProcessingQueue(self):
+ """
+ should be called from the init method
+ of the derived classes
+ """
self.steps_queue = Queue.Queue()
self.stepscheck_timer = QtCore.QTimer()
self.stepscheck_timer.timeout.connect(self.processStepsQueue)
- self.stepscheck_timer.start(100)
+ self.stepscheck_timer.start(self.STEPS_TIMER_MS)
# we need to keep a reference to child threads
self.threads = []
def do_checks(self):
+ """
+ main entry point for checks.
+ it calls _do_checks in derived classes,
+ and it expects it to be a generator
+ yielding a tuple in the form (("message", progress_int), checkfunction)
+ """
# yo dawg, I heard you like checks
# so I put a __do_checks in your do_checks
@@ -168,7 +190,7 @@ class WithStepsMixIn(object):
def __do_checks(fun=None, queue=None):
- for checkcase in fun():
+ for checkcase in fun(): # pragma: no cover
checkmsg, checkfun = checkcase
queue.put(checkmsg)
@@ -180,15 +202,34 @@ class WithStepsMixIn(object):
__do_checks,
fun=self._do_checks,
queue=self.steps_queue))
- t.finished.connect(self.on_checks_validation_ready)
+ if hasattr(self, 'on_checks_validation_ready'):
+ t.finished.connect(self.on_checks_validation_ready)
t.begin()
self.threads.append(t)
+ def processStepsQueue(self):
+ """
+ consume steps queue
+ and pass messages
+ to the ui updater functions
+ """
+ while self.steps_queue.qsize():
+ try:
+ status = self.steps_queue.get(0)
+ if status == "failed":
+ self.set_failed_icon()
+ else:
+ self.onStepStatusChanged(*status)
+ except Queue.Empty: # pragma: no cover
+ pass
+
def fail(self, err=None):
"""
return failed state
and send error notification as
- a nice side effect
+ a nice side effect. this function is called from
+ the _do_checks check functions returned in the
+ generator.
"""
wizard = self.wizard()
senderr = lambda err: wizard.set_validation_error(
@@ -202,38 +243,29 @@ class WithStepsMixIn(object):
def launch_checks(self):
self.do_checks()
+ # (gui) presentation stuff begins #####################
+
# slot
#@QtCore.pyqtSlot(str, int)
def onStepStatusChanged(self, status, progress=None):
if status not in ("head_sentinel", "end_sentinel"):
self.add_status_line(status)
if status in ("end_sentinel"):
- self.checks_finished = True
+ #self.checks_finished = True
self.set_checked_icon()
if progress and hasattr(self, 'progress'):
self.progress.setValue(progress)
self.progress.update()
- def processStepsQueue(self):
- """
- consume steps queue
- and pass messages
- to the ui updater functions
- """
- while self.steps_queue.qsize():
- try:
- status = self.steps_queue.get(0)
- if status == "failed":
- self.set_failed_icon()
- else:
- self.onStepStatusChanged(*status)
- except Queue.Empty:
- pass
-
def setupSteps(self):
self.steps = ProgressStepContainer()
# steps table widget
- self.stepsTableWidget = StepsTableWidget(self)
+ if isinstance(self, QtCore.QObject):
+ parent = self
+ else:
+ parent = None
+ import ipdb;ipdb.set_trace()
+ self.stepsTableWidget = StepsTableWidget(parent=parent)
zeros = (0, 0, 0, 0)
self.stepsTableWidget.setContentsMargins(*zeros)
self.errors = OrderedDict()
@@ -295,6 +327,8 @@ class WithStepsMixIn(object):
# setting cell widget.
# see note on StepsTableWidget about plans to
# change this for a better solution.
+ if not hasattr(self, 'steps'):
+ return
index = len(self.steps)
table = self.stepsTableWidget
_index = index - 1 if current else index - 2
@@ -340,6 +374,9 @@ class WithStepsMixIn(object):
def is_done(self):
return self.done
+ # convenience for going back and forth
+ # in the wizard pages.
+
def go_back(self):
self.wizard().back()
diff --git a/src/leap/gui/tests/__init__.py b/src/leap/gui/tests/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/src/leap/gui/tests/__init__.py
diff --git a/src/leap/gui/test_mainwindow_rc.py b/src/leap/gui/tests/test_mainwindow_rc.py
index c5abb4aa..67b9fae0 100644
--- a/src/leap/gui/test_mainwindow_rc.py
+++ b/src/leap/gui/tests/test_mainwindow_rc.py
@@ -27,3 +27,6 @@ class MainWindowResourcesTest(unittest.TestCase):
self.assertEqual(
hashlib.md5(mainwindow_rc.qt_resource_data).hexdigest(),
'53e196f29061d8f08f112e5a2e64eb53')
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/src/leap/gui/tests/test_progress.py b/src/leap/gui/tests/test_progress.py
new file mode 100644
index 00000000..ff6a0bf1
--- /dev/null
+++ b/src/leap/gui/tests/test_progress.py
@@ -0,0 +1,284 @@
+import sys
+import unittest
+import Queue
+
+import mock
+
+from leap.testing import qunittest
+from leap.testing import pyqt
+
+from PyQt4 import QtGui
+from PyQt4 import QtCore
+from PyQt4.QtTest import QTest
+from PyQt4.QtCore import Qt
+
+from leap.gui import progress
+
+
+class ProgressStepTestCase(unittest.TestCase):
+
+ def test_step_attrs(self):
+ ps = progress.ProgressStep
+ step = ps('test', False, 1)
+ # instance
+ self.assertEqual(step.index, 1)
+ self.assertEqual(step.name, "test")
+ self.assertEqual(step.done, False)
+ step = ps('test2', True, 2)
+ self.assertEqual(step.index, 2)
+ self.assertEqual(step.name, "test2")
+ self.assertEqual(step.done, True)
+
+ # class methods and attrs
+ self.assertEqual(ps.columns(), ('name', 'done'))
+ self.assertEqual(ps.NAME, 0)
+ self.assertEqual(ps.DONE, 1)
+
+
+class ProgressStepContainerTestCase(unittest.TestCase):
+ def setUp(self):
+ self.psc = progress.ProgressStepContainer()
+
+ def addSteps(self, number):
+ Step = progress.ProgressStep
+ for n in range(number):
+ self.psc.addStep(Step("%s" % n, False, n))
+
+ def test_attrs(self):
+ self.assertEqual(self.psc.columns,
+ ('name', 'done'))
+
+ def test_add_steps(self):
+ Step = progress.ProgressStep
+ self.assertTrue(len(self.psc) == 0)
+ self.psc.addStep(Step('one', False, 0))
+ self.assertTrue(len(self.psc) == 1)
+ self.psc.addStep(Step('two', False, 1))
+ self.assertTrue(len(self.psc) == 2)
+
+ def test_del_all_steps(self):
+ self.assertTrue(len(self.psc) == 0)
+ self.addSteps(5)
+ self.assertTrue(len(self.psc) == 5)
+ self.psc.removeAllSteps()
+ self.assertTrue(len(self.psc) == 0)
+
+ def test_del_step(self):
+ Step = progress.ProgressStep
+ self.addSteps(5)
+ self.assertTrue(len(self.psc) == 5)
+ self.psc.removeStep(self.psc.step(4))
+ self.assertTrue(len(self.psc) == 4)
+ self.psc.removeStep(self.psc.step(4))
+ self.psc.removeStep(Step('none', False, 5))
+ self.psc.removeStep(self.psc.step(4))
+
+ def test_iter(self):
+ self.addSteps(10)
+ self.assertEqual(
+ [x.index for x in self.psc],
+ [x for x in range(10)])
+
+
+class StepsTableWidgetTestCase(unittest.TestCase):
+
+ def setUp(self):
+ self.app = QtGui.QApplication(sys.argv)
+ QtGui.qApp = self.app
+ self.stw = progress.StepsTableWidget()
+
+ def tearDown(self):
+ QtGui.qApp = None
+ self.app = None
+
+ def test_defaults(self):
+ self.assertTrue(isinstance(self.stw, QtGui.QTableWidget))
+ self.assertEqual(self.stw.focusPolicy(), 0)
+
+
+class TestWithStepsClass(QtGui.QWidget, progress.WithStepsMixIn):
+
+ def __init__(self):
+ self.setupStepsProcessingQueue()
+ self.statuses = []
+ self.current_page = "testpage"
+
+ def onStepStatusChanged(self, *args):
+ """
+ blank out this gui method
+ that will add status lines
+ """
+ self.statuses.append(args)
+
+
+class WithStepsMixInTestCase(qunittest.TestCase):
+
+ TIMER_WAIT = 2 * progress.WithStepsMixIn.STEPS_TIMER_MS / 1000.0
+
+ # XXX can spy on signal connections
+
+ def setUp(self):
+ self.app = QtGui.QApplication(sys.argv)
+ QtGui.qApp = self.app
+ self.stepy = TestWithStepsClass()
+ #self.connects = []
+ #pyqt.enableSignalDebugging(
+ #connectCall=lambda *args: self.connects.append(args))
+ #self.assertEqual(self.connects, [])
+ #self.stepy.stepscheck_timer.timeout.disconnect(
+ #self.stepy.processStepsQueue)
+
+ def tearDown(self):
+ QtGui.qApp = None
+ self.app = None
+
+ def test_has_queue(self):
+ s = self.stepy
+ self.assertTrue(hasattr(s, 'steps_queue'))
+ self.assertTrue(isinstance(s.steps_queue, Queue.Queue))
+ self.assertTrue(isinstance(s.stepscheck_timer, QtCore.QTimer))
+
+ def test_do_checks_delegation(self):
+ s = self.stepy
+
+ _do_checks = mock.Mock()
+ _do_checks.return_value = (
+ (("test", 0), lambda: None),
+ (("test", 0), lambda: None))
+ s._do_checks = _do_checks
+ s.do_checks()
+ self.waitFor(seconds=self.TIMER_WAIT)
+ _do_checks.assert_called_with()
+ self.assertEqual(len(s.statuses), 2)
+
+ # test that a failed test interrupts the run
+
+ s.statuses = []
+ _do_checks = mock.Mock()
+ _do_checks.return_value = (
+ (("test", 0), lambda: None),
+ (("test", 0), lambda: False),
+ (("test", 0), lambda: None))
+ s._do_checks = _do_checks
+ s.do_checks()
+ self.waitFor(seconds=self.TIMER_WAIT)
+ _do_checks.assert_called_with()
+ self.assertEqual(len(s.statuses), 2)
+
+ def test_process_queue(self):
+ s = self.stepy
+ q = s.steps_queue
+ s.set_failed_icon = mock.MagicMock()
+ with self.assertRaises(AssertionError):
+ q.put('foo')
+ self.waitFor(seconds=self.TIMER_WAIT)
+ s.set_failed_icon.assert_called_with()
+ q.put("failed")
+ self.waitFor(seconds=self.TIMER_WAIT)
+ s.set_failed_icon.assert_called_with()
+
+ def test_on_checks_validation_ready_called(self):
+ s = self.stepy
+ s.on_checks_validation_ready = mock.MagicMock()
+
+ _do_checks = mock.Mock()
+ _do_checks.return_value = (
+ (("test", 0), lambda: None),)
+ s._do_checks = _do_checks
+ s.do_checks()
+
+ self.waitFor(seconds=self.TIMER_WAIT)
+ s.on_checks_validation_ready.assert_called_with()
+
+ def test_fail(self):
+ s = self.stepy
+
+ s.wizard = mock.Mock()
+ wizard = s.wizard.return_value
+ wizard.set_validation_error.return_value = True
+ s.completeChanged = mock.Mock()
+ s.completeChanged.emit.return_value = True
+
+ self.assertFalse(s.fail(err="foo"))
+ self.waitFor(seconds=self.TIMER_WAIT)
+ wizard.set_validation_error.assert_called_with('testpage', 'foo')
+ s.completeChanged.emit.assert_called_with()
+
+ # with no args
+ s.wizard = mock.Mock()
+ wizard = s.wizard.return_value
+ wizard.set_validation_error.return_value = True
+ s.completeChanged = mock.Mock()
+ s.completeChanged.emit.return_value = True
+
+ self.assertFalse(s.fail())
+ self.waitFor(seconds=self.TIMER_WAIT)
+ with self.assertRaises(AssertionError):
+ wizard.set_validation_error.assert_called_with()
+ s.completeChanged.emit.assert_called_with()
+
+ def test_done(self):
+ s = self.stepy
+ s.done = False
+
+ s.completeChanged = mock.Mock()
+ s.completeChanged.emit.return_value = True
+
+ self.assertFalse(s.is_done())
+ s.set_done()
+ self.assertTrue(s.is_done())
+ s.completeChanged.emit.assert_called_with()
+
+ s.completeChanged = mock.Mock()
+ s.completeChanged.emit.return_value = True
+ s.set_undone()
+ self.assertFalse(s.is_done())
+
+ def test_back_and_next(self):
+ s = self.stepy
+ s.wizard = mock.Mock()
+ wizard = s.wizard.return_value
+ wizard.back.return_value = True
+ wizard.next.return_value = True
+ s.go_back()
+ wizard.back.assert_called_with()
+ s.go_next()
+ wizard.next.assert_called_with()
+
+ def test_on_step_statuschanged_slot(self):
+ s = self.stepy
+ s.onStepStatusChanged = progress.WithStepsMixIn.onStepStatusChanged
+ s.add_status_line = mock.Mock()
+ s.set_checked_icon = mock.Mock()
+ s.progress = mock.Mock()
+ s.progress.setValue.return_value = True
+ s.progress.update.return_value = True
+
+ s.onStepStatusChanged(s, "end_sentinel")
+ s.set_checked_icon.assert_called_with()
+
+ s.onStepStatusChanged(s, "foo")
+ s.add_status_line.assert_called_with("foo")
+
+ s.onStepStatusChanged(s, "bar", 42)
+ s.progress.setValue.assert_called_with(42)
+ s.progress.update.assert_called_with()
+
+ def test_steps_and_errors(self):
+ s = self.stepy
+ s.setupSteps()
+ self.assertTrue(isinstance(s.steps, progress.ProgressStepContainer))
+ self.assertEqual(s.errors, {})
+
+
+
+class InlineValidationPageTestCase(unittest.TestCase):
+ pass
+
+
+class ValidationPage(unittest.TestCase):
+ pass
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/src/leap/gui/tests/test_threads.py b/src/leap/gui/tests/test_threads.py
new file mode 100644
index 00000000..06c19606
--- /dev/null
+++ b/src/leap/gui/tests/test_threads.py
@@ -0,0 +1,27 @@
+import unittest
+
+import mock
+from leap.gui import threads
+
+
+class FunThreadTestCase(unittest.TestCase):
+
+ def setUp(self):
+ self.fun = mock.MagicMock()
+ self.fun.return_value = "foo"
+ self.t = threads.FunThread(fun=self.fun)
+
+ def test_thread(self):
+ self.t.begin()
+ self.t.wait()
+ self.fun.assert_called()
+ del self.t
+
+ def test_run(self):
+ # this is called by PyQt
+ self.t.run()
+ del self.t
+ self.fun.assert_called()
+
+if __name__ == "__main__":
+ unittest.main()
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-09 15:11:17 +0000