"""Test Database features on a client that does not connect.

@classmethod

def setUpClass(cls):

cls.client = MongoClient(connect=False)

def test_name(self):

self.assertRaises(TypeError, Database, self.client, 4)

self.assertRaises(InvalidName, Database, self.client, "my db")

self.assertRaises(InvalidName, Database, self.client, 'my"db')

self.assertRaises(InvalidName, Database, self.client, "my\x00db")

self.assertRaises(InvalidName, Database,

self.client, u"my\u0000db")

self.assertEqual("name", Database(self.client, "name").name)

def test_get_collection(self):

codec_options = CodecOptions(tz_aware=True)

write_concern = WriteConcern(w=2, j=True)

read_concern = ReadConcern('majority')

coll = self.client.pymongo_test.get_collection(

'foo', codec_options, ReadPreference.SECONDARY, write_concern,

read_concern)

self.assertEqual('foo', coll.name)

self.assertEqual(codec_options, coll.codec_options)

self.assertEqual(ReadPreference.SECONDARY, coll.read_preference)

self.assertEqual(write_concern, coll.write_concern)

self.assertEqual(read_concern, coll.read_concern)

def test_getattr(self):

db = self.client.pymongo_test

self.assertTrue(isinstance(db['_does_not_exist'], Collection))

with self.assertRaises(AttributeError) as context:

db._does_not_exist

# Message should be: "AttributeError: Database has no attribute

# '_does_not_exist'. To access the _does_not_exist collection,

# use database['_does_not_exist']".

self.assertIn("has no attribute '_does_not_exist'",

str(context.exception))

def test_iteration(self):

def test_equality(self):

self.assertNotEqual(Database(self.client, "test"),

Database(self.client, "mike"))

self.assertEqual(Database(self.client, "test"),

Database(self.client, "test"))

# Explicitly test inequality

self.assertFalse(Database(self.client, "test") !=

Database(self.client, "test"))

def test_hashable(self):

self.assertIn(self.client.test, {Database(self.client, "test")})

def test_get_coll(self):

db = Database(self.client, "pymongo_test")

self.assertEqual(db.test, db["test"])

self.assertEqual(db.test, Collection(db, "test"))

self.assertNotEqual(db.test, Collection(db, "mike"))

self.assertEqual(db.test.mike, db["test.mike"])

def test_repr(self):

self.assertEqual(repr(Database(self.client, "pymongo_test")),

"Database(%r, %s)" % (self.client,

repr(u"pymongo_test")))

def test_create_collection(self):

db = Database(self.client, "pymongo_test")

db.test.insert_one({"hello": "world"})

self.assertRaises(CollectionInvalid, db.create_collection, "test")

db.drop_collection("test")

self.assertRaises(TypeError, db.create_collection, 5)

self.assertRaises(TypeError, db.create_collection, None)

self.assertRaises(InvalidName, db.create_collection, "coll..ection")

test = db.create_collection("test")

self.assertTrue(u"test" in db.list_collection_names())

test.insert_one({"hello": u"world"})

self.assertEqual(db.test.find_one()["hello"], "world")

db.drop_collection("test.foo")

db.create_collection("test.foo")

self.assertTrue(u"test.foo" in db.list_collection_names())

self.assertRaises(CollectionInvalid, db.create_collection, "test.foo")

def _test_collection_names(self, meth, **no_system_kwargs):

db = Database(self.client, "pymongo_test")

db.test.insert_one({"dummy": u"object"})

db.test.mike.insert_one({"dummy": u"object"})

colls = getattr(db, meth)()

self.assertTrue("test" in colls)

self.assertTrue("test.mike" in colls)

for coll in colls:

self.assertTrue("$" not in coll)

db.systemcoll.test.insert_one({})

no_system_collections = getattr(db, meth)(**no_system_kwargs)

for coll in no_system_collections:

self.assertTrue(not coll.startswith("system."))

self.assertIn("systemcoll.test", no_system_collections)

# Force more than one batch.

db = self.client.many_collections

for i in range(101):

db["coll" + str(i)].insert_one({})

# No Error

try:

getattr(db, meth)()

finally:

self.client.drop_database("many_collections")

def test_collection_names(self):

self._test_collection_names(

'collection_names', include_system_collections=False)

def test_list_collection_names(self):

self._test_collection_names(

'list_collection_names', filter={

"name": {"$regex": r"^(?!system\.)"}})

def test_list_collection_names_filter(self):

listener = OvertCommandListener()

results = listener.results

client = rs_or_single_client(event_listeners=[listener])

db = client[self.db.name]

db.capped.drop()

db.create_collection("capped", capped=True, size=4096)

db.capped.insert_one({})

db.non_capped.insert_one({})

self.addCleanup(client.drop_database, db.name)

# Should not send nameOnly.

for filter in ({'options.capped': True},

{'options.capped': True, 'name': 'capped'}):

results.clear()

names = db.list_collection_names(filter=filter)

self.assertEqual(names, ["capped"])

self.assertNotIn("nameOnly", results["started"][0].command)

# Should send nameOnly (except on 2.6).

for filter in (None, {}, {'name': {'$in': ['capped', 'non_capped']}}):

results.clear()

names = db.list_collection_names(filter=filter)

self.assertIn("capped", names)

self.assertIn("non_capped", names)

command = results["started"][0].command

if client_context.version >= (3, 0):

self.assertIn("nameOnly", command)

self.assertTrue(command["nameOnly"])

else:

self.assertNotIn("nameOnly", command)

def test_list_collections(self):

self.client.drop_database("pymongo_test")

db = Database(self.client, "pymongo_test")

db.test.insert_one({"dummy": u"object"})

db.test.mike.insert_one({"dummy": u"object"})

results = db.list_collections()

colls = [result["name"] for result in results]

# All the collections present.

self.assertTrue("test" in colls)

self.assertTrue("test.mike" in colls)

# No collection containing a '$'.

for coll in colls:

self.assertTrue("$" not in coll)

# Duplicate check.

coll_cnt = {}

for coll in colls:

try:

# Found duplicate.

coll_cnt[coll] += 1

self.assertTrue(False)

except KeyError:

coll_cnt[coll] = 1

coll_cnt = {}

# Checking if is there any collection which don't exists.

if (len(set(colls) - set(["test","test.mike"])) == 0 or

len(set(colls) - set(["test","test.mike","system.indexes"])) == 0):

self.assertTrue(True)

else:

self.assertTrue(False)

colls = db.list_collections(filter={"name": {"$regex": "^test$"}})

self.assertEqual(1, len(list(colls)))

colls = db.list_collections(filter={"name": {"$regex": "^test.mike$"}})

self.assertEqual(1, len(list(colls)))

db.drop_collection("test")

db.create_collection("test", capped=True, size=4096)

results = db.list_collections(filter={'options.capped': True})

colls = [result["name"] for result in results]

# Checking only capped collections are present

self.assertTrue("test" in colls)

self.assertFalse("test.mike" in colls)

# No collection containing a '$'.

for coll in colls:

self.assertTrue("$" not in coll)

# Duplicate check.

coll_cnt = {}

for coll in colls:

try:

# Found duplicate.

coll_cnt[coll] += 1

self.assertTrue(False)

except KeyError:

coll_cnt[coll] = 1

coll_cnt = {}

# Checking if is there any collection which don't exists.

if (len(set(colls) - set(["test"])) == 0 or

len(set(colls) - set(["test","system.indexes"])) == 0):

self.assertTrue(True)

else:

self.assertTrue(False)

self.client.drop_database("pymongo_test")

def test_collection_names_single_socket(self):

# Test that Database.collection_names only requires one socket.

client = rs_or_single_client(maxPoolSize=1)

client.drop_database('test_collection_names_single_socket')

db = client.test_collection_names_single_socket

for i in range(200):

db.create_collection(str(i))

db.list_collection_names() # Must not hang.

client.drop_database('test_collection_names_single_socket')

def test_drop_collection(self):

db = Database(self.client, "pymongo_test")

self.assertRaises(TypeError, db.drop_collection, 5)

self.assertRaises(TypeError, db.drop_collection, None)

db.test.insert_one({"dummy": u"object"})

self.assertTrue("test" in db.list_collection_names())

db.drop_collection("test")

self.assertFalse("test" in db.list_collection_names())

db.test.insert_one({"dummy": u"object"})

self.assertTrue("test" in db.list_collection_names())

db.drop_collection(u"test")

self.assertFalse("test" in db.list_collection_names())

db.test.insert_one({"dummy": u"object"})

self.assertTrue("test" in db.list_collection_names())

db.drop_collection(db.test)

self.assertFalse("test" in db.list_collection_names())

db.test.insert_one({"dummy": u"object"})

self.assertTrue("test" in db.list_collection_names())

db.test.drop()

self.assertFalse("test" in db.list_collection_names())

db.test.drop()

db.drop_collection(db.test.doesnotexist)

if client_context.version.at_least(3, 3, 9) and client_context.is_rs:

db_wc = Database(self.client, 'pymongo_test',

write_concern=IMPOSSIBLE_WRITE_CONCERN)

with self.assertRaises(WriteConcernError):

db_wc.drop_collection('test')

def test_validate_collection(self):

db = self.client.pymongo_test

self.assertRaises(TypeError, db.validate_collection, 5)

self.assertRaises(TypeError, db.validate_collection, None)

db.test.insert_one({"dummy": u"object"})

self.assertRaises(OperationFailure, db.validate_collection,

"test.doesnotexist")

self.assertRaises(OperationFailure, db.validate_collection,

db.test.doesnotexist)

self.assertTrue(db.validate_collection("test"))

self.assertTrue(db.validate_collection(db.test))

self.assertTrue(db.validate_collection(db.test, full=True))

self.assertTrue(db.validate_collection(db.test, scandata=True))

self.assertTrue(db.validate_collection(db.test, scandata=True, full=True))

self.assertTrue(db.validate_collection(db.test, True, True))

@client_context.require_version_min(4, 3, 3)

def test_validate_collection_background(self):

db = self.client.pymongo_test

db.test.insert_one({"dummy": u"object"})

coll = db.test

self.assertTrue(db.validate_collection(coll, background=False))

# The inMemory storage engine does not support background=True.

if client_context.storage_engine != 'inMemory':

self.assertTrue(db.validate_collection(coll, background=True))

self.assertTrue(

db.validate_collection(coll, scandata=True, background=True))

# The server does not support background=True with full=True.

# Assert that we actually send the background option by checking

# that this combination fails.

with self.assertRaises(OperationFailure):

db.validate_collection(coll, full=True, background=True)

@client_context.require_no_mongos

@ignore_deprecations

def test_profiling_levels(self):

db = self.client.pymongo_test

self.assertEqual(db.profiling_level(), OFF) # default

self.assertRaises(ValueError, db.set_profiling_level, 5.5)

self.assertRaises(ValueError, db.set_profiling_level, None)

self.assertRaises(ValueError, db.set_profiling_level, -1)

self.assertRaises(TypeError, db.set_profiling_level, SLOW_ONLY, 5.5)

self.assertRaises(TypeError, db.set_profiling_level, SLOW_ONLY, '1')

db.set_profiling_level(SLOW_ONLY)

self.assertEqual(db.profiling_level(), SLOW_ONLY)

db.set_profiling_level(ALL)

self.assertEqual(db.profiling_level(), ALL)

db.set_profiling_level(OFF)

self.assertEqual(db.profiling_level(), OFF)

db.set_profiling_level(SLOW_ONLY, 50)

self.assertEqual(50, db.command("profile", -1)['slowms'])

db.set_profiling_level(ALL, -1)

self.assertEqual(-1, db.command("profile", -1)['slowms'])

db.set_profiling_level(OFF, 100) # back to default

self.assertEqual(100, db.command("profile", -1)['slowms'])

@client_context.require_no_mongos

@client_context.require_version_min(3, 6)

@ignore_deprecations

def test_profiling_sample_rate(self):

db = self.client.pymongo_test

with self.assertRaises(TypeError):

db.set_profiling_level(SLOW_ONLY, 50, sample_rate='1')

with self.assertRaises(TypeError):

db.set_profiling_level(SLOW_ONLY, 50, sample_rate=1)

db.set_profiling_level(SLOW_ONLY, 50, sample_rate=0.0)

db.set_profiling_level(SLOW_ONLY, 50, sample_rate=1.0)

db.set_profiling_level(SLOW_ONLY, 50, sample_rate=0.5)

profile = db.command("profile", -1)

self.assertEqual(50, profile['slowms'])

self.assertEqual(0.5, profile['sampleRate'])

db.set_profiling_level(OFF, 100) # back to default

self.assertEqual(100, db.command("profile", -1)['slowms'])

@client_context.require_no_mongos

@client_context.require_version_min(4, 4, 2)

@ignore_deprecations

def test_profiling_filter(self):

db = self.client.pymongo_test

db.set_profiling_level(ALL, filter={'ns': {'$eq': 'test.test'}})

profile = db.command("profile", -1)

self.assertEqual({'ns': {'$eq': 'test.test'}}, profile['filter'])

# filter='unset' resets the filter back to the default.

db.set_profiling_level(OFF, 100, filter='unset')

self.assertEqual(100, db.command("profile", -1)['slowms'])

@client_context.require_no_mongos

@ignore_deprecations

def test_profiling_info(self):

db = self.client.pymongo_test

db.system.profile.drop()

db.set_profiling_level(ALL)

db.test.find_one()

db.set_profiling_level(OFF)

info = db.profiling_info()

self.assertTrue(isinstance(info, list))

# Check if we're going to fail because of SERVER-4754, in which

# profiling info isn't collected if mongod was started with --auth

if server_started_with_auth(self.client):

raise SkipTest(

"We need SERVER-4754 fixed for the rest of this test to pass"

)

self.assertTrue(len(info) >= 1)

# These basically clue us in to server changes.

self.assertTrue(isinstance(info[0]['responseLength'], int))

self.assertTrue(isinstance(info[0]['millis'], int))

self.assertTrue(isinstance(info[0]['client'], string_type))

self.assertTrue(isinstance(info[0]['user'], string_type))

self.assertTrue(isinstance(info[0]['ns'], string_type))

self.assertTrue(isinstance(info[0]['op'], string_type))

self.assertTrue(isinstance(info[0]["ts"], datetime.datetime))

def test_profiling_helpers_deprecated(self):

filter = DeprecationFilter('error')

self.addCleanup(filter.stop)

db = self.client.pymongo_test

self.assertRaises(DeprecationWarning, db.profiling_level)

self.assertRaises(DeprecationWarning, db.profiling_info)

self.assertRaises(DeprecationWarning, db.set_profiling_level, OFF)

# SERVER-47817 removes the resetError command.

@client_context.require_version_max(4, 9)

@client_context.require_no_mongos

@ignore_deprecations

def test_errors(self):

# We must call getlasterror, etc. on same socket as last operation.

db = rs_or_single_client(maxPoolSize=1).pymongo_test

db.reset_error_history()

self.assertEqual(None, db.error())

if client_context.supports_getpreverror:

self.assertEqual(None, db.previous_error())

db.test.insert_one({"_id": 1})

unacked = db.test.with_options(write_concern=WriteConcern(w=0))

unacked.insert_one({"_id": 1})

self.assertTrue(db.error())

if client_context.supports_getpreverror:

self.assertTrue(db.previous_error())

unacked.insert_one({"_id": 1})

self.assertTrue(db.error())

if client_context.supports_getpreverror:

prev_error = db.previous_error()

self.assertEqual(prev_error["nPrev"], 1)

del prev_error["nPrev"]

prev_error.pop("lastOp", None)

error = db.error()

error.pop("lastOp", None)

# getLastError includes "connectionId" in recent

# server versions, getPrevError does not.

error.pop("connectionId", None)

self.assertEqualReply(error, prev_error)

db.test.find_one()

self.assertEqual(None, db.error())

if client_context.supports_getpreverror:

self.assertTrue(db.previous_error())

self.assertEqual(db.previous_error()["nPrev"], 2)

db.reset_error_history()

self.assertEqual(None, db.error())

if client_context.supports_getpreverror:

self.assertEqual(None, db.previous_error())

def test_command(self):

self.maxDiff = None

db = self.client.admin

first = db.command("buildinfo")

second = db.command({"buildinfo": 1})

third = db.command("buildinfo", 1)

self.assertEqualReply(first, second)

self.assertEqualReply(second, third)

# We use 'aggregate' as our example command, since it's an easy way to

# retrieve a BSON regex from a collection using a command. But until

# MongoDB 2.3.2, aggregation turned regexes into strings: SERVER-6470.

# Note: MongoDB 3.5.2 requires the 'cursor' or 'explain' option for

# aggregate.

@client_context.require_version_max(3, 5, 0)

def test_command_with_regex(self):

db = self.client.pymongo_test

db.test.drop()

db.test.insert_one({'r': re.compile('.*')})

db.test.insert_one({'r': Regex('.*')})

result = db.command('aggregate', 'test', pipeline=[])

for doc in result['result']:

self.assertTrue(isinstance(doc['r'], Regex))

def test_password_digest(self):

self.assertRaises(TypeError, auth._password_digest, 5)

self.assertRaises(TypeError, auth._password_digest, True)

self.assertRaises(TypeError, auth._password_digest, None)

self.assertTrue(isinstance(auth._password_digest("mike", "password"),

text_type))

self.assertEqual(auth._password_digest("mike", "password"),

u"cd7e45b3b2767dc2fa9b6b548457ed00")

self.assertEqual(auth._password_digest("mike", "password"),

auth._password_digest(u"mike", u"password"))

self.assertEqual(auth._password_digest("Gustave", u"Dor\xe9"),

u"81e0e2364499209f466e75926a162d73")

@client_context.require_auth

def test_authenticate_add_remove_user(self):

# "self.client" is logged in as root.

auth_db = self.client.pymongo_test

def check_auth(username, password):

c = rs_or_single_client_noauth(

username=username,

password=password,

authSource="pymongo_test")

c.pymongo_test.collection.find_one()

# Configuration errors

self.assertRaises(ValueError, auth_db.add_user, "user", '')

self.assertRaises(TypeError, auth_db.add_user, "user", 'password', 15)

self.assertRaises(TypeError, auth_db.add_user,

"user", 'password', 'True')

self.assertRaises(ConfigurationError, auth_db.add_user,

"user", 'password', True, roles=['read'])

with warnings.catch_warnings():

warnings.simplefilter("error", DeprecationWarning)

self.assertRaises(DeprecationWarning, auth_db.add_user,

"user", "password")

self.assertRaises(DeprecationWarning, auth_db.add_user,

"user", "password", True)

with ignore_deprecations():

self.assertRaises(ConfigurationError, auth_db.add_user,

"user", "password", digestPassword=True)

# Add / authenticate / remove

auth_db.add_user("mike", "password", roles=["read"])

self.addCleanup(remove_all_users, auth_db)

self.assertRaises(TypeError, check_auth, 5, "password")

self.assertRaises(TypeError, check_auth, "mike", 5)

self.assertRaises(OperationFailure,

check_auth, "mike", "not a real password")

self.assertRaises(OperationFailure, check_auth, "faker", "password")

check_auth("mike", "password")

if not client_context.version.at_least(3, 7, 2) or HAVE_STRINGPREP:

# Unicode name and password.

check_auth(u"mike", u"password")

auth_db.remove_user("mike")

self.assertRaises(

OperationFailure, check_auth, "mike", "password")

# Add / authenticate / change password

self.assertRaises(

OperationFailure, check_auth, "Gustave", u"Dor\xe9")

auth_db.add_user("Gustave", u"Dor\xe9", roles=["read"])

check_auth("Gustave", u"Dor\xe9")

# Change password.

auth_db.add_user("Gustave", "password", roles=["read"])

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):

# "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,

{})

@client_context.require_auth

@ignore_deprecations

def test_default_roles(self):

# "self.client" is logged in as root.

auth_admin = self.client.admin

auth_admin.add_user('test_default_roles', 'pass')

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(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')

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)

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):

# "self.client" is logged in as root.

auth_db = self.client.pymongo_test

auth_db.add_user("amalia", "password", roles=["userAdmin"])

self.addCleanup(client_context.drop_user, "pymongo_test", "amalia")

db = rs_or_single_client_noauth(username="amalia",

password="password",

authSource="pymongo_test").pymongo_test

# 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"})

@client_context.require_auth

@ignore_deprecations

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

admin_db_auth.add_user(

'ro-admin',

'pass',

roles=["userAdmin", "readAnyDatabase"])

self.addCleanup(client_context.drop_user, 'admin', 'ro-admin')

users_db_auth.add_user(

'user', 'pass', roles=["userAdmin", "readWrite"])

self.addCleanup(remove_all_users, users_db_auth)

# Non-root client.

listener = EventListener()

client = rs_or_single_client_noauth(event_listeners=[listener])

admin_db = client.admin

users_db = client.pymongo_test

other_db = client.pymongo_test1

self.assertRaises(OperationFailure, users_db.test.find_one)

self.assertEqual(listener.started_command_names(), ['find'])

listener.reset()

# Regular user should be able to query its own db, but

# no other.

users_db.authenticate('user', 'pass')

if client_context.version.at_least(3, 0):

self.assertEqual(listener.started_command_names()[0], 'saslStart')

else:

self.assertEqual(listener.started_command_names()[0], 'getnonce')

self.assertEqual(0, users_db.test.count_documents({}))

self.assertRaises(OperationFailure, other_db.test.find_one)

listener.reset()

# Admin read-only user should be able to query any db,

# but not write.

admin_db.authenticate('ro-admin', 'pass')

if client_context.version.at_least(3, 0):

self.assertEqual(listener.started_command_names()[0], 'saslStart')

else:

self.assertEqual(listener.started_command_names()[0], 'getnonce')

self.assertEqual(None, other_db.test.find_one())

self.assertRaises(OperationFailure,

other_db.test.insert_one, {})

# Close all sockets.

client.close()

listener.reset()

# We should still be able to write to the regular user's db.

self.assertTrue(users_db.test.delete_many({}))

names = listener.started_command_names()

if client_context.version.at_least(4, 4, -1):

# No speculation with multiple users (but we do skipEmptyExchange).

self.assertEqual(

names, ['saslStart', 'saslContinue', 'saslStart',

'saslContinue', 'delete'])

elif client_context.version.at_least(3, 0):

self.assertEqual(

names, ['saslStart', 'saslContinue', 'saslContinue',

'saslStart', 'saslContinue', 'saslContinue', 'delete'])

else:

self.assertEqual(

names, ['getnonce', 'authenticate',

'getnonce', 'authenticate', 'delete'])

# And read from other dbs...

self.assertEqual(0, other_db.test.count_documents({}))

# But still not write to other dbs.

self.assertRaises(OperationFailure,

other_db.test.insert_one, {})

def test_id_ordering(self):

# PyMongo attempts to have _id show up first

# when you iterate key/value pairs in a document.

# This isn't reliable since python dicts don't

# guarantee any particular order. This will never

# work right in Jython or any Python or environment

# with hash randomization enabled (e.g. tox).

db = self.client.pymongo_test

db.test.drop()

db.test.insert_one(SON([("hello", "world"),

("_id", 5)]))

db = self.client.get_database(

"pymongo_test", codec_options=CodecOptions(document_class=SON))

cursor = db.test.find()

for x in cursor:

for (k, v) in x.items():

self.assertEqual(k, "_id")

break

def test_deref(self):

db = self.client.pymongo_test

db.test.drop()

self.assertRaises(TypeError, db.dereference, 5)

self.assertRaises(TypeError, db.dereference, "hello")

self.assertRaises(TypeError, db.dereference, None)

self.assertEqual(None, db.dereference(DBRef("test", ObjectId())))

obj = {"x": True}

key = db.test.insert_one(obj).inserted_id

self.assertEqual(obj, db.dereference(DBRef("test", key)))

self.assertEqual(obj,

db.dereference(DBRef("test", key, "pymongo_test")))

self.assertRaises(ValueError,

db.dereference, DBRef("test", key, "foo"))

self.assertEqual(None, db.dereference(DBRef("test", 4)))

obj = {"_id": 4}

db.test.insert_one(obj)

self.assertEqual(obj, db.dereference(DBRef("test", 4)))

def test_deref_kwargs(self):

db = self.client.pymongo_test

db.test.drop()

db.test.insert_one({"_id": 4, "foo": "bar"})

db = self.client.get_database(

"pymongo_test", codec_options=CodecOptions(document_class=SON))

self.assertEqual(SON([("foo", "bar")]),

db.dereference(DBRef("test", 4),

projection={"_id": False}))

@client_context.require_no_auth

@client_context.require_version_max(4, 1, 0)

def test_eval(self):

db = self.client.pymongo_test

db.test.drop()

with ignore_deprecations():

self.assertRaises(TypeError, db.eval, None)

self.assertRaises(TypeError, db.eval, 5)

self.assertRaises(TypeError, db.eval, [])

self.assertEqual(3, db.eval("function (x) {return x;}", 3))

self.assertEqual(3, db.eval(u"function (x) {return x;}", 3))

self.assertEqual(None,

db.eval("function (x) {db.test.save({y:x});}", 5))

self.assertEqual(db.test.find_one()["y"], 5)

self.assertEqual(5, db.eval("function (x, y) {return x + y;}", 2, 3))

self.assertEqual(5, db.eval("function () {return 5;}"))

self.assertEqual(5, db.eval("2 + 3;"))

self.assertEqual(5, db.eval(Code("2 + 3;")))

self.assertRaises(OperationFailure, db.eval, Code("return i;"))

self.assertEqual(2, db.eval(Code("return i;", {"i": 2})))

self.assertEqual(5, db.eval(Code("i + 3;", {"i": 2})))

self.assertRaises(OperationFailure, db.eval, "5 ++ 5;")

# TODO some of these tests belong in the collection level testing.

def test_insert_find_one(self):

db = self.client.pymongo_test

db.test.drop()

a_doc = SON({"hello": u"world"})

a_key = db.test.insert_one(a_doc).inserted_id

self.assertTrue(isinstance(a_doc["_id"], ObjectId))

self.assertEqual(a_doc["_id"], a_key)

self.assertEqual(a_doc, db.test.find_one({"_id": a_doc["_id"]}))

self.assertEqual(a_doc, db.test.find_one(a_key))

self.assertEqual(None, db.test.find_one(ObjectId()))

self.assertEqual(a_doc, db.test.find_one({"hello": u"world"}))

self.assertEqual(None, db.test.find_one({"hello": u"test"}))

b = db.test.find_one()

b["hello"] = u"mike"

db.test.replace_one({"_id": b["_id"]}, b)

self.assertNotEqual(a_doc, db.test.find_one(a_key))

self.assertEqual(b, db.test.find_one(a_key))

self.assertEqual(b, db.test.find_one())

count = 0

for _ in db.test.find():

count += 1

self.assertEqual(count, 1)

def test_long(self):

db = self.client.pymongo_test

db.test.drop()

db.test.insert_one({"x": long(9223372036854775807)})

retrieved = db.test.find_one()['x']

self.assertEqual(Int64(9223372036854775807), retrieved)

self.assertIsInstance(retrieved, Int64)

db.test.delete_many({})

db.test.insert_one({"x": Int64(1)})

retrieved = db.test.find_one()['x']

self.assertEqual(Int64(1), retrieved)

self.assertIsInstance(retrieved, Int64)

def test_delete(self):

db = self.client.pymongo_test

db.test.drop()

db.test.insert_one({"x": 1})

db.test.insert_one({"x": 2})

db.test.insert_one({"x": 3})

length = 0

for _ in db.test.find():

length += 1

self.assertEqual(length, 3)

db.test.delete_one({"x": 1})

length = 0

for _ in db.test.find():

length += 1

self.assertEqual(length, 2)

db.test.delete_one(db.test.find_one())

db.test.delete_one(db.test.find_one())

self.assertEqual(db.test.find_one(), None)

db.test.insert_one({"x": 1})

db.test.insert_one({"x": 2})

db.test.insert_one({"x": 3})

self.assertTrue(db.test.find_one({"x": 2}))

db.test.delete_one({"x": 2})

self.assertFalse(db.test.find_one({"x": 2}))

self.assertTrue(db.test.find_one())

db.test.delete_many({})

self.assertFalse(db.test.find_one())

@client_context.require_no_auth

@client_context.require_version_max(4, 1, 0)

def test_system_js(self):

db = self.client.pymongo_test

db.system.js.delete_many({})

self.assertEqual(0, db.system.js.count_documents({}))

db.system_js.add = "function(a, b) { return a + b; }"

self.assertEqual('add', db.system.js.find_one()['_id'])

self.assertEqual(1, db.system.js.count_documents({}))

self.assertEqual(6, db.system_js.add(1, 5))

del db.system_js.add

self.assertEqual(0, db.system.js.count_documents({}))

db.system_js['add'] = "function(a, b) { return a + b; }"

self.assertEqual('add', db.system.js.find_one()['_id'])

self.assertEqual(1, db.system.js.count_documents({}))

self.assertEqual(6, db.system_js['add'](1, 5))

del db.system_js['add']

self.assertEqual(0, db.system.js.count_documents({}))

self.assertRaises(OperationFailure, db.system_js.add, 1, 5)

# TODO right now CodeWScope doesn't work w/ system js

# db.system_js.scope = Code("return hello;", {"hello": 8})

# self.assertEqual(8, db.system_js.scope())

self.assertRaises(OperationFailure, db.system_js.non_existant)

def test_system_js_list(self):

db = self.client.pymongo_test

db.system.js.delete_many({})

self.assertEqual([], db.system_js.list())

db.system_js.foo = "function() { return 'blah'; }"

self.assertEqual(["foo"], db.system_js.list())

db.system_js.bar = "function() { return 'baz'; }"

self.assertEqual(set(["foo", "bar"]), set(db.system_js.list()))

del db.system_js.foo

self.assertEqual(["bar"], db.system_js.list())

def test_command_response_without_ok(self):

# Sometimes (SERVER-10891) the server's response to a badly-formatted