本文整理汇总了Python中test.utils.rs_or_single_client_noauth函数的典型用法代码示例。如果您正苦于以下问题:Python rs_or_single_client_noauth函数的具体用法?Python rs_or_single_client_noauth怎么用?Python rs_or_single_client_noauth使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了rs_or_single_client_noauth函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_make_user_readonly
def test_make_user_readonly(self):
# "self.client" is logged in as root.
auth_db = self.client.pymongo_test
# Make a read-write user.
auth_db.add_user('jesse', 'pw')
self.addCleanup(remove_all_users, auth_db)
# Check that we're read-write by default.
c = rs_or_single_client_noauth(username='jesse',
password='pw',
authSource='pymongo_test')
c.pymongo_test.collection.insert_one({})
# Make the user read-only.
auth_db.add_user('jesse', 'pw', read_only=True)
c = rs_or_single_client_noauth(username='jesse',
password='pw',
authSource='pymongo_test')
self.assertRaises(OperationFailure,
c.pymongo_test.collection.insert_one,
{})
开发者ID:ShaneHarvey,项目名称:mongo-python-driver,代码行数:25,代码来源:test_database.py
示例2: test_db_authenticate_threaded
def test_db_authenticate_threaded(self):
db = client_context.client.db
coll = db.test
coll.drop()
coll.insert_one({'_id': 1})
client_context.create_user(
'db',
'user',
'pass',
roles=['dbOwner'])
self.addCleanup(db.command, 'dropUser', 'user')
db = rs_or_single_client_noauth().db
db.authenticate('user', 'pass')
# No error.
db.authenticate('user', 'pass')
db = rs_or_single_client_noauth().db
threads = []
for _ in range(4):
threads.append(DBAuthenticateThread(db, 'user', 'pass'))
for thread in threads:
thread.start()
for thread in threads:
thread.join()
self.assertTrue(thread.success)
开发者ID:ShaneHarvey,项目名称:mongo-python-driver,代码行数:28,代码来源:test_auth.py
示例3: test_auth_from_uri
def test_auth_from_uri(self):
self.client.admin.add_user("admin", "pass", roles=["root"])
self.addCleanup(self.client.admin.remove_user, "admin")
self.addCleanup(remove_all_users, self.client.pymongo_test)
self.client.pymongo_test.add_user("user", "pass", roles=["userAdmin", "readWrite"])
with self.assertRaises(OperationFailure):
connected(rs_or_single_client("mongodb://a:[email protected]%s:%d" % (host, port)))
# No error.
connected(rs_or_single_client_noauth("mongodb://admin:[email protected]%s:%d" % (host, port)))
# Wrong database.
uri = "mongodb://admin:[email protected]%s:%d/pymongo_test" % (host, port)
with self.assertRaises(OperationFailure):
connected(rs_or_single_client(uri))
# No error.
connected(rs_or_single_client_noauth("mongodb://user:[email protected]%s:%d/pymongo_test" % (host, port)))
# Auth with lazy connection.
rs_or_single_client(
"mongodb://user:[email protected]%s:%d/pymongo_test" % (host, port), connect=False
).pymongo_test.test.find_one()
# Wrong password.
bad_client = rs_or_single_client("mongodb://user:[email protected]%s:%d/pymongo_test" % (host, port), connect=False)
self.assertRaises(OperationFailure, bad_client.pymongo_test.test.find_one)
开发者ID:rychipman,项目名称:mongo-python-driver,代码行数:30,代码来源:test_client.py
示例4: test_uri_options
def test_uri_options(self):
# Test default to admin
host, port = client_context.host, client_context.port
client = rs_or_single_client_noauth(
'mongodb://admin:[email protected]%s:%d' % (host, port))
self.assertTrue(client.admin.command('dbstats'))
if client_context.is_rs:
uri = ('mongodb://admin:[email protected]%s:%d/?replicaSet=%s' % (
host, port, client_context.replica_set_name))
client = single_client_noauth(uri)
self.assertTrue(client.admin.command('dbstats'))
db = client.get_database(
'admin', read_preference=ReadPreference.SECONDARY)
self.assertTrue(db.command('dbstats'))
# Test explicit database
uri = 'mongodb://user:[email protected]%s:%d/pymongo_test' % (host, port)
client = rs_or_single_client_noauth(uri)
self.assertRaises(OperationFailure, client.admin.command, 'dbstats')
self.assertTrue(client.pymongo_test.command('dbstats'))
if client_context.is_rs:
uri = ('mongodb://user:[email protected]%s:%d/pymongo_test?replicaSet=%s' % (
host, port, client_context.replica_set_name))
client = single_client_noauth(uri)
self.assertRaises(OperationFailure,
client.admin.command, 'dbstats')
self.assertTrue(client.pymongo_test.command('dbstats'))
db = client.get_database(
'pymongo_test', read_preference=ReadPreference.SECONDARY)
self.assertTrue(db.command('dbstats'))
# Test authSource
uri = ('mongodb://user:[email protected]%s:%d'
'/pymongo_test2?authSource=pymongo_test' % (host, port))
client = rs_or_single_client_noauth(uri)
self.assertRaises(OperationFailure,
client.pymongo_test2.command, 'dbstats')
self.assertTrue(client.pymongo_test.command('dbstats'))
if client_context.is_rs:
uri = ('mongodb://user:[email protected]%s:%d/pymongo_test2?replicaSet='
'%s;authSource=pymongo_test' % (
host, port, client_context.replica_set_name))
client = single_client_noauth(uri)
self.assertRaises(OperationFailure,
client.pymongo_test2.command, 'dbstats')
self.assertTrue(client.pymongo_test.command('dbstats'))
db = client.get_database(
'pymongo_test', read_preference=ReadPreference.SECONDARY)
self.assertTrue(db.command('dbstats'))
开发者ID:ShaneHarvey,项目名称:mongo-python-driver,代码行数:52,代码来源:test_auth.py
示例5: test_delegated_auth
def test_delegated_auth(self):
self.client.pymongo_test2.foo.drop()
self.client.pymongo_test2.foo.insert_one({})
# User definition with no roles in pymongo_test.
self.client.pymongo_test.add_user('user', 'pass', roles=[])
# Delegate auth to pymongo_test.
self.client.pymongo_test2.add_user('user',
userSource='pymongo_test',
roles=['read'])
auth_c = rs_or_single_client_noauth()
self.assertRaises(OperationFailure,
auth_c.pymongo_test2.foo.find_one)
# Auth must occur on the db where the user is defined.
self.assertRaises(OperationFailure,
auth_c.pymongo_test2.authenticate,
'user', 'pass')
# Auth directly
self.assertTrue(auth_c.pymongo_test.authenticate('user', 'pass'))
self.assertTrue(auth_c.pymongo_test2.foo.find_one())
auth_c.pymongo_test.logout()
self.assertRaises(OperationFailure,
auth_c.pymongo_test2.foo.find_one)
# Auth using source
self.assertTrue(auth_c.pymongo_test2.authenticate(
'user', 'pass', source='pymongo_test'))
self.assertTrue(auth_c.pymongo_test2.foo.find_one())
# Must logout from the db authenticate was called on.
auth_c.pymongo_test2.logout()
self.assertRaises(OperationFailure,
auth_c.pymongo_test2.foo.find_one)
开发者ID:nly,项目名称:mongo-python-driver,代码行数:30,代码来源:test_auth.py
示例6: test_multiple_logins
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'])
self.addCleanup(remove_all_users, self.client.pymongo_test)
client = rs_or_single_client_noauth(
"mongodb://user1:[email protected]%s:%d/pymongo_test" % (host, port))
client.pymongo_test.test.find_one()
with self.assertRaises(OperationFailure):
# Can't log in to the same database with multiple users.
client.pymongo_test.authenticate('user2', 'pass')
client.pymongo_test.test.find_one()
client.pymongo_test.logout()
with self.assertRaises(OperationFailure):
client.pymongo_test.test.find_one()
client.pymongo_test.authenticate('user2', 'pass')
client.pymongo_test.test.find_one()
with self.assertRaises(OperationFailure):
client.pymongo_test.authenticate('user1', 'pass')
client.pymongo_test.test.find_one()
开发者ID:gregbanks,项目名称:mongo-python-driver,代码行数:25,代码来源:test_client.py
示例7: check_auth
def check_auth(username, password):
c = rs_or_single_client_noauth(
username=username,
password=password,
authSource="pymongo_test")
c.pymongo_test.collection.find_one()
开发者ID:ShaneHarvey,项目名称:mongo-python-driver,代码行数:7,代码来源:test_database.py
示例8: test_readonly
def test_readonly(self):
# We test that an authorization failure aborts the batch and is raised
# as OperationFailure.
cli = rs_or_single_client_noauth(username='readonly', password='pw',
authSource='pymongo_test')
coll = cli.pymongo_test.test
coll.find_one()
self.assertRaises(OperationFailure, coll.bulk_write,
[InsertOne({'x': 1})])
开发者ID:ShaneHarvey,项目名称:mongo-python-driver,代码行数:9,代码来源:test_bulk.py
示例9: test_readonly
def test_readonly(self):
# We test that an authorization failure aborts the batch and is raised
# as OperationFailure.
cli = rs_or_single_client_noauth()
db = cli.pymongo_test
coll = db.test
db.authenticate('readonly', 'pw')
bulk = coll.initialize_ordered_bulk_op()
bulk.insert({'x': 1})
self.assertRaises(OperationFailure, bulk.execute)
开发者ID:nly,项目名称:mongo-python-driver,代码行数:10,代码来源:test_bulk.py
示例10: test_gridfs_readonly
def test_gridfs_readonly(self):
# "self.client" is logged in as root. Make a read-only user.
auth_db = self.client.test_gridfs_readonly
auth_db.add_user('readonly', 'pw', readOnly=True)
self.addCleanup(remove_all_users, auth_db)
db = rs_or_single_client_noauth().test_gridfs_readonly
db.authenticate('readonly', 'pw')
fs = gridfs.GridFS(db)
file = fs.new_file()
file._ensure_index()
fs.list()
开发者ID:uranusjr,项目名称:mongo-python-driver,代码行数:13,代码来源:test_gridfs.py
示例11: test_scram_sha1
def test_scram_sha1(self):
host, port = client_context.host, client_context.port
client = rs_or_single_client_noauth()
self.assertTrue(client.pymongo_test.authenticate(
'user', 'pass', mechanism='SCRAM-SHA-1'))
client.pymongo_test.command('dbstats')
client = rs_or_single_client_noauth(
'mongodb://user:[email protected]%s:%d/pymongo_test?authMechanism=SCRAM-SHA-1'
% (host, port))
client.pymongo_test.command('dbstats')
if client_context.is_rs:
uri = ('mongodb://user:pass'
'@%s:%d/pymongo_test?authMechanism=SCRAM-SHA-1'
'&replicaSet=%s' % (host, port,
client_context.replica_set_name))
client = single_client_noauth(uri)
client.pymongo_test.command('dbstats')
db = client.get_database(
'pymongo_test', read_preference=ReadPreference.SECONDARY)
db.command('dbstats')
开发者ID:nly,项目名称:mongo-python-driver,代码行数:22,代码来源:test_auth.py
示例12: test_authenticate_multiple
def test_authenticate_multiple(self):
# "self.client" is logged in as root.
self.client.drop_database("pymongo_test")
self.client.drop_database("pymongo_test1")
admin_db_auth = self.client.admin
users_db_auth = self.client.pymongo_test
# Non-root client.
client = rs_or_single_client_noauth()
admin_db = client.admin
users_db = client.pymongo_test
other_db = client.pymongo_test1
self.assertRaises(OperationFailure, users_db.test.find_one)
if client_context.version.at_least(2, 5, 3, -1):
admin_db_auth.add_user('ro-admin', 'pass',
roles=["userAdmin", "readAnyDatabase"])
else:
admin_db_auth.add_user('ro-admin', 'pass', read_only=True)
self.addCleanup(admin_db_auth.remove_user, 'ro-admin')
users_db_auth.add_user('user', 'pass',
roles=["userAdmin", "readWrite"])
self.addCleanup(remove_all_users, users_db_auth)
# Regular user should be able to query its own db, but
# no other.
users_db.authenticate('user', 'pass')
self.assertEqual(0, users_db.test.count())
self.assertRaises(OperationFailure, other_db.test.find_one)
# Admin read-only user should be able to query any db,
# but not write.
admin_db.authenticate('ro-admin', 'pass')
self.assertEqual(None, other_db.test.find_one())
self.assertRaises(OperationFailure,
other_db.test.insert_one, {})
# Close all sockets.
client.close()
# We should still be able to write to the regular user's db.
self.assertTrue(users_db.test.delete_many({}))
# And read from other dbs...
self.assertEqual(0, other_db.test.count())
# But still not write to other dbs.
self.assertRaises(OperationFailure,
other_db.test.insert_one, {})
开发者ID:jonnyhsu,项目名称:mongo-python-driver,代码行数:51,代码来源:test_database.py
示例13: test_no_remove
def test_no_remove(self):
# We test that an authorization failure aborts the batch and is raised
# as OperationFailure.
cli = rs_or_single_client_noauth()
db = cli.pymongo_test
coll = db.test
db.authenticate('noremove', 'pw')
bulk = coll.initialize_ordered_bulk_op()
bulk.insert({'x': 1})
bulk.find({'x': 2}).upsert().replace_one({'x': 2})
bulk.find({}).remove() # Prohibited.
bulk.insert({'x': 3}) # Never attempted.
self.assertRaises(OperationFailure, bulk.execute)
self.assertEqual(set([1, 2]), set(self.coll.distinct('x')))
开发者ID:nly,项目名称:mongo-python-driver,代码行数:14,代码来源:test_bulk.py
示例14: test_no_remove
def test_no_remove(self):
# We test that an authorization failure aborts the batch and is raised
# as OperationFailure.
cli = rs_or_single_client_noauth(username='noremove', password='pw',
authSource='pymongo_test')
coll = cli.pymongo_test.test
coll.find_one()
requests = [
InsertOne({'x': 1}),
ReplaceOne({'x': 2}, {'x': 2}, upsert=True),
DeleteMany({}), # Prohibited.
InsertOne({'x': 3}), # Never attempted.
]
self.assertRaises(OperationFailure, coll.bulk_write, requests)
self.assertEqual(set([1, 2]), set(self.coll.distinct('x')))
开发者ID:ShaneHarvey,项目名称:mongo-python-driver,代码行数:15,代码来源:test_bulk.py
示例15: setUp
def setUp(self):
client = rs_or_single_client_noauth()
client_context.client.admin.add_user('admin', 'pass',
roles=['userAdminAnyDatabase',
'dbAdminAnyDatabase',
'readWriteAnyDatabase',
'clusterAdmin'])
client.admin.authenticate('admin', '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
开发者ID:nly,项目名称:mongo-python-driver,代码行数:16,代码来源:test_auth.py
示例16: test_ipv6
def test_ipv6(self):
if client_context.auth_enabled:
auth_str = "%s:%[email protected]" % (db_user, db_pwd)
else:
auth_str = ""
uri = "mongodb://%s[::1]:%d" % (auth_str, port)
if client_context.is_rs:
uri += '/?replicaSet=' + client_context.replica_set_name
client = rs_or_single_client_noauth(uri)
client.pymongo_test.test.insert_one({"dummy": u("object")})
client.pymongo_test_bernie.test.insert_one({"dummy": u("object")})
dbs = client.database_names()
self.assertTrue("pymongo_test" in dbs)
self.assertTrue("pymongo_test_bernie" in dbs)
开发者ID:gregbanks,项目名称:mongo-python-driver,代码行数:17,代码来源:test_client.py
示例17: test_new_user_cmds
def test_new_user_cmds(self):
# "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")
db = rs_or_single_client_noauth().pymongo_test
db.authenticate("amalia", "password")
# This tests the ability to update user attributes.
db.add_user("amalia", "new_password",
customData={"secret": "koalas"})
user_info = db.command("usersInfo", "amalia")
self.assertTrue(user_info["users"])
amalia_user = user_info["users"][0]
self.assertEqual(amalia_user["user"], "amalia")
self.assertEqual(amalia_user["customData"], {"secret": "koalas"})
开发者ID:jonnyhsu,项目名称:mongo-python-driver,代码行数:18,代码来源:test_database.py
示例18: test_auto_auth_login
def test_auto_auth_login(self):
client = rs_or_single_client_noauth()
self.assertRaises(OperationFailure, client.auth_test.test.find_one)
# Admin auth
client.admin.authenticate(db_user, db_pwd)
nthreads = 10
threads = []
for _ in range(nthreads):
t = AutoAuthenticateThreads(client.auth_test.test, 10)
t.start()
threads.append(t)
joinall(threads)
for t in threads:
self.assertTrue(t.success)
开发者ID:BiosPsucheZoe,项目名称:mongo-python-driver,代码行数:18,代码来源:test_threads.py
示例19: test_make_user_readonly
def test_make_user_readonly(self):
# "self.client" is logged in as root.
auth_db = self.client.pymongo_test
db = rs_or_single_client_noauth().pymongo_test
# Make a read-write user.
auth_db.add_user('jesse', 'pw')
self.addCleanup(remove_all_users, auth_db)
# Check that we're read-write by default.
db.authenticate('jesse', 'pw')
db.collection.insert_one({})
db.logout()
# Make the user read-only.
auth_db.add_user('jesse', 'pw', read_only=True)
db.authenticate('jesse', 'pw')
self.assertRaises(OperationFailure, db.collection.insert_one, {})
开发者ID:jonnyhsu,项目名称:mongo-python-driver,代码行数:19,代码来源:test_database.py
示例20: test_auto_auth_login
def test_auto_auth_login(self):
# Create the database upfront to workaround SERVER-39167.
self.client.auth_test.test.insert_one({})
self.addCleanup(self.client.drop_database, "auth_test")
client = rs_or_single_client_noauth()
self.assertRaises(OperationFailure, client.auth_test.test.find_one)
# Admin auth
client.admin.authenticate(db_user, db_pwd)
nthreads = 10
threads = []
for _ in range(nthreads):
t = AutoAuthenticateThreads(client.auth_test.test, 10)
t.start()
threads.append(t)
joinall(threads)
for t in threads:
self.assertTrue(t.success)
开发者ID:ShaneHarvey,项目名称:mongo-python-driver,代码行数:21,代码来源:test_threads.py
注:本文中的test.utils.rs_or_single_client_noauth函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论