mongodb/mongo-python-driver
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

PYTHON-1488 - Fix auth tests for MongoDB 3.7

Browse Source
This commit is contained in:
Bernie Hackett 2018-02-26 15:32:55 -08:00
parent 0aefc6c19b
commit 4fa72033dd
7 changed files with 101 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
test/__init__.py
View File

@ -39,6 +39,7 @@ from functools import wraps
import pymongo
import pymongo.errors
from bson.son import SON
from bson.py3compat import _unicode
from pymongo import common
from pymongo.common import partition_node
@ -104,6 +105,16 @@ def _connect(host, port, **kwargs):
return client
def _create_user(authdb, user, pwd=None, roles=None, **kwargs):
cmd = SON([('createUser', user)])
# X509 doesn't use a password
if pwd:
cmd['pwd'] = pwd
cmd['roles'] = roles or ['root']
cmd.update(**kwargs)
return authdb.command(cmd)
class client_knobs(object):
def __init__(
self,
@ -208,7 +219,7 @@ class ClientContext(object):
if self.auth_enabled:
# See if db_user already exists.
if not self._check_user_provided():
self.client.admin.add_user(db_user, db_pwd, roles=['root'])
_create_user(self.client.admin, db_user, db_pwd)
self.client = _connect(host,
port,
@ -372,6 +383,14 @@ class ClientContext(object):
return decorate
return make_wrapper(func)
def create_user(self, dbname, user, pwd=None, roles=None, **kwargs):
kwargs['writeConcern'] = {'w': self.w}
return _create_user(self.client[dbname], user, pwd, roles, **kwargs)
def drop_user(self, dbname, user):
self.client[dbname].command(
'dropUser', user, writeConcern={'w': self.w})
def require_connection(self, func):
"""Run a test only if we can connect to MongoDB."""
return self._require(

37
test/test_auth.py
View File

@ -334,11 +334,11 @@ class TestSCRAMSHA1(unittest.TestCase):
{}).get('authenticationMechanisms', ''):
raise SkipTest('SCRAM-SHA-1 mechanism not enabled')
client = client_context.client
client.pymongo_test.add_user(
'user', 'pass',
roles=['userAdmin', 'readWrite'],
writeConcern={'w': client_context.w})
client_context.create_user(
'pymongo_test', 'user', 'pass', roles=['userAdmin', 'readWrite'])
def tearDown(self):
client_context.drop_user('pymongo_test', 'user')
def test_scram_sha1(self):
host, port = client_context.host, client_context.port
@ -365,33 +365,20 @@ class TestSCRAMSHA1(unittest.TestCase):
'pymongo_test', read_preference=ReadPreference.SECONDARY)
db.command('dbstats')
def tearDown(self):
client_context.client.pymongo_test.remove_user('user')
class TestAuthURIOptions(unittest.TestCase):
@client_context.require_auth
def setUp(self):
client_context.client.admin.add_user('admin', 'pass',
roles=['userAdminAnyDatabase',
'dbAdminAnyDatabase',
'readWriteAnyDatabase',
'clusterAdmin'])
client = rs_or_single_client_noauth(username='admin', password='pass')
client.pymongo_test.add_user('user', 'pass',
roles=['userAdmin', 'readWrite'])
if client_context.is_rs:
# Make sure the admin user is replicated after calling add_user
# above. This avoids a race in the replica set tests below.
client.admin.command('getLastError', w=client_context.w)
self.client = client
client_context.create_user('admin', 'admin', 'pass')
client_context.create_user(
'pymongo_test', 'user', 'pass', ['userAdmin', 'readWrite'])
self.client = rs_or_single_client_noauth(
username='admin', password='pass')
def tearDown(self):
self.client.pymongo_test.remove_user('user')
self.client.admin.remove_user('admin')
self.client = None
client_context.drop_user('pymongo_test', 'user')
client_context.drop_user('admin', 'admin')
def test_uri_options(self):
# Test default to admin

5
test/test_bulk.py
View File

@ -340,7 +340,8 @@ class BulkAuthorizationTestBase(BulkTestBase):
def setUp(self):
super(BulkAuthorizationTestBase, self).setUp()
self.db.add_user('readonly', 'pw', roles=['read'])
client_context.create_user(
self.db.name, 'readonly', 'pw', ['read'])
self.db.command(
'createRole', 'noremove',
privileges=[{
@ -349,7 +350,7 @@ class BulkAuthorizationTestBase(BulkTestBase):
}],
roles=[])
self.db.add_user('noremove', 'pw', roles=['noremove'])
client_context.create_user(self.db.name, 'noremove', 'pw', ['noremove'])
def tearDown(self):
self.db.command('dropRole', 'noremove')

18
test/test_client.py
View File

@ -625,12 +625,12 @@ class TestClient(IntegrationTest):
@client_context.require_auth
def test_auth_from_uri(self):
host, port = client_context.host, client_context.port
self.client.admin.add_user("admin", "pass", roles=["root"])
self.addCleanup(self.client.admin.remove_user, 'admin')
client_context.create_user("admin", "admin", "pass")
self.addCleanup(client_context.drop_user, "admin", "admin")
self.addCleanup(remove_all_users, self.client.pymongo_test)
self.client.pymongo_test.add_user(
"user", "pass", roles=['userAdmin', 'readWrite'])
client_context.create_user(
"pymongo_test", "user", "pass", roles=['userAdmin', 'readWrite'])
with self.assertRaises(OperationFailure):
connected(rs_or_single_client(
@ -664,8 +664,8 @@ class TestClient(IntegrationTest):
@client_context.require_auth
def test_username_and_password(self):
self.client.admin.add_user("ad min", "pa/ss", roles=["root"])
self.addCleanup(self.client.admin.remove_user, "ad min")
client_context.create_user("admin", "ad min", "pa/ss")
self.addCleanup(client_context.drop_user, "admin", "ad min")
c = rs_or_single_client(username="ad min", password="pa/ss")
@ -684,8 +684,10 @@ class TestClient(IntegrationTest):
@client_context.require_auth
@ignore_deprecations
def test_multiple_logins(self):
self.client.pymongo_test.add_user('user1', 'pass', roles=['readWrite'])
self.client.pymongo_test.add_user('user2', 'pass', roles=['readWrite'])
client_context.create_user(
'pymongo_test', 'user1', 'pass', roles=['readWrite'])
client_context.create_user(
'pymongo_test', 'user2', 'pass', roles=['readWrite'])
self.addCleanup(remove_all_users, self.client.pymongo_test)
client = rs_or_single_client_noauth(

64
test/test_database.py
View File

@ -503,8 +503,12 @@ class TestDatabase(IntegrationTest):
self.assertRaises(ConfigurationError, auth_db.add_user,
"user", "password", digestPassword=True)
extra = {}
if client_context.version.at_least(3, 7, 2):
extra['mechanisms'] = ['SCRAM-SHA-1']
# Add / authenticate / remove
auth_db.add_user("mike", "password", roles=["read"])
auth_db.add_user("mike", "password", roles=["read"], **extra)
self.addCleanup(remove_all_users, auth_db)
self.assertRaises(TypeError, check_auth, 5, "password")
self.assertRaises(TypeError, check_auth, "mike", 5)
@ -522,21 +526,26 @@ class TestDatabase(IntegrationTest):
# Add / authenticate / change password
self.assertRaises(OperationFailure, check_auth, "Gustave", u"Dor\xe9")
auth_db.add_user("Gustave", u"Dor\xe9", roles=["read"])
auth_db.add_user("Gustave", u"Dor\xe9", roles=["read"], **extra)
check_auth("Gustave", u"Dor\xe9")
# Change password.
auth_db.add_user("Gustave", "password", roles=["read"])
auth_db.add_user("Gustave", "password", roles=["read"], **extra)
self.assertRaises(OperationFailure, check_auth, "Gustave", u"Dor\xe9")
check_auth("Gustave", u"password")
@client_context.require_auth
@ignore_deprecations
def test_make_user_readonly(self):
extra = {}
if client_context.version.at_least(3, 7, 2):
extra['mechanisms'] = ['SCRAM-SHA-1']
# "self.client" is logged in as root.
auth_db = self.client.pymongo_test
# Make a read-write user.
auth_db.add_user('jesse', 'pw')
auth_db.add_user('jesse', 'pw', **extra)
self.addCleanup(remove_all_users, auth_db)
# Check that we're read-write by default.
@ -547,7 +556,7 @@ class TestDatabase(IntegrationTest):
c.pymongo_test.collection.insert_one({})
# Make the user read-only.
auth_db.add_user('jesse', 'pw', read_only=True)
auth_db.add_user('jesse', 'pw', read_only=True, **extra)
c = rs_or_single_client_noauth(username='jesse',
password='pw',
@ -558,40 +567,50 @@ class TestDatabase(IntegrationTest):
{})
@client_context.require_auth
@ignore_deprecations
def test_default_roles(self):
extra = {}
if client_context.version.at_least(3, 7, 2):
extra['mechanisms'] = ['SCRAM-SHA-1']
# "self.client" is logged in as root.
auth_admin = self.client.admin
auth_admin.add_user('test_default_roles', 'pass')
self.addCleanup(auth_admin.remove_user, 'test_default_roles')
auth_admin.add_user('test_default_roles', 'pass', **extra)
self.addCleanup(client_context.drop_user, 'admin', 'test_default_roles')
info = auth_admin.command(
'usersInfo', 'test_default_roles')['users'][0]
self.assertEqual("root", info['roles'][0]['role'])
# Read only "admin" user
auth_admin.add_user('ro-admin', 'pass', read_only=True)
self.addCleanup(auth_admin.remove_user, 'ro-admin')
auth_admin.add_user('ro-admin', 'pass', read_only=True, **extra)
self.addCleanup(client_context.drop_user, 'admin', 'ro-admin')
info = auth_admin.command('usersInfo', 'ro-admin')['users'][0]
self.assertEqual("readAnyDatabase", info['roles'][0]['role'])
# "Non-admin" user
auth_db = self.client.pymongo_test
auth_db.add_user('user', 'pass')
auth_db.add_user('user', 'pass', **extra)
self.addCleanup(remove_all_users, auth_db)
info = auth_db.command('usersInfo', 'user')['users'][0]
self.assertEqual("dbOwner", info['roles'][0]['role'])
# Read only "Non-admin" user
auth_db.add_user('ro-user', 'pass', read_only=True)
auth_db.add_user('ro-user', 'pass', read_only=True, **extra)
info = auth_db.command('usersInfo', 'ro-user')['users'][0]
self.assertEqual("read", info['roles'][0]['role'])
@client_context.require_auth
@ignore_deprecations
def test_new_user_cmds(self):
extra = {}
if client_context.version.at_least(3, 7, 2):
extra['mechanisms'] = ['SCRAM-SHA-1']
# "self.client" is logged in as root.
auth_db = self.client.pymongo_test
auth_db.add_user("amalia", "password", roles=["userAdmin"])
self.addCleanup(auth_db.remove_user, "amalia")
auth_db.add_user("amalia", "password", roles=["userAdmin"], **extra)
self.addCleanup(client_context.drop_user, "pymongo_test", "amalia")
db = rs_or_single_client_noauth(username="amalia",
password="password",
@ -599,7 +618,7 @@ class TestDatabase(IntegrationTest):
# This tests the ability to update user attributes.
db.add_user("amalia", "new_password",
customData={"secret": "koalas"})
customData={"secret": "koalas"}, **extra)
user_info = db.command("usersInfo", "amalia")
self.assertTrue(user_info["users"])
@ -610,6 +629,10 @@ class TestDatabase(IntegrationTest):
@client_context.require_auth
@ignore_deprecations
def test_authenticate_multiple(self):
extra = {}
if client_context.version.at_least(3, 7, 2):
extra['mechanisms'] = ['SCRAM-SHA-1']
# "self.client" is logged in as root.
self.client.drop_database("pymongo_test")
self.client.drop_database("pymongo_test1")
@ -624,12 +647,15 @@ class TestDatabase(IntegrationTest):
self.assertRaises(OperationFailure, users_db.test.find_one)
admin_db_auth.add_user('ro-admin', 'pass',
roles=["userAdmin", "readAnyDatabase"])
admin_db_auth.add_user(
'ro-admin',
'pass',
roles=["userAdmin", "readAnyDatabase"],
**extra)
self.addCleanup(admin_db_auth.remove_user, 'ro-admin')
users_db_auth.add_user('user', 'pass',
roles=["userAdmin", "readWrite"])
self.addCleanup(client_context.drop_user, 'admin', 'ro-admin')
users_db_auth.add_user(
'user', 'pass', roles=["userAdmin", "readWrite"], **extra)
self.addCleanup(remove_all_users, users_db_auth)
# Regular user should be able to query its own db, but

16
test/test_session.py
View File

@ -227,6 +227,7 @@ class TestSession(IntegrationTest):
self._test_ops(client, *ops)
@client_context.require_auth
@ignore_deprecations
def test_user_admin(self):
listener = SessionTestListener()
client = rs_or_single_client(event_listeners=[listener])
@ -234,11 +235,15 @@ class TestSession(IntegrationTest):
self.addCleanup(client.drop_database, 'pymongo_test')
db = client.pymongo_test
extra = {'roles': ['read']}
if client_context.version.at_least(3, 7, 2):
extra['mechanisms'] = ['SCRAM-SHA-1']
self._test_ops(
client,
(db.add_user, ['session-test', 'pass'], {'roles': ['read']}),
(db.add_user, ['session-test', 'pass'], extra),
# Do it again to test updateUser command.
(db.add_user, ['session-test', 'pass'], {'roles': ['read']}),
(db.add_user, ['session-test', 'pass'], extra),
(db.remove_user, ['session-test'], {}))
def test_collection(self):
@ -979,10 +984,9 @@ class TestSessionsMultiAuth(IntegrationTest):
def setUp(self):
super(TestSessionsMultiAuth, self).setUp()
client = rs_or_single_client() # Logged in as root.
db = client.pymongo_test
db.add_user('second-user', 'pass', roles=['readWrite'])
self.addCleanup(db.remove_user, 'second-user')
client_context.create_user(
'pymongo_test', 'second-user', 'pass', roles=['readWrite'])
self.addCleanup(client_context.drop_user, 'pymongo_test','second-user')
@ignore_deprecations
def test_session_authenticate_multiple(self):

2
test/test_ssl.py
View File

@ -516,7 +516,7 @@ class TestSSL(IntegrationTest):
ssl_client.admin.authenticate(db_user, db_pwd)
# Give x509 user all necessary privileges.
ssl_client['$external'].add_user(MONGODB_X509_USERNAME, roles=[
client_context.create_user('$external', MONGODB_X509_USERNAME, roles=[
{'role': 'readWriteAnyDatabase', 'db': 'admin'},
{'role': 'userAdminAnyDatabase', 'db': 'admin'}])