Permalink
Browse files

Added unit tests, TODO: Finish the async test suite.

--HG--
rename : SampleMojoProject/Apps/SampleApp/testdb.py => SampleMojoProject/Apps/SampleApp/dbtest.py
rename : SampleMojoProject/Apps/SampleApp/testForm.py => SampleMojoProject/Apps/SampleApp/formtest.py
  • Loading branch information...
1 parent b41c8a3 commit f9759fba7647c9b17afb7ef20f6fdc3a441dcbf7 @LBiNationalTrust LBiNationalTrust committed Sep 20, 2012
View
@@ -38,7 +38,7 @@ def set_password(userObj, new_password):
"""
new_pw = bcrypt.hashpw(new_password, bcrypt.gensalt())
- userObj.password = new_pw
+ userObj.password = unicode(new_pw)
return userObj
@@ -81,11 +81,12 @@ def is_member_of(userObj, groupObj):
"""
Will determine is a user is a member of a group. Returns ``Tru`` or ``False``
"""
+
if userObj.groups:
- if groupObj in userObj.groups:
- return True
- else:
- return False
+ for group in userObj.groups:
+ if group.group_name == groupObj.group_name:
+ return True
+ return False
else:
return False
@@ -108,8 +109,13 @@ def remove_from_group(userObj, groupObj):
Removes a user from the group, returns ``True`` if successful, ``False`` if the user is not a member of the group
"""
if is_member_of(userObj, groupObj):
- userObj.groups.remove(groupObj)
- return True
+ new_list = []
+ for group in userObj.groups:
+ if group.group_name != groupObj.group_name:
+ new_list.append(group)
+
+ userObj.groups = new_list
+ return userObj
else:
- return False
+ return userObj
View
@@ -38,7 +38,7 @@ class User(Model):
username = StringField(allow_empty=False)
password = StringField(allow_empty=False)
email = StringField()
- groups = ListField()
+ groups = ListField(of=Group)
active = BooleanField(default = True)
profile = EmbeddedModelField(to=Profile)
is_superuser = BooleanField(default = False)
@@ -0,0 +1 @@
+__author__ = 'buhrm'
@@ -0,0 +1,95 @@
+import unittest
+import datetime
+
+from Mojo.Auth.models import User, Group
+from Mojo.Auth.AuthManager import *
+
+testGroup = Group({
+ 'group_name' : u'testGroup'
+})
+
+new_group = Group({
+ 'group_name':u'second_group'
+})
+
+testUser = User({
+ 'username':u'testuser',
+ 'password': u'$2a$12$mwFMSikk.7YgXbzCFYIjNeTthsmDMBunwdkYQiIkk38c2gji/c7N.',
+ 'email':u'test@test.com',
+ 'groups':[testGroup],
+ 'active':True,
+ 'profile':None,
+ 'is_superuser':False,
+ 'last_login':datetime.datetime.now(),
+ 'date_joined':datetime.datetime.now(),
+ })
+
+testSuperUser = User({
+ 'username':u'testuser',
+ 'password': u'$2a$12$mwFMSikk.7YgXbzCFYIjNeTthsmDMBunwdkYQiIkk38c2gji/c7N.',
+ 'email':u'test@test.com',
+ 'groups':None,
+ 'active':True,
+ 'profile':None,
+ 'is_superuser':True,
+ 'last_login':datetime.datetime.now(),
+ 'date_joined':datetime.datetime.now(),
+ })
+
+class AuthManagerTest(unittest.TestCase):
+
+ def test_authenticate_positive(self):
+ self.assertEqual(authenticate(testUser, u'realpasswd'), True)
+
+ def test_authenticate_negative(self):
+ self.assertEqual(authenticate(testUser, u'fakepasswd'), False)
+
+ def test_set_password(self):
+ real_password = u'realpasswd_changed'
+ temp_UserObj = set_password(testUser, real_password)
+ self.assertEqual(authenticate(temp_UserObj, u'realpasswd_changed'), True)
+
+ def test_make_random_password(self):
+ pw = make_random_password()
+ pwlen = len(pw)
+ self.assertEqual(pwlen, 8)
+
+ def test_login(self):
+ tempUserObj = login(testUser)
+ now = datetime.datetime.now()
+ ten_seconds_in_past = now - datetime.timedelta(seconds =10)
+
+ isInRange = False
+ if tempUserObj.last_login < now:
+ if tempUserObj.last_login > ten_seconds_in_past:
+ isInRange = True
+
+ self.assertEqual(isInRange, True)
+
+
+ def test_is_member_of(self):
+
+ checkMember = is_member_of(testUser,testGroup)
+
+ self.assertEqual(checkMember, True)
+
+ def test_add_to_group(self):
+ thisUser = add_to_group(testUser, new_group)
+
+ inGroup = False
+ if is_member_of(thisUser, new_group):
+ inGroup = True
+
+ self.assertEqual(inGroup, True)
+
+ def test_remove_from_group(self):
+ returnedUser = remove_from_group(testUser, new_group)
+
+ inGroup = True
+ if not is_member_of(returnedUser, new_group):
+ inGroup = False
+
+ self.assertEqual(inGroup, False)
+
+if __name__ == '__main__':
+ unittest.main()
@@ -85,7 +85,6 @@ def save(self, document, callback):
def delete(self, documents, callback):
return_values = []
for document in documents:
- print 'Deleting: ', document
ret_val = yield gen.Task(self.session._db[self.collection_name].remove, document._id)
return_values.append(ret_val)
@@ -0,0 +1 @@
+__author__ = 'buhrm'
@@ -0,0 +1,196 @@
+import unittest
+from datetime import datetime, timedelta, tzinfo
+from tornado.ioloop import IOLoop
+from tornado import testing
+from bson import ObjectId
+
+from Mojo.Backends.AsyncmongoBackend.asyncmongo_backend import Session, Collection
+from Mojo.ObjectMapper.ModelPrototype import Model
+from Mojo.ObjectMapper.Fields import *
+
+DB = Session(host='127.0.0.1', port=27017, db_name='asyncmongoBackendTest')
+
+class AuthManagerTest(testing.AsyncTestCase):
+ def get_new_ioloop(self):
+ return IOLoop.instance()
+
+ def test_find_insert_find_one_delete(self):
+ class testModel(Model):
+ _id = ObjectIDField()
+ title = StringField()
+ date_published = DateTimeField()
+ published = BooleanField(default=True)
+ integer_field = IntegerField()
+ float_field = FloatField()
+ list_field = ListField()
+
+ now = datetime(2012, 1, 1, 1,1,1)
+
+ insert_data = {
+ 'title':u'test title',
+ 'date_published': now,
+ 'published': True,
+ 'integer_field': 42,
+ 'float_field': 3.14,
+ 'list_field': [u'life',u'the universe', u'everything']
+ }
+
+ testInstantiatedModel = testModel(insert_data)
+ collectionObj = Collection(DB, testModel)
+
+ #Clear the DB
+ collectionObj.find({}, callback=self.stop)
+ allObjects = self.wait()
+
+ collectionObj.delete(allObjects, callback=self.stop)
+ ignore = self.wait()
+
+ collectionObj.insert([testInstantiatedModel], callback=self.stop)
+
+ ret_vals = self.wait()
+
+ collectionObj.find_one({'title':'test title'}, callback=self.stop)
+
+ thisObj = self.wait()
+
+ as_dict = {}
+ if thisObj:
+ as_dict = dict(thisObj)
+ itemID = as_dict['_id']
+ del as_dict['_id']
+
+ collectionObj.delete([thisObj], callback=self.stop)
+
+ ignore = self.wait()
+
+ self.assertEqual(as_dict, insert_data)
+
+ def test_save(self):
+ class testModel(Model):
+ _id = ObjectIDField()
+ title = StringField()
+ date_published = DateTimeField()
+ published = BooleanField()
+ integer_field = IntegerField()
+ float_field = FloatField()
+ list_field = ListField()
+
+ now = datetime(2012, 1, 1, 1,1,1)
+
+ insert_data = {
+ 'title':u'test title',
+ 'date_published': now,
+ 'published': True,
+ 'integer_field': 42,
+ 'float_field': 3.14,
+ 'list_field': [u'life',u'the universe', u'everything']
+ }
+
+ changed_data = {
+ 'title':u'consider phlebas',
+ 'date_published': now,
+ 'published': False,
+ 'integer_field': 43,
+ 'float_field': 3.15,
+ 'list_field': None
+ }
+
+ testInstantiatedModel = testModel(insert_data)
+ collectionObj = Collection(DB, testModel)
+
+ #Clear the DB
+ collectionObj.find({}, callback=self.stop)
+ allObjects = self.wait()
+
+ collectionObj.delete(allObjects, callback=self.stop)
+ ignore = self.wait()
+
+ collectionObj.insert([testInstantiatedModel], callback=self.stop)
+
+ ret_vals = self.wait()
+
+ collectionObj.find_one({'title':u'test title'}, callback=self.stop)
+
+ thisObj = self.wait()
+
+ as_dict = {}
+ if thisObj:
+ thisObj.title = u'consider phlebas'
+ thisObj.date_published = now
+ thisObj.published = False
+ thisObj.integer_field = 43
+ thisObj.float_field = 3.15
+ thisObj.list_field = None
+
+
+ collectionObj.save(thisObj, callback=self.stop)
+ save_return = self.wait()
+
+ collectionObj.find_one({'title':u'consider phlebas'}, callback=self.stop)
+
+ changedObj = self.wait()
+
+ if changedObj:
+ as_dict = dict(changedObj)
+ del as_dict['_id']
+
+ collectionObj.delete([changedObj], callback=self.stop)
+
+ ignore = self.wait()
+
+ self.assertEqual(as_dict, changed_data)
+
+ def test_delete(self):
+ class testModel(Model):
+ _id = ObjectIDField()
+ title = StringField()
+ date_published = DateTimeField()
+ published = BooleanField()
+ integer_field = IntegerField()
+ float_field = FloatField()
+ list_field = ListField()
+
+ now = datetime(2012, 1, 1, 1,1,1)
+
+ insert_data = {
+ 'title':u'test title',
+ 'date_published': now,
+ 'published': True,
+ 'integer_field': 42,
+ 'float_field': 3.14,
+ 'list_field': [u'life',u'the universe', u'everything']
+ }
+
+
+ testInstantiatedModel = testModel(insert_data)
+ collectionObj = Collection(DB, testModel)
+
+ #Clear the DB
+ collectionObj.find({}, callback=self.stop)
+ allObjects = self.wait()
+
+ collectionObj.delete(allObjects, callback=self.stop)
+ ignore = self.wait()
+
+ collectionObj.insert([testInstantiatedModel], callback=self.stop)
+
+ ret_vals = self.wait()
+
+ collectionObj.find_one({'title':u'test title'}, callback=self.stop)
+
+ thisObj = self.wait()
+
+ emptyObj = {'value':True}
+
+ if thisObj:
+ collectionObj.delete([thisObj], callback=self.stop)
+ ret = self.wait()
+ collectionObj.find_one({'title':u'test title'}, callback=self.stop)
+ emptyObj = self.wait()
+
+
+ self.assertEqual(emptyObj, None)
+
+
+if __name__ == '__main__':
+ testing.main()
@@ -1,5 +1,6 @@
from Mojo.Backends.base_interface import CollectionModelInterface, SessionInterface
import pymongo
+import logging
class Session(SessionInterface):
def _setup_connection(self):
@@ -25,22 +26,27 @@ def insert(self, documents, *args, **kwargs):
else:
insert_list = []
for doc in documents:
- del doc.__data__['_id']
+ del doc['_id']
insert_list.append(doc.get_value())
ids = self.session._db[self.collection_name].insert(insert_list)
index = 0
for doc in documents:
- doc.__data__['_id'] = ids[index]
+ doc['_id'] = ids[index]
index += 1
ret_vals.append(doc)
return ret_vals
def save(self, document, *args, **kwargs):
- returnID = self.session._db[self.collection_name].save(document.get_value(), *args, **kwargs)
+ temp_doc = dict(document)
+
+ if not document._id:
+ del temp_doc['_id']
+
+ returnID = self.session._db[self.collection_name].save(temp_doc, *args, **kwargs)
document._id = returnID
return document
@@ -0,0 +1 @@
+__author__ = 'buhrm'
Oops, something went wrong.

0 comments on commit f9759fb

Please sign in to comment.