Skip to content
Browse files

Testing of create-succeed/fail-destroy flow of the Django/Nimbus User…

…s now working.
  • Loading branch information...
1 parent 0014e6d commit 28f8e128724b5446a1db00c88e1cd87f9003471a @clemesha-ooi clemesha-ooi committed
View
9 web/src/python/nimbusweb/portal/nimbus/models.py
@@ -1,3 +1,4 @@
+import sys
from django.db import models
from django.contrib.auth.models import User, UserManager
import remote
@@ -28,9 +29,5 @@ class UserProfile(models.Model):
query_secret = models.TextField(null=True)
nimbus_userid = models.TextField(null=True)
-# register userprofile with the django auth system
-def user_post_save(sender, instance, **kwargs):
- profile, new = UserProfile.objects.get_or_create(user=instance)
-models.signals.post_save.connect(user_post_save, User)
-
-models.signals.post_save.connect(remote.nimbus_user_create, User)
+if sys.argv.count("test") != 1: #are we running tests? better way?
+ models.signals.post_save.connect(remote.nimbus_user_create, User)
View
35 web/src/python/nimbusweb/portal/nimbus/remote.py
@@ -1,18 +1,18 @@
import sys
from django.conf import settings
-from nimbusrest.connection import Connection
+from nimbusrest.admin.connection import AdminConnection
import models
-def nimbus_user_create_remote(user_instance, path="/users/create", nimbus_key=None, nimbus_secret=None):
+
+def nimbus_user_create_remote(user_instance, nimbus_key=None, nimbus_secret=None):
"""Use the Nimbus API to register a new Nimbus User.
`user_instance` is a `Django User` instance.
"""
- service_uri = getattr(settings, "NIMBUS_SERVICE_URI", "http://some.sensible.default/")
- conn = Connection(service_uri, nimbus_key, nimbus_secret)
- body = {"username":user_instance.username} #XXX what else is needed?
- response = conn.post_json(path, body) #let caller handle errors
- return response
+ service_uri = getattr(settings, "NIMBUS_SERVICE_URI", None)
+ conn = AdminConnection(service_uri, nimbus_key, nimbus_secret)
+ nimbus_user = conn.add_user(user_instance)
+ return nimbus_user
def nimbus_user_create(sender, instance, **kwargs):
"""Django User model `post_save` function.
@@ -21,22 +21,21 @@ def nimbus_user_create(sender, instance, **kwargs):
creation attempt, or, in the case of failure, remove
the recently created `Django User` and send failure
message back to User create Form.
+
+ Notes:
+ - Only attempt to create Nimbus User on Django User creation.
"""
- print "complete_nimbus_user_create => ", sender, instance, kwargs
- remote_user_creator = kwargs.get("remote_user_creator")
- if remote_user_creator is None:
- remote_user_creator = nimbus_user_create_remote
- # only attempt to create Nimbus User on Django User creation.
if kwargs.get('created'):
- response = remote_user_creator(instance)
+ remote_user_creator = kwargs.get("remote_user_creator")
+ if remote_user_creator is None:
+ remote_user_creator = nimbus_user_create_remote
try:
- response = remote_user_creator(instance)
+ nimbus_user = remote_user_creator(instance)
except:
#Nimbus User failed to be created, delete Django User
instance.delete() #XXX how to handle this?
- raise Exception(sys.exc_type)
- up = models.UserProfile.objects.get(user=instance)
- up.nimbus_userid = response["nimbus_userid"]
+ raise Exception(sys.exc_info())
+ up, created = models.UserProfile.objects.get_or_create(user=instance)
+ up.nimbus_userid = nimbus_user.user_id
up.save()
return True
-
View
39 web/src/python/nimbusweb/portal/nimbus/tests.py
@@ -2,42 +2,51 @@
from django.test import TestCase
from django.contrib.auth.models import User
+from django.conf import settings
+#from nimbusrest.admin import User as NimbusUser
import remote
-class FakeUser(object):
- username = "test_username"
- email = "test@example.com"
- password = "test_password"
+class FakeNimbusUser(object):
+ user_id = "test_nimbus_userid"
class CreateNimbusUserTest(unittest.TestCase):
def setUp(self):
- #self.user = User.objects.create_user("test_username", "test@email.com", "test_password")
- #creating a real user kicks off a real 'models.signals.post_save.connect'! So do this:
- self.user = FakeUser()
- self.ok_resp = {"nimbus_userid":"abc123"} #, "state":"created"}
+ self.django_user = User.objects.create_user("test_username", "test@email.com", "test_password")
+ self.nimbus_user = FakeNimbusUser()
+ self.fail = False
+ self.passthrough = lambda x:x
- def _fake_remote_user_creator(self, user_instance, fail=False):
+ def _fake_remote_user_creator(self, user_instance):
"""Creator real request, fake response and failure states.
Get necessary data from `user_instance` to form a correct
request to create a remote Nimbus User.
"""
- if not fail:
- return self.ok_resp
+ if not self.fail:
+ return self.nimbus_user
else:
raise Exception("fake error in '_fake_remote_user_creator'")
def test_create_user(self):
"""
"""
- passthrough = lambda x:x
- created = remote.nimbus_user_create(passthrough, self.user, created=True,
+ created = remote.nimbus_user_create(self.passthrough, self.django_user, created=True,
remote_user_creator=self._fake_remote_user_creator)
- self.assertEquals(created, True)
+ user = User.objects.get(username="test_username")
+ self.assertEquals(user.username, "test_username")
+ self.django_user.delete()
def test_create_user_failed(self):
#test that User rollback is successful.
- pass
+ self.fail = True
+ try:
+ created = remote.nimbus_user_create(self.passthrough, self.django_user, created=True,
+ remote_user_creator=self._fake_remote_user_creator)
+ except:
+ pass
+ user = User.objects.filter(username="test_username")
+ #make sure the User does not exist
+ self.assertEquals(len(user), 0)

0 comments on commit 28f8e12

Please sign in to comment.
Something went wrong with that request. Please try again.