Create structure for colab signals

Create class and methods to handle signal registration, connection and sending Signed-off-by: Lucas Kanashiro <kanashiro.duarte@gmail.com> Signed-off-by: Lucas Moura <lucas.moura128@gmail.com>

Create structure for colab signals
Create class and methods to handle signal registration, connection and sending Signed-off-by: Lucas Kanashiro <kanashiro.duarte@gmail.com> Signed-off-by: Lucas Moura <lucas.moura128@gmail.com>
Lucas Kanashiro · Sergio Oliveira
1 parent 9773eddc
Showing 7 changed files with 142 additions and 0 deletions Show diff stats
colab/settings.py
colab/signals/__init__.py
colab/signals/apps.py
colab/signals/celery.py
colab/signals/signals.py
colab/signals/tests/__init__.py
colab/signals/tests/test_signals.py
@@ -57,6 +57,7 @@ INSTALLED_APPS = (
     'colab.search',
     'colab.tz',
     'colab.utils',
+    'colab.signals',
 )
  
 ROOT_URLCONF = 'colab.urls'
@@ -0,0 +1,6 @@
+
+from django.apps import AppConfig
+
+
+class SignalsConfig(AppConfig):
+    name = 'colab.signals'
@@ -0,0 +1,22 @@
+from __future__ import absolute_import
+
+import os
+
+from celery import Celery
+
+# set the default Django settings module for the 'celery' program.
+os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'colab.settings')
+
+from django.conf import settings
+
+app = Celery('colab')
+
+app.config_from_object('django.conf:settings')
+app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
+
+app.conf.update(
+    CELERY_RESULT_BACKEND='djcelery.backends.database:DatabaseBackend',
+)
+app.conf.update(
+    CELERY_RESULT_BACKEND='djcelery.backends.cache:CacheBackend',
+)
@@ -0,0 +1,31 @@
+from django.dispatch import Signal
+import colab.signals.celery
+
+class Signals():
+    def __init__(self):
+        self.registered_signals = {}
+        self.signal_instances = {}
+
+
+    def register_signal(self, plugin_name, list_signals):
+        for signal in list_signals:
+            if signal in self.registered_signals:
+                if not plugin_name in self.registered_signals[signal]:
+                    self.registered_signals[signal].append(plugin_name)
+            else:
+                self.registered_signals[signal] = []
+                self.registered_signals[signal].append(plugin_name)
+                self.signal_instances[signal] = Signal()
+
+
+    def connect_signal(self, signal_name, sender, handling_method):
+        if signal_name in self.signal_instances:
+            self.signal_instances[signal_name].connect(handling_method,
+                    sender=sender)
+        else:
+            raise Exception("Signal does not exist!")
+
+
+    def send(self, signal_name, sender, **kwargs):
+        if signal_name in self.signal_instances:
+            self.signal_instances[signal_name].send(sender=sender)
@@ -0,0 +1,82 @@
+"""
+Test Signals class.
+Objective: Test parameters, and behavior.
+"""
+
+from django.test import TestCase
+from colab.signals.signals import Signals
+from mock import patch
+
+
+class SignalsTest(TestCase):
+
+    def setUp(self):
+       self.list_signal = ['a', 'b', 'c']
+       self.plugin_name = 'test_signal'
+       self.signals = Signals()
+
+
+    def test_register_signal_(self):
+       self.signals.register_signal(self.plugin_name, self.list_signal)
+
+       signal_name ='a'
+       signal_list = ['test_signal']
+       self.assertEqual(len(self.signals.registered_signals[signal_name]), 1)
+       self.assertEqual(self.signals.registered_signals[signal_name],
+               signal_list)
+
+   
+    def test_register_signal_already_registered(self):
+       signal_name ='a'
+       signal_list = ['test_signal']
+
+       self.signals.register_signal(self.plugin_name, self.list_signal)
+       self.assertEqual(len(self.signals.registered_signals[signal_name]), 1)
+
+       self.signals.register_signal(self.plugin_name, self.list_signal)
+       self.assertEqual(len(self.signals.registered_signals[signal_name]), 1)
+       self.assertEqual(self.signals.registered_signals[signal_name],
+               signal_list)
+
+    
+    def test_connect_non_registered_signal(self):
+        sender = 'Test'
+        handling_method = 'Test'
+        signal_name = 'Test'
+        
+        self.assertRaises(Exception, self.signals.connect_signal, signal_name,
+                sender, handling_method) 
+
+
+    @patch('colab.signals.signals.Signal.connect')
+    def test_connect_already_registered_signal(self, mock):
+        sender = 'Test'
+        handling_method = 'Test'
+        signal_name = 'a'
+        
+        self.signals.register_signal(self.plugin_name, self.list_signal)
+
+        self.signals.connect_signal(signal_name, sender, handling_method)
+        args, kwargs = mock.call_args
+        
+        self.assertEqual(args[0], handling_method)
+        self.assertEqual(kwargs['sender'], sender)
+        self.assertTrue(mock.is_called)
+        
+
+    @patch('colab.signals.signals.Signal.send')
+    def test_send_signal(self, mock):
+        sender = 'Test'
+        handling_method = 'Test'
+        signal_name = 'a'
+        
+        self.signals.register_signal(self.plugin_name, self.list_signal)
+        self.signals.send(signal_name, sender)
+
+        args, kwargs = mock.call_args
+
+        self.assertEqual(kwargs['sender'], sender)
+        self.assertTrue(mock.is_called)
+        
+
+
...	...	@@ -57,6 +57,7 @@ INSTALLED_APPS = (
57	57	'colab.search',
58	58	'colab.tz',
59	59	'colab.utils',
	60	+ 'colab.signals',
60	61	)
61	62
62	63	ROOT_URLCONF = 'colab.urls'
...	...
...	...	@@ -0,0 +1,6 @@
	1	+
	2	+from django.apps import AppConfig
	3	+
	4	+
	5	+class SignalsConfig(AppConfig):
	6	+ name = 'colab.signals'
...	...
...	...	@@ -0,0 +1,22 @@
	1	+from __future__ import absolute_import
	2	+
	3	+import os
	4	+
	5	+from celery import Celery
	6	+
	7	+# set the default Django settings module for the 'celery' program.
	8	+os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'colab.settings')
	9	+
	10	+from django.conf import settings
	11	+
	12	+app = Celery('colab')
	13	+
	14	+app.config_from_object('django.conf:settings')
	15	+app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
	16	+
	17	+app.conf.update(
	18	+ CELERY_RESULT_BACKEND='djcelery.backends.database:DatabaseBackend',
	19	+)
	20	+app.conf.update(
	21	+ CELERY_RESULT_BACKEND='djcelery.backends.cache:CacheBackend',
	22	+)
...	...
...	...	@@ -0,0 +1,31 @@
	1	+from django.dispatch import Signal
	2	+import colab.signals.celery
	3	+
	4	+class Signals():
	5	+ def __init__(self):
	6	+ self.registered_signals = {}
	7	+ self.signal_instances = {}
	8	+
	9	+
	10	+ def register_signal(self, plugin_name, list_signals):
	11	+ for signal in list_signals:
	12	+ if signal in self.registered_signals:
	13	+ if not plugin_name in self.registered_signals[signal]:
	14	+ self.registered_signals[signal].append(plugin_name)
	15	+ else:
	16	+ self.registered_signals[signal] = []
	17	+ self.registered_signals[signal].append(plugin_name)
	18	+ self.signal_instances[signal] = Signal()
	19	+
	20	+
	21	+ def connect_signal(self, signal_name, sender, handling_method):
	22	+ if signal_name in self.signal_instances:
	23	+ self.signal_instances[signal_name].connect(handling_method,
	24	+ sender=sender)
	25	+ else:
	26	+ raise Exception("Signal does not exist!")
	27	+
	28	+
	29	+ def send(self, signal_name, sender, **kwargs):
	30	+ if signal_name in self.signal_instances:
	31	+ self.signal_instances[signal_name].send(sender=sender)
...	...
...	...	@@ -0,0 +1,82 @@
	1	+"""
	2	+Test Signals class.
	3	+Objective: Test parameters, and behavior.
	4	+"""
	5	+
	6	+from django.test import TestCase
	7	+from colab.signals.signals import Signals
	8	+from mock import patch
	9	+
	10	+
	11	+class SignalsTest(TestCase):
	12	+
	13	+ def setUp(self):
	14	+ self.list_signal = ['a', 'b', 'c']
	15	+ self.plugin_name = 'test_signal'
	16	+ self.signals = Signals()
	17	+
	18	+
	19	+ def test_register_signal_(self):
	20	+ self.signals.register_signal(self.plugin_name, self.list_signal)
	21	+
	22	+ signal_name ='a'
	23	+ signal_list = ['test_signal']
	24	+ self.assertEqual(len(self.signals.registered_signals[signal_name]), 1)
	25	+ self.assertEqual(self.signals.registered_signals[signal_name],
	26	+ signal_list)
	27	+
	28	+
	29	+ def test_register_signal_already_registered(self):
	30	+ signal_name ='a'
	31	+ signal_list = ['test_signal']
	32	+
	33	+ self.signals.register_signal(self.plugin_name, self.list_signal)
	34	+ self.assertEqual(len(self.signals.registered_signals[signal_name]), 1)
	35	+
	36	+ self.signals.register_signal(self.plugin_name, self.list_signal)
	37	+ self.assertEqual(len(self.signals.registered_signals[signal_name]), 1)
	38	+ self.assertEqual(self.signals.registered_signals[signal_name],
	39	+ signal_list)
	40	+
	41	+
	42	+ def test_connect_non_registered_signal(self):
	43	+ sender = 'Test'
	44	+ handling_method = 'Test'
	45	+ signal_name = 'Test'
	46	+
	47	+ self.assertRaises(Exception, self.signals.connect_signal, signal_name,
	48	+ sender, handling_method)
	49	+
	50	+
	51	+ @patch('colab.signals.signals.Signal.connect')
	52	+ def test_connect_already_registered_signal(self, mock):
	53	+ sender = 'Test'
	54	+ handling_method = 'Test'
	55	+ signal_name = 'a'
	56	+
	57	+ self.signals.register_signal(self.plugin_name, self.list_signal)
	58	+
	59	+ self.signals.connect_signal(signal_name, sender, handling_method)
	60	+ args, kwargs = mock.call_args
	61	+
	62	+ self.assertEqual(args[0], handling_method)
	63	+ self.assertEqual(kwargs['sender'], sender)
	64	+ self.assertTrue(mock.is_called)
	65	+
	66	+
	67	+ @patch('colab.signals.signals.Signal.send')
	68	+ def test_send_signal(self, mock):
	69	+ sender = 'Test'
	70	+ handling_method = 'Test'
	71	+ signal_name = 'a'
	72	+
	73	+ self.signals.register_signal(self.plugin_name, self.list_signal)
	74	+ self.signals.send(signal_name, sender)
	75	+
	76	+ args, kwargs = mock.call_args
	77	+
	78	+ self.assertEqual(kwargs['sender'], sender)
	79	+ self.assertTrue(mock.is_called)
	80	+
	81	+
	82	+
...	...