本文整理汇总了Python中test.utils.remove_all_users函数的典型用法代码示例。如果您正苦于以下问题:Python remove_all_users函数的具体用法?Python remove_all_users怎么用?Python remove_all_users使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了remove_all_users函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_copy_db
def test_copy_db(self):
authed_client = auth_context.client
if is_mongos(authed_client):
raise SkipTest("SERVER-6427")
c = MongoClient(host, port)
authed_client.admin.add_user("admin", "password")
c.admin.authenticate("admin", "password")
c.drop_database("pymongo_test")
c.drop_database("pymongo_test1")
c.pymongo_test.test.insert({"foo": "bar"})
try:
c.pymongo_test.add_user("mike", "password")
self.assertRaises(OperationFailure, c.copy_database,
"pymongo_test", "pymongo_test1",
username="foo", password="bar")
self.assertFalse("pymongo_test1" in c.database_names())
self.assertRaises(OperationFailure, c.copy_database,
"pymongo_test", "pymongo_test1",
username="mike", password="bar")
self.assertFalse("pymongo_test1" in c.database_names())
c.copy_database("pymongo_test", "pymongo_test1",
username="mike", password="password")
self.assertTrue("pymongo_test1" in c.database_names())
self.assertEqual("bar", c.pymongo_test1.test.find_one()["foo"])
finally:
# Cleanup
remove_all_users(c.pymongo_test)
c.admin.remove_user("admin")
c.disconnect()
开发者ID:hedgepigdaniel,项目名称:mongo-python-driver,代码行数:35,代码来源:test_auth.py
示例2: tearDown
def tearDown(self):
db = self.client.pymongo_test
db.logout()
db.authenticate('dbOwner', 'pw')
db.command('dropRole', 'noremove')
remove_all_users(db)
db.logout()
开发者ID:1060460048,项目名称:mongo-python-driver,代码行数:7,代码来源:test_bulk.py
示例3: test_mongodb_x509_auth
def test_mongodb_x509_auth(self):
# Expects the server to be running with the server.pem, ca.pem
# and crl.pem provided in mongodb and the server tests as well as
# --auth
#
# --sslPEMKeyFile=jstests/libs/server.pem
# --sslCAFile=jstests/libs/ca.pem
# --sslCRLFile=jstests/libs/crl.pem
# --auth
if not CERT_SSL:
raise SkipTest("No mongod available over SSL with certs")
client = MongoClient(host, port, ssl=True, ssl_certfile=CLIENT_PEM)
if not version.at_least(client, (2, 5, 3, -1)):
raise SkipTest("MONGODB-X509 tests require MongoDB 2.5.3 or newer")
if not server_started_with_auth(client):
raise SkipTest('Authentication is not enabled on server')
# Give admin all necessary privileges.
client['$external'].add_user(MONGODB_X509_USERNAME, roles=[
{'role': 'readWriteAnyDatabase', 'db': 'admin'},
{'role': 'userAdminAnyDatabase', 'db': 'admin'}])
coll = client.pymongo_test.test
self.assertRaises(OperationFailure, coll.count)
self.assertTrue(client.admin.authenticate(MONGODB_X509_USERNAME,
mechanism='MONGODB-X509'))
self.assertTrue(coll.remove())
uri = ('mongodb://%[email protected]%s:%d/?authMechanism='
'MONGODB-X509' % (quote_plus(MONGODB_X509_USERNAME), host, port))
# SSL options aren't supported in the URI...
self.assertTrue(MongoClient(uri, ssl=True, ssl_certfile=CLIENT_PEM))
# Cleanup
remove_all_users(client['$external'])
client['$external'].logout()
开发者ID:3rf,项目名称:mongo-python-driver,代码行数:33,代码来源:test_ssl.py
示例4: test_make_user_readonly
def test_make_user_readonly(self):
admin = self.client.admin
auth_context.client.admin.add_user('admin', 'pw')
admin.authenticate('admin', 'pw')
db = self.client.pymongo_test
try:
# Make a read-write user.
db.add_user('jesse', 'pw')
admin.logout()
# Check that we're read-write by default.
db.authenticate('jesse', 'pw')
db.collection.insert({})
db.logout()
# Make the user read-only.
admin.authenticate('admin', 'pw')
db.add_user('jesse', 'pw', read_only=True)
admin.logout()
db.authenticate('jesse', 'pw')
self.assertRaises(OperationFailure, db.collection.insert, {})
finally:
# Cleanup
admin.authenticate('admin', 'pw')
remove_all_users(db)
admin.remove_user("admin")
admin.logout()
开发者ID:arunbaabte,项目名称:DjangoAppService,代码行数:30,代码来源:test_auth.py
示例5: test_copy_db
def test_copy_db(self):
authed_client = auth_context.client
if version.at_least(authed_client, (2, 7, 2)):
raise SkipTest("SERVER-17034")
authed_client.admin.add_user("admin", "password")
c = self._get_client()
c.admin.authenticate("admin", "password")
c.drop_database("pymongo_test1")
c.pymongo_test.test.insert({"foo": "bar"})
try:
c.pymongo_test.add_user("mike", "password")
self.assertRaises(OperationFailure, c.copy_database,
"pymongo_test", "pymongo_test1",
username="foo", password="bar")
self.assertFalse("pymongo_test1" in c.database_names())
self.assertRaises(OperationFailure, c.copy_database,
"pymongo_test", "pymongo_test1",
username="mike", password="bar")
self.assertFalse("pymongo_test1" in c.database_names())
c.copy_database("pymongo_test", "pymongo_test1",
username="mike", password="password")
self.assertTrue("pymongo_test1" in c.database_names())
res = c.pymongo_test1.test.find_one(_must_use_master=True)
self.assertEqual("bar", res["foo"])
finally:
# Cleanup
remove_all_users(c.pymongo_test)
c.admin.remove_user("admin")
c.close()
开发者ID:arunbaabte,项目名称:DjangoAppService,代码行数:34,代码来源:test_auth.py
示例6: test_authenticate_and_request
def test_authenticate_and_request(self):
if (is_mongos(self.client) and not
version.at_least(self.client, (2, 0, 0))):
raise SkipTest("Auth with sharding requires MongoDB >= 2.0.0")
# Database.authenticate() needs to be in a request - check that it
# always runs in a request, and that it restores the request state
# (in or not in a request) properly when it's finished.
self.assertFalse(self.client.auto_start_request)
db = self.client.pymongo_test
remove_all_users(db)
db.add_user("mike", "password",
roles=["userAdmin", "dbAdmin", "readWrite"])
self.assertFalse(self.client.in_request())
self.assertTrue(db.authenticate("mike", "password"))
self.assertFalse(self.client.in_request())
request_cx = get_client(auto_start_request=True)
request_db = request_cx.pymongo_test
self.assertTrue(request_cx.in_request())
self.assertTrue(request_db.authenticate("mike", "password"))
self.assertTrue(request_cx.in_request())
# just make sure there are no exceptions here
db.remove_user("mike")
db.logout()
request_db.logout()
开发者ID:cch2000,项目名称:mongo-python-driver,代码行数:27,代码来源:test_database.py
示例7: test_authenticate_and_safe
def test_authenticate_and_safe(self):
if (is_mongos(self.client) and not
version.at_least(self.client, (2, 0, 0))):
raise SkipTest("Auth with sharding requires MongoDB >= 2.0.0")
db = self.client.auth_test
remove_all_users(db)
db.add_user("bernie", "password",
roles=["userAdmin", "dbAdmin", "readWrite"])
db.authenticate("bernie", "password")
db.test.remove({})
self.assertTrue(db.test.insert({"bim": "baz"}))
self.assertEqual(1, db.test.count())
self.assertEqual(1,
db.test.update({"bim": "baz"},
{"$set": {"bim": "bar"}}).get('n'))
self.assertEqual(1,
db.test.remove({}).get('n'))
self.assertEqual(0, db.test.count())
db.remove_user("bernie")
db.logout()
开发者ID:cch2000,项目名称:mongo-python-driver,代码行数:25,代码来源:test_database.py
示例8: test_init_disconnected_with_auth
def test_init_disconnected_with_auth(self):
c = self._get_client()
if not server_started_with_auth(c):
raise SkipTest('Authentication is not enabled on server')
c.admin.add_user("admin", "pass")
c.admin.authenticate("admin", "pass")
try:
c.pymongo_test.add_user("user", "pass", roles=['readWrite', 'userAdmin'])
# Auth with lazy connection.
host = one(self.hosts)
uri = "mongodb://user:[email protected]%s:%d/pymongo_test?replicaSet=%s" % (
host[0], host[1], self.name)
authenticated_client = MongoReplicaSetClient(uri, _connect=False)
authenticated_client.pymongo_test.test.find_one()
# Wrong password.
bad_uri = "mongodb://user:[email protected]%s:%d/pymongo_test?replicaSet=%s" % (
host[0], host[1], self.name)
bad_client = MongoReplicaSetClient(bad_uri, _connect=False)
self.assertRaises(
OperationFailure, bad_client.pymongo_test.test.find_one)
finally:
# Clean up.
remove_all_users(c.pymongo_test)
remove_all_users(c.admin)
开发者ID:MrMission,项目名称:spider,代码行数:30,代码来源:test_replica_set_client.py
示例9: test_auth_network_error
def test_auth_network_error(self):
# Make sure there's no semaphore leak if we get a network error
# when authenticating a new socket with cached credentials.
auth_client = self._get_client()
if not server_started_with_auth(auth_client):
raise SkipTest('Authentication is not enabled on server')
auth_client.admin.add_user('admin', 'password')
auth_client.admin.authenticate('admin', 'password')
try:
# Get a client with one socket so we detect if it's leaked.
c = self._get_client(max_pool_size=1, waitQueueTimeoutMS=1)
# Simulate an authenticate() call on a different socket.
credentials = auth._build_credentials_tuple(
'MONGODB-CR', 'admin',
unicode('admin'), unicode('password'),
{})
c._cache_credentials('test', credentials, connect=False)
# Cause a network error on the actual socket.
pool = get_pool(c)
socket_info = one(pool.sockets)
socket_info.sock.close()
# In __check_auth, the client authenticates its socket with the
# new credential, but gets a socket.error. Should be reraised as
# AutoReconnect.
self.assertRaises(AutoReconnect, c.test.collection.find_one)
# No semaphore leak, the pool is allowed to make a new socket.
c.test.collection.find_one()
finally:
remove_all_users(auth_client.admin)
开发者ID:AlexSnet,项目名称:oneline,代码行数:35,代码来源:test_replica_set_client.py
示例10: tearDown
def tearDown(self):
# Remove auth users from databases
self.client.admin.authenticate("admin-user", "password")
remove_all_users(self.client.auth_test)
self.client.drop_database('auth_test')
remove_all_users(self.client.admin)
# Clear client reference so that RSC's monitor thread
# dies.
self.client = None
开发者ID:Parastou,项目名称:mongo-python-driver,代码行数:9,代码来源:test_threads.py
示例11: test_authenticate_multiple
def test_authenticate_multiple(self):
client = get_client()
if is_mongos(client) and not version.at_least(self.client, (2, 2, 0)):
raise SkipTest("Need mongos >= 2.2.0")
if not server_started_with_auth(client):
raise SkipTest("Authentication is not enabled on server")
# Setup
users_db = client.pymongo_test
admin_db = client.admin
other_db = client.pymongo_test1
users_db.test.remove()
other_db.test.remove()
admin_db.add_user("admin", "pass", roles=["userAdminAnyDatabase", "dbAdmin", "clusterAdmin", "readWrite"])
try:
self.assertTrue(admin_db.authenticate("admin", "pass"))
if version.at_least(self.client, (2, 5, 3, -1)):
admin_db.add_user("ro-admin", "pass", roles=["userAdmin", "readAnyDatabase"])
else:
admin_db.add_user("ro-admin", "pass", read_only=True)
users_db.add_user("user", "pass", roles=["userAdmin", "readWrite"])
admin_db.logout()
self.assertRaises(OperationFailure, users_db.test.find_one)
# 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(0, other_db.test.count())
self.assertRaises(OperationFailure, other_db.test.insert, {})
# Force close all sockets
client.disconnect()
# We should still be able to write to the regular user's db
self.assertTrue(users_db.test.remove())
# 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, {})
# Cleanup
finally:
admin_db.logout()
users_db.logout()
admin_db.authenticate("admin", "pass")
remove_all_users(users_db)
remove_all_users(admin_db)
开发者ID:Tokutek,项目名称:mongo-python-driver,代码行数:57,代码来源:test_database.py
示例12: test_default_roles
def test_default_roles(self):
if not version.at_least(self.client, (2, 5, 3, -1)):
raise SkipTest("Default roles only exist in MongoDB >= 2.5.3")
if not server_started_with_auth(self.client):
raise SkipTest('Authentication is not enabled on server')
# "Admin" user
db = self.client.admin
db.add_user('admin', 'pass')
try:
db.authenticate('admin', 'pass')
info = db.command('usersInfo', 'admin')['users'][0]
self.assertEqual("root", info['roles'][0]['role'])
# Read only "admin" user
db.add_user('ro-admin', 'pass', read_only=True)
db.logout()
db.authenticate('ro-admin', 'pass')
info = db.command('usersInfo', 'ro-admin')['users'][0]
self.assertEqual("readAnyDatabase", info['roles'][0]['role'])
db.logout()
# Cleanup
finally:
db.authenticate('admin', 'pass')
remove_all_users(db)
db.logout()
db.connection.disconnect()
# "Non-admin" user
db = self.client.pymongo_test
db.add_user('user', 'pass')
try:
db.authenticate('user', 'pass')
info = db.command('usersInfo', 'user')['users'][0]
self.assertEqual("dbOwner", info['roles'][0]['role'])
# Read only "Non-admin" user
db.add_user('ro-user', 'pass', read_only=True)
db.logout()
db.authenticate('ro-user', 'pass')
info = db.command('usersInfo', 'ro-user')['users'][0]
self.assertEqual("read", info['roles'][0]['role'])
db.logout()
# Cleanup
finally:
db.authenticate('user', 'pass')
remove_all_users(db)
db.logout()
开发者ID:ArturFis,项目名称:mongo-python-driver,代码行数:51,代码来源:test_database.py
示例13: test_auth_from_uri
def test_auth_from_uri(self):
c = MongoClient(host, port)
# Sharded auth not supported before MongoDB 2.0
if is_mongos(c) and not version.at_least(c, (2, 0, 0)):
raise SkipTest("Auth with sharding requires MongoDB >= 2.0.0")
if not server_started_with_auth(c):
raise SkipTest('Authentication is not enabled on server')
c.admin.add_user("admin", "pass")
c.admin.authenticate("admin", "pass")
try:
c.pymongo_test.add_user("user", "pass", roles=['userAdmin', 'readWrite'])
self.assertRaises(ConfigurationError, MongoClient,
"mongodb://foo:[email protected]%s:%d" % (host, port))
self.assertRaises(ConfigurationError, MongoClient,
"mongodb://admin:[email protected]%s:%d" % (host, port))
self.assertRaises(ConfigurationError, MongoClient,
"mongodb://user:[email protected]%s:%d" % (host, port))
MongoClient("mongodb://admin:[email protected]%s:%d" % (host, port))
self.assertRaises(ConfigurationError, MongoClient,
"mongodb://admin:[email protected]%s:%d/pymongo_test" %
(host, port))
self.assertRaises(ConfigurationError, MongoClient,
"mongodb://user:[email protected]%s:%d/pymongo_test" %
(host, port))
MongoClient("mongodb://user:[email protected]%s:%d/pymongo_test" %
(host, port))
# Auth with lazy connection.
MongoClient(
"mongodb://user:[email protected]%s:%d/pymongo_test" % (host, port),
_connect=False).pymongo_test.test.find_one()
# Wrong password.
bad_client = MongoClient(
"mongodb://user:[email protected]%s:%d/pymongo_test" % (host, port),
_connect=False)
# If auth fails with lazy connection, MongoClient raises
# AutoReconnect instead of the more appropriate OperationFailure,
# PYTHON-517.
self.assertRaises(
PyMongoError, bad_client.pymongo_test.test.find_one)
finally:
# Clean up.
remove_all_users(c.pymongo_test)
remove_all_users(c.admin)
开发者ID:quantopian,项目名称:mongo-python-driver,代码行数:50,代码来源:test_client.py
示例14: setUp
def setUp(self):
client = self._get_client()
if not server_started_with_auth(client):
raise SkipTest("Authentication is not enabled on server")
self.client = client
remove_all_users(self.client.admin)
self.client.admin.add_user('admin-user', 'password',
roles=['clusterAdmin',
'dbAdminAnyDatabase',
'readWriteAnyDatabase',
'userAdminAnyDatabase'])
self.client.admin.authenticate("admin-user", "password")
remove_all_users(self.client.auth_test)
self.client.auth_test.add_user("test-user", "password",
roles=['readWrite'])
开发者ID:Parastou,项目名称:mongo-python-driver,代码行数:15,代码来源:test_threads.py
示例15: test_mongodb_x509_auth
def test_mongodb_x509_auth(self):
# Expects the server to be running with the server.pem, ca.pem
# and crl.pem provided in mongodb and the server tests as well as
# --auth
#
# --sslPEMKeyFile=/path/to/pymongo/test/certificates/server.pem
# --sslCAFile=/path/to/pymongo/test/certificates/ca.pem
# --sslCRLFile=/path/to/pymongo/test/certificates/crl.pem
# --auth
if not CERT_SSL:
raise SkipTest("No mongod available over SSL with certs")
client = MongoClient(host, port, ssl=True, ssl_certfile=CLIENT_PEM)
if not version.at_least(client, (2, 5, 3, -1)):
raise SkipTest("MONGODB-X509 tests require MongoDB 2.5.3 or newer")
if not server_started_with_auth(client):
raise SkipTest("Authentication is not enabled on server")
# Give admin all necessary privileges.
client["$external"].add_user(
MONGODB_X509_USERNAME,
roles=[{"role": "readWriteAnyDatabase", "db": "admin"}, {"role": "userAdminAnyDatabase", "db": "admin"}],
)
coll = client.pymongo_test.test
self.assertRaises(OperationFailure, coll.count)
self.assertTrue(client.admin.authenticate(MONGODB_X509_USERNAME, mechanism="MONGODB-X509"))
self.assertTrue(coll.remove())
uri = "mongodb://%[email protected]%s:%d/?authMechanism=" "MONGODB-X509" % (quote_plus(MONGODB_X509_USERNAME), host, port)
# SSL options aren't supported in the URI...
self.assertTrue(MongoClient(uri, ssl=True, ssl_certfile=CLIENT_PEM))
# Should require a username
uri = "mongodb://%s:%d/?authMechanism=MONGODB-X509" % (host, port)
client_bad = MongoClient(uri, ssl=True, ssl_certfile=CLIENT_PEM)
self.assertRaises(OperationFailure, client_bad.pymongo_test.test.remove)
# Auth should fail if username and certificate do not match
uri = "mongodb://%[email protected]%s:%d/?authMechanism=" "MONGODB-X509" % (quote_plus("not the username"), host, port)
self.assertRaises(ConfigurationError, MongoClient, uri, ssl=True, ssl_certfile=CLIENT_PEM)
self.assertRaises(OperationFailure, client.admin.authenticate, "not the username", mechanism="MONGODB-X509")
# Invalid certificate (using CA certificate as client certificate)
uri = "mongodb://%[email protected]%s:%d/?authMechanism=" "MONGODB-X509" % (quote_plus(MONGODB_X509_USERNAME), host, port)
self.assertRaises(ConnectionFailure, MongoClient, uri, ssl=True, ssl_certfile=CA_PEM)
self.assertRaises(ConnectionFailure, MongoClient, pair, ssl=True, ssl_certfile=CA_PEM)
# Cleanup
remove_all_users(client["$external"])
client["$external"].logout()
开发者ID:hedgepigdaniel,项目名称:mongo-python-driver,代码行数:48,代码来源:test_ssl.py
示例16: test_copy_db_auth
def test_copy_db_auth(self):
# See SERVER-6427.
cx = self.get_client()
if cx.is_mongos:
raise SkipTest("Can't copy database with auth via mongos.")
target_db_name = "motor_test_2"
collection = cx.motor_test.test_collection
yield collection.remove()
yield collection.insert({"_id": 1})
yield cx.admin.add_user("admin", "password")
yield cx.admin.authenticate("admin", "password")
try:
yield cx.motor_test.add_user("mike", "password")
with assert_raises(pymongo.errors.OperationFailure):
yield cx.copy_database("motor_test", target_db_name, username="foo", password="bar")
with assert_raises(pymongo.errors.OperationFailure):
yield cx.copy_database("motor_test", target_db_name, username="mike", password="bar")
# Copy a database using name and password.
yield cx.copy_database("motor_test", target_db_name, username="mike", password="password")
self.assertEqual({"_id": 1}, (yield cx[target_db_name].test_collection.find_one()))
yield cx.drop_database(target_db_name)
finally:
yield remove_all_users(cx.motor_test)
yield cx.admin.remove_user("admin")
开发者ID:hackdog,项目名称:motor,代码行数:33,代码来源:motor_client_test_generic.py
示例17: test_copy_db_auth_concurrent
def test_copy_db_auth_concurrent(self):
# SERVER-6427, can't copy database via mongos with auth.
yield skip_if_mongos(self.cx)
n_copies = 2
test_db_names = ['motor_test_%s' % i for i in range(n_copies)]
# 1. Drop old test DBs
yield self.cx.drop_database('motor_test')
yield self.collection.insert({'_id': 1})
yield self.drop_databases(test_db_names)
# 2. Copy a test DB N times at once
try:
# self.cx is logged in as root.
yield self.cx.motor_test.add_user('mike', 'password')
client = self.get_client()
results = yield [
client.copy_database(
'motor_test', test_db_name,
username='mike', password='password')
for test_db_name in test_db_names]
self.assertTrue(all(isinstance(i, dict) for i in results))
yield self.check_copydb_results({'_id': 1}, test_db_names)
finally:
yield remove_all_users(client.motor_test)
yield self.drop_databases(
test_db_names,
authenticated_client=self.cx)
开发者ID:Vallher,项目名称:motor,代码行数:33,代码来源:motor_client_test_generic.py
示例18: test_copy_db_auth_concurrent
def test_copy_db_auth_concurrent(self):
cx = self.get_client()
yield skip_if_mongos(cx)
n_copies = 2
test_db_names = ['motor_test_%s' % i for i in range(n_copies)]
# 1. Drop old test DBs
yield cx.drop_database('motor_test')
yield self.drop_databases(test_db_names)
# 2. Copy a test DB N times at once
collection = cx.motor_test.test_collection
yield collection.remove()
yield collection.insert({'_id': 1})
yield cx.admin.add_user('admin', 'password', )
yield cx.admin.authenticate('admin', 'password')
try:
yield cx.motor_test.add_user('mike', 'password')
results = yield [
cx.copy_database(
'motor_test', test_db_name,
username='mike', password='password')
for test_db_name in test_db_names]
self.assertTrue(all(isinstance(i, dict) for i in results))
yield self.check_copydb_results({'_id': 1}, test_db_names)
finally:
yield remove_all_users(cx.motor_test)
yield self.drop_databases(test_db_names, authenticated_client=cx)
yield cx.admin.remove_user('admin')
开发者ID:devlato,项目名称:motor,代码行数:35,代码来源:motor_client_test_generic.py
示例19: test_auth_from_uri
def test_auth_from_uri(self):
if not test.env.auth:
raise SkipTest('Authentication is not enabled on server')
# self.db is logged in as root.
yield from remove_all_users(self.db)
db = self.db
try:
yield from db.add_user(
'mike', 'password',
roles=['userAdmin', 'readWrite'])
client = motor_asyncio.AsyncIOMotorClient(
'mongodb://u:[email protected]%s:%d' % (host, port),
io_loop=self.loop)
# Note: open() only calls ismaster, doesn't throw auth errors.
yield from client.open()
with self.assertRaises(OperationFailure):
yield from client.db.collection.find_one()
client = motor_asyncio.AsyncIOMotorClient(
'mongodb://mike:[email protected]%s:%d/%s' %
(host, port, db.name),
io_loop=self.loop)
yield from client[db.name].collection.find_one()
finally:
yield from db.remove_user('mike')
开发者ID:jettify,项目名称:motor,代码行数:30,代码来源:test_asyncio_client.py
示例20: test_copy_db_auth
def test_copy_db_auth(self):
# SERVER-6427, can't copy database via mongos with auth.
yield skip_if_mongos(self.cx)
yield self.collection.insert({'_id': 1})
try:
# self.cx is logged in as root.
yield self.cx.motor_test.add_user('mike', 'password')
client = self.get_client()
target_db_name = 'motor_test_2'
with assert_raises(pymongo.errors.OperationFailure):
yield client.copy_database(
'motor_test', target_db_name,
username='foo', password='bar')
with assert_raises(pymongo.errors.OperationFailure):
yield client.copy_database(
'motor_test', target_db_name,
username='mike', password='bar')
# Copy a database using name and password.
yield client.copy_database(
'motor_test', target_db_name,
username='mike', password='password')
self.assertEqual(
{'_id': 1},
(yield client[target_db_name].test_collection.find_one()))
yield client.drop_database(target_db_name)
finally:
yield remove_all_users(self.cx.motor_test)
开发者ID:Vallher,项目名称:motor,代码行数:35,代码来源:motor_client_test_generic.py
注:本文中的test.utils.remove_all_users函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论