Test coverage

force_bdss/core_plugins/dummy/dummy_notification_listener/dummy_notification_listener.py

@@ -12,9 +12,11 @@ class DummyNotificationListener(BaseNotificationListener):
             print(event.__class__.__name__)
         elif isinstance(event, MCOProgressEvent):
             print(event.__class__.__name__, event.input, event.output)
+        else:
+            print(event.__class__.__name__)
 
     def initialize(self, model):
-        print("Initializing persistent state")
+        print("Initializing")
 
     def finalize(self, model):
-        print("Finalizing persistent state")
+        print("Finalizing")

force_bdss/core_plugins/dummy/dummy_notification_listener/tests/test_dummy_notification_listener.py

+import unittest
+
+from force_bdss.api import BaseMCOEvent
+from force_bdss.notification_listeners.base_notification_listener_factory \
+    import \
+    BaseNotificationListenerFactory
+from force_bdss.notification_listeners.base_notification_listener_model \
+    import \
+    BaseNotificationListenerModel
+from force_bdss.tests.utils import captured_output
+
+try:
+    import mock
+except ImportError:
+    from unittest import mock
+
+from force_bdss.core_plugins.dummy.dummy_notification_listener \
+    .dummy_notification_listener import \
+    DummyNotificationListener
+
+
+class TestDummyNotificationListener(unittest.TestCase):
+    def test_initialization(self):
+        listener = DummyNotificationListener(
+            mock.Mock(spec=BaseNotificationListenerFactory))
+        model = mock.Mock(spec=BaseNotificationListenerModel)
+        event = mock.Mock(spec=BaseMCOEvent)
+        with captured_output() as (out, err):
+            listener.initialize(model)
+            listener.deliver(model, event)
+            listener.finalize(model)
+
+        self.assertEqual(out.getvalue(),
+                         "Initializing\nBaseMCOEvent\nFinalizing\n")

force_bdss/core_plugins/dummy/dummy_notification_listener/tests/test_dummy_notification_listener_factory.py

+
+import unittest
+
+from envisage.plugin import Plugin
+
+from force_bdss.core_plugins.dummy.dummy_notification_listener\
+    .dummy_notification_listener_factory import \
+    DummyNotificationListenerFactory
+
+try:
+    import mock
+except ImportError:
+    from unittest import mock
+
+
+class TestDummyNotificationListenerFactory(unittest.TestCase):
+    def test_create_methods(self):
+        factory = DummyNotificationListenerFactory(mock.Mock(spec=Plugin))
+        model = factory.create_model()
+        self.assertEqual(model.factory, factory)
+
+        listener = factory.create_listener()
+        self.assertEqual(listener.factory, factory)

force_bdss/core_plugins/dummy/dummy_notification_listener/tests/test_dummy_notification_listener_model.py

+import unittest
+
+from envisage.plugin import Plugin
+
+from force_bdss.core_plugins.dummy.dummy_notification_listener\
+    .dummy_notification_listener_factory import \
+    DummyNotificationListenerFactory
+from force_bdss.core_plugins.dummy.dummy_notification_listener\
+    .dummy_notification_listener_model import \
+    DummyNotificationListenerModel
+
+try:
+    import mock
+except ImportError:
+    from unittest import mock
+
+
+class TestDummyNotificationListenerModel(unittest.TestCase):
+    def test_initialization(self):
+        factory = mock.Mock(spec=DummyNotificationListenerFactory)
+        model = DummyNotificationListenerModel(factory)
+        self.assertEqual(model.factory, factory)

force_bdss/core_plugins/dummy/ui_notification/tests/test_ui_notification.py

+import unittest
+
+from force_bdss.core_plugins.dummy.ui_notification.ui_notification import \
+    UINotification
+from force_bdss.core_plugins.dummy.ui_notification.ui_notification_factory \
+    import \
+    UINotificationFactory
+from force_bdss.core_plugins.dummy.ui_notification.ui_notification_model \
+    import \
+    UINotificationModel
+from force_bdss.mco.events import MCOStartEvent
+
+try:
+    import mock
+except ImportError:
+    from unittest import mock
+
+import zmq
+
+
+class TestUINotification(unittest.TestCase):
+    def setUp(self):
+        self.context = mock.Mock(spec=zmq.Context)
+        self.rep_socket = mock.Mock(spec=zmq.Socket)
+        self.pub_socket = mock.Mock(spec=zmq.Socket)
+
+        listener = UINotification(
+            mock.Mock(spec=UINotificationFactory)
+        )
+        listener._context = self.context
+        listener._rep_socket = self.rep_socket
+        listener._pub_socket = self.pub_socket
+        listener._rep_socket.recv.return_value = "SYNC".encode("utf-8")
+        self.listener = listener
+
+    def test_deliver(self):
+        listener = self.listener
+        listener.deliver(mock.Mock(spec=UINotificationModel), MCOStartEvent())
+        self.assertEqual(
+            listener._rep_socket.send_multipart.call_args[0][0],
+            ['EVENT\nMCO_START'.encode('utf-8')])
+        self.assertEqual(
+            listener._pub_socket.send.call_args[0][0],
+            'EVENT\nMCO_START'.encode('utf-8'))
+
+    def test_finalize(self):
+        listener = self.listener
+        listener.finalize(mock.Mock())
+        self.assertTrue(self.context.term.called)
+        self.assertTrue(self.rep_socket.close.called)
+        self.assertTrue(self.pub_socket.close.called)
+        self.assertIsNone(listener._context)
+        self.assertIsNone(listener._rep_socket)
+        self.assertIsNone(listener._pub_socket)
+
+    def test_initialize(self):
+        with mock.patch('force_bdss.core_plugins.dummy.ui_notification.'
+                        'ui_notification.zmq') as mock_zmq:
+            mock_zmq.Context.return_value = self.context
+            self.context.socket.return_value = self.rep_socket
+            listener = UINotification(
+                mock.Mock(spec=UINotificationFactory)
+            )
+            listener.initialize(mock.Mock())
+            self.assertTrue(listener._pub_socket.bind.called)

force_bdss/core_plugins/dummy/ui_notification/ui_notification.py

 import errno
 import logging
-from traits.api import Any, List
+from traits.api import List, Instance
 
 from force_bdss.api import (
     BaseNotificationListener,
@@ -17,14 +17,14 @@ class UINotification(BaseNotificationListener):
     Notification engine for the UI. Uses zeromq for the traffic handling.
     """
     #: The ZMQ context.
-    _context = Any()
+    _context = Instance(zmq.Context)
 
     #: The pubsub socket.
-    _pub_socket = Any()
+    _pub_socket = Instance(zmq.Socket)
 
     #: The synchronization socket to recover already sent information at a
     #: later stage
-    _rep_socket = Any()
+    _rep_socket = Instance(zmq.Socket)
 
     #: The cache of messages as they are sent out.
     _msg_cache = List()
@@ -42,12 +42,13 @@ class UINotification(BaseNotificationListener):
         if data and data[0:4] == "SYNC".encode("utf-8"):
             self._rep_socket.send_multipart(self._msg_cache)
 
-        msg = self._format_event(event)
+        msg = _format_event(event)
         if msg is not None:
             self._msg_cache.append(msg)
             self._pub_socket.send(msg)
 
     def initialize(self, model):
+        print("XX")
         self._context = zmq.Context()
         self._pub_socket = self._context.socket(zmq.PUB)
         self._pub_socket.bind("tcp://*:12345")
@@ -64,17 +65,19 @@ class UINotification(BaseNotificationListener):
         self._rep_socket = None
         self._context = None
 
-    def _format_event(self, event):
-        if isinstance(event, MCOStartEvent):
-            data = "MCO_START"
-        elif isinstance(event, MCOFinishEvent):
-            data = "MCO_FINISH"
-        elif isinstance(event, MCOProgressEvent):
-            data = "MCO_PROGRESS\n{}\n{}".format(
-                " ".join([str(x) for x in event.input]),
-                " ".join([str(x) for x in event.output]))
-        else:
-            return None
-
-        return ("EVENT\n{}".format(data)).encode("utf-8")
+
+def _format_event(event):
+    """Converts the event into a byte sequence to be transferred via zmq"""
+    if isinstance(event, MCOStartEvent):
+        data = "MCO_START"
+    elif isinstance(event, MCOFinishEvent):
+        data = "MCO_FINISH"
+    elif isinstance(event, MCOProgressEvent):
+        data = "MCO_PROGRESS\n{}\n{}".format(
+            " ".join([str(x) for x in event.input]),
+            " ".join([str(x) for x in event.output]))
+    else:
+        return None
+
+    return ("EVENT\n{}".format(data)).encode("utf-8")
 

force_bdss/tests/utils.py

+import sys
+from contextlib import contextmanager
+from six import StringIO
+
+@contextmanager
+def captured_output():
+    new_out, new_err = StringIO(), StringIO()
+    old_out, old_err = sys.stdout, sys.stderr
+    try:
+        sys.stdout, sys.stderr = new_out, new_err
+        yield sys.stdout, sys.stderr
+    finally:
+        sys.stdout, sys.stderr = old_out, old_err