Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Browse files

Fix a bug in and pep8ify test_auth.py

  • Loading branch information...
commit 346fa24c1739b5685bb5fc50cec6f4d2178b076f 1 parent 6a5a8b9
@behackett behackett authored
Showing with 31 additions and 18 deletions.
  1. +31 −18 test/test_auth.py
View
49 test/test_auth.py
@@ -37,6 +37,23 @@
GSSAPI_PORT = int(os.environ.get('GSSAPI_PORT', '27017'))
PRINCIPAL = os.environ.get('PRINCIPAL')
+
+class AutoAuthenticateThread(threading.Thread):
+ """Used in testing threaded authentication.
+ """
+
+ def __init__(self, database):
+ super(AutoAuthenticateThread, self).__init__()
+ self.database = database
+ self.success = True
+
+ def run(self):
+ try:
+ self.database.command('dbstats')
+ except OperationFailure:
+ self.success = False
+
+
class TestGSSAPI(unittest.TestCase):
def setUp(self):
@@ -80,25 +97,17 @@ def test_gssapi_threaded(self):
self.assertTrue(client.test.authenticate(PRINCIPAL,
mechanism='GSSAPI'))
- result = True
- def try_command():
- try:
- client.foo.command('dbstats')
- except OperationFailure:
- result = False
-
threads = []
- for _ in xrange(2):
- threads.append(threading.Thread(target=try_command))
+ for _ in xrange(4):
+ threads.append(AutoAuthenticateThread(client.foo))
for thread in threads:
thread.start()
for thread in threads:
thread.join()
- self.assertTrue(result)
+ self.assertTrue(thread.success)
set_name = client.admin.command('ismaster').get('setName')
if set_name:
- result = True
preference = ReadPreference.SECONDARY
client = MongoReplicaSetClient(GSSAPI_HOST,
replicaSet=set_name,
@@ -108,13 +117,13 @@ def try_command():
self.assertTrue(client.foo.command('dbstats'))
threads = []
- for _ in xrange(2):
- threads.append(threading.Thread(target=try_command))
+ for _ in xrange(4):
+ threads.append(AutoAuthenticateThread(client.foo))
for thread in threads:
thread.start()
for thread in threads:
thread.join()
- self.assertTrue(result)
+ self.assertTrue(thread.success)
class TestAuthURIOptions(unittest.TestCase):
@@ -228,21 +237,25 @@ def test_delegated_auth(self):
userSource='pymongo_test',
roles=['read'])
self.client.admin.logout()
- self.assertRaises(OperationFailure, self.client.pymongo_test2.foo.find_one)
+ self.assertRaises(OperationFailure,
+ self.client.pymongo_test2.foo.find_one)
# Auth must occur on the db where the user is defined.
- self.assertFalse(self.client.pymongo_test2.authenticate('user', 'pass'))
+ self.assertFalse(self.client.pymongo_test2.authenticate('user',
+ 'pass'))
# Auth directly
self.assertTrue(self.client.pymongo_test.authenticate('user', 'pass'))
self.assertTrue(self.client.pymongo_test2.foo.find_one())
self.client.pymongo_test.logout()
- self.assertRaises(OperationFailure, self.client.pymongo_test2.foo.find_one)
+ self.assertRaises(OperationFailure,
+ self.client.pymongo_test2.foo.find_one)
# Auth using source
self.assertTrue(self.client.pymongo_test2.authenticate(
'user', 'pass', source='pymongo_test'))
self.assertTrue(self.client.pymongo_test2.foo.find_one())
# Must logout from the db authenticate was called on.
self.client.pymongo_test2.logout()
- self.assertRaises(OperationFailure, self.client.pymongo_test2.foo.find_one)
+ self.assertRaises(OperationFailure,
+ self.client.pymongo_test2.foo.find_one)
if __name__ == "__main__":
Please sign in to comment.
Something went wrong with that request. Please try again.