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

use cursorInfo command instead of intr collection

Browse Source
This commit is contained in:
Mike Dirolf 2009-02-23 18:08:52 -05:00
parent f7a1349f74
commit 33e563147e
1 changed files with 14 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
test/test_cursor.py
View File

@ -282,48 +282,47 @@ class TestCursor(unittest.TestCase):
def test_kill_cursors(self):
db = self.db
db.drop_collection("test")
intr = self.db.connection().intr
client_cursors = intr.cursors.find_one()["clientCursors_size"]
by_location = intr.cursors.find_one()["byLocation_size"]
client_cursors = db._command({"cursorInfo": 1})["clientCursors_size"]
by_location = db._command({"cursorInfo": 1})["byLocation_size"]
for i in range(10000):
db.test.insert({"i": i})
self.assertEqual(client_cursors, intr.cursors.find_one()["clientCursors_size"])
self.assertEqual(by_location, intr.cursors.find_one()["byLocation_size"])
self.assertEqual(client_cursors, db._command({"cursorInfo": 1})["clientCursors_size"])
self.assertEqual(by_location, db._command({"cursorInfo": 1})["byLocation_size"])
for _ in range(10):
db.test.find_one()
self.assertEqual(client_cursors, intr.cursors.find_one()["clientCursors_size"])
self.assertEqual(by_location, intr.cursors.find_one()["byLocation_size"])
self.assertEqual(client_cursors, db._command({"cursorInfo": 1})["clientCursors_size"])
self.assertEqual(by_location, db._command({"cursorInfo": 1})["byLocation_size"])
for _ in range(10):
for x in db.test.find():
break
self.assertEqual(client_cursors, intr.cursors.find_one()["clientCursors_size"])
self.assertEqual(by_location, intr.cursors.find_one()["byLocation_size"])
self.assertEqual(client_cursors, db._command({"cursorInfo": 1})["clientCursors_size"])
self.assertEqual(by_location, db._command({"cursorInfo": 1})["byLocation_size"])
a = db.test.find()
for x in a:
break
self.assertNotEqual(client_cursors, intr.cursors.find_one()["clientCursors_size"])
self.assertNotEqual(by_location, intr.cursors.find_one()["byLocation_size"])
self.assertNotEqual(client_cursors, db._command({"cursorInfo": 1})["clientCursors_size"])
self.assertNotEqual(by_location, db._command({"cursorInfo": 1})["byLocation_size"])
del a
self.assertEqual(client_cursors, intr.cursors.find_one()["clientCursors_size"])
self.assertEqual(by_location, intr.cursors.find_one()["byLocation_size"])
self.assertEqual(client_cursors, db._command({"cursorInfo": 1})["clientCursors_size"])
self.assertEqual(by_location, db._command({"cursorInfo": 1})["byLocation_size"])
a = db.test.find().limit(10)
for x in a:
break
self.assertEqual(client_cursors, intr.cursors.find_one()["clientCursors_size"])
self.assertEqual(by_location, intr.cursors.find_one()["byLocation_size"])
self.assertEqual(client_cursors, db._command({"cursorInfo": 1})["clientCursors_size"])
self.assertEqual(by_location, db._command({"cursorInfo": 1})["byLocation_size"])
if __name__ == "__main__":
unittest.main()