Merge pull request #47 from avojnovicDk/signals

Added signals for user creation, update and delete.

Author: rdegges

flask_stormpath/models.py

@@ -3,10 +3,19 @@
 
 from flask import current_app
 from six import text_type
+
+from blinker import Namespace
+
 from stormpath.resources.account import Account
 from stormpath.resources.provider import Provider
 
 
+stormpath_signals = Namespace()
+user_created = stormpath_signals.signal('user-created')
+user_updated = stormpath_signals.signal('user-updated')
+user_deleted = stormpath_signals.signal('user-deleted')
+
+
 class User(Account):
     """
     The base User object.
@@ -44,6 +53,22 @@ def is_authenticated(self):
         """
         return True
 
+    def save(self):
+        """
+        Send signal after user is updated.
+        """
+        return_value = super(User, self).save()
+        user_updated.send(self, user=self)
+        return return_value
+
+    def delete(self):
+        """
+        Send signal after user is deleted.
+        """
+        return_value = super(User, self).delete()
+        user_deleted.send(self, user=self)
+        return return_value
+
     @classmethod
     def create(self, email, password, given_name, surname, username=None, middle_name=None, custom_data=None, status='ENABLED'):
         """
@@ -84,6 +109,7 @@ def create(self, email, password, given_name, surname, username=None, middle_nam
             'status': status,
         })
         _user.__class__ = User
+        user_created.send(self, user=_user)
 
         return _user
 

requirements.txt

@@ -1,3 +1,4 @@
 Sphinx>=1.2.1
 pytest>=2.5.2
 pytest-xdist>=1.10
+blinker==1.3

setup.py

@@ -57,6 +57,7 @@ def run(self):
         'facebook-sdk==0.4.0',
         'oauth2client==1.2',
         'stormpath==2.0.7',
+        'blinker==1.3'
     ],
     dependency_links=[
         'git+git://github.com/pythonforfacebook/facebook-sdk.git@e65d06158e48388b3932563f1483ca77065951b3#egg=facebook-sdk-1.0.0-alpha',

tests/helpers.py

@@ -43,6 +43,15 @@ def tearDown(self):
             directory.delete()
 
 
+class SignalReceiver(object):
+    received_signals = None
+
+    def signal_user_receiver_function(self, sender, user):
+        if self.received_signals is None:
+            self.received_signals = []
+        self.received_signals.append((sender, user))
+
+
 def bootstrap_client():
     """
     Create a new Stormpath Client from environment variables.

tests/test_signals.py

@@ -0,0 +1,128 @@
+"""Run tests for signals."""
+
+from flask.ext.login import user_logged_in
+from flask.ext.stormpath.models import (
+    User,
+    user_created,
+    user_deleted,
+    user_updated
+)
+
+from .helpers import StormpathTestCase, SignalReceiver
+
+
+class TestSignals(StormpathTestCase):
+    """Test signals."""
+
+    def test_user_created_signal(self):
+        # Subscribe to signals for user creation
+        signal_receiver = SignalReceiver()
+        user_created.connect(signal_receiver.signal_user_receiver_function)
+
+        # Register new account
+        with self.app.test_client() as c:
+            resp = c.post('/register', data={
+                'given_name': 'Randall',
+                'middle_name': 'Clark',
+                'surname': 'Degges',
+                'email': '[email protected]',
+                'password': 'woot1LoveCookies!',
+            })
+            self.assertEqual(resp.status_code, 302)
+
+        # Check that signal for user creation is received
+        self.assertEqual(len(signal_receiver.received_signals), 1)
+        received_signal = signal_receiver.received_signals[0]
+        # User instance is received
+        self.assertIsInstance(received_signal[1], User)
+        # Correct user instance is received
+        created_user = received_signal[1]
+        self.assertEqual(created_user.email, '[email protected]')
+        self.assertEqual(created_user.surname, 'Degges')
+
+    def test_user_logged_in_signal(self):
+        # Subscribe to signals for user login
+        signal_receiver = SignalReceiver()
+        user_logged_in.connect(signal_receiver.signal_user_receiver_function)
+
+        # Create a user.
+        with self.app.app_context():
+            User.create(
+                username = 'rdegges',
+                given_name = 'Randall',
+                surname = 'Degges',
+                email = '[email protected]',
+                password = 'woot1LoveCookies!',
+            )
+
+        # Attempt a login using username and password.
+        with self.app.test_client() as c:
+            resp = c.post('/login', data={
+                'login': 'rdegges',
+                'password': 'woot1LoveCookies!',
+            })
+            self.assertEqual(resp.status_code, 302)
+
+        # Check that signal for user login is received
+        self.assertEqual(len(signal_receiver.received_signals), 1)
+        received_signal = signal_receiver.received_signals[0]
+        # User instance is received
+        self.assertIsInstance(received_signal[1], User)
+        # Correct user instance is received
+        logged_in_user = received_signal[1]
+        self.assertEqual(logged_in_user.email, '[email protected]')
+        self.assertEqual(logged_in_user.surname, 'Degges')
+
+    def test_user_is_updated_signal(self):
+        # Subscribe to signals for user update
+        signal_receiver = SignalReceiver()
+        user_updated.connect(signal_receiver.signal_user_receiver_function)
+
+        with self.app.app_context():
+
+            # Ensure all requied fields are properly set.
+            user = User.create(
+                email = '[email protected]',
+                password = 'woot1LoveCookies!',
+                given_name = 'Randall',
+                surname = 'Degges',
+            )
+
+            user.middle_name = 'Clark'
+            user.save()
+
+        # Check that signal for user update is received
+        self.assertEqual(len(signal_receiver.received_signals), 1)
+        received_signal = signal_receiver.received_signals[0]
+        # User instance is received
+        self.assertIsInstance(received_signal[1], User)
+        # Correct user instance is received
+        updated_user = received_signal[1]
+        self.assertEqual(updated_user.email, '[email protected]')
+        self.assertEqual(updated_user.middle_name, 'Clark')
+
+    def test_user_is_deleted_signal(self):
+        # Subscribe to signals for user delete
+        signal_receiver = SignalReceiver()
+        user_deleted.connect(signal_receiver.signal_user_receiver_function)
+
+        with self.app.app_context():
+
+            # Ensure all requied fields are properly set.
+            user = User.create(
+                email = '[email protected]',
+                password = 'woot1LoveCookies!',
+                given_name = 'Randall',
+                surname = 'Degges',
+            )
+
+            user.delete()
+
+        # Check that signal for user deletion is received
+        self.assertEqual(len(signal_receiver.received_signals), 1)
+        received_signal = signal_receiver.received_signals[0]
+        # User instance is received
+        self.assertIsInstance(received_signal[1], User)
+        # Correct user instance is received
+        deleted_user = received_signal[1]
+        self.assertEqual(deleted_user.email, '[email protected]')