pub/mongoengine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'master' of https://github.com/MongoEngine/mongoengine into swat_url_fix

Browse Source
This commit is contained in:
Bastien Gérard
2018-09-16 23:03:16 +02:00
parent 8b5df3ca17 dd0fdcfdd4
commit 57ccfcfc1b
87 changed files with 5596 additions and 3458 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
tests/__init__.py
View File

@@ -1,4 +1,4 @@
from all_warnings import AllWarnings
from document import *
from queryset import *
from fields import *
from .all_warnings import AllWarnings
from .document import *
from .queryset import *
from .fields import *

16
tests/document/__init__.py
View File

@@ -1,13 +1,13 @@
import unittest
from class_methods import *
from delta import *
from dynamic import *
from indexes import *
from inheritance import *
from instance import *
from json_serialisation import *
from validation import *
from .class_methods import *
from .delta import *
from .dynamic import *
from .indexes import *
from .inheritance import *
from .instance import *
from .json_serialisation import *
from .validation import *
if __name__ == '__main__':
unittest.main()

29
tests/document/class_methods.py
View File

@@ -5,6 +5,7 @@ from mongoengine import *
from mongoengine.queryset import NULLIFY, PULL
from mongoengine.connection import get_db
from tests.utils import needs_mongodb_v26
__all__ = ("ClassMethodsTest", )
@@ -65,10 +66,10 @@ class ClassMethodsTest(unittest.TestCase):
"""
collection_name = 'person'
self.Person(name='Test').save()
self.assertTrue(collection_name in self.db.collection_names())
self.assertIn(collection_name, self.db.collection_names())
self.Person.drop_collection()
self.assertFalse(collection_name in self.db.collection_names())
self.assertNotIn(collection_name, self.db.collection_names())
def test_register_delete_rule(self):
"""Ensure that register delete rule adds a delete rule to the document
@@ -187,6 +188,26 @@ class ClassMethodsTest(unittest.TestCase):
self.assertEqual(BlogPostWithTags.compare_indexes(), { 'missing': [], 'extra': [] })
self.assertEqual(BlogPostWithCustomField.compare_indexes(), { 'missing': [], 'extra': [] })
@needs_mongodb_v26
def test_compare_indexes_for_text_indexes(self):
""" Ensure that compare_indexes behaves correctly for text indexes """
class Doc(Document):
a = StringField()
b = StringField()
meta = {'indexes': [
{'fields': ['$a', "$b"],
'default_language': 'english',
'weights': {'a': 10, 'b': 2}
}
]}
Doc.drop_collection()
Doc.ensure_indexes()
actual = Doc.compare_indexes()
expected = {'missing': [], 'extra': []}
self.assertEqual(actual, expected)
def test_list_indexes_inheritance(self):
""" ensure that all of the indexes are listed regardless of the super-
or sub-class that we call it from
@@ -319,7 +340,7 @@ class ClassMethodsTest(unittest.TestCase):
meta = {'collection': collection_name}
Person(name="Test User").save()
self.assertTrue(collection_name in self.db.collection_names())
self.assertIn(collection_name, self.db.collection_names())
user_obj = self.db[collection_name].find_one()
self.assertEqual(user_obj['name'], "Test User")
@@ -328,7 +349,7 @@ class ClassMethodsTest(unittest.TestCase):
self.assertEqual(user_obj.name, "Test User")
Person.drop_collection()
self.assertFalse(collection_name in self.db.collection_names())
self.assertNotIn(collection_name, self.db.collection_names())
def test_collection_name_and_primary(self):
"""Ensure that a collection with a specified name may be used.

4
tests/document/delta.py
View File

@@ -694,7 +694,7 @@ class DeltaTest(unittest.TestCase):
organization.employees.append(person)
updates, removals = organization._delta()
self.assertEqual({}, removals)
self.assertTrue('employees' in updates)
self.assertIn('employees', updates)
def test_delta_with_dbref_false(self):
person, organization, employee = self.circular_reference_deltas_2(Document, Document, False)
@@ -709,7 +709,7 @@ class DeltaTest(unittest.TestCase):
organization.employees.append(person)
updates, removals = organization._delta()
self.assertEqual({}, removals)
self.assertTrue('employees' in updates)
self.assertIn('employees', updates)
def test_nested_nested_fields_mark_as_changed(self):
class EmbeddedDoc(EmbeddedDocument):

6
tests/document/dynamic.py
View File

@@ -174,8 +174,8 @@ class DynamicTest(unittest.TestCase):
Employee.drop_collection()
self.assertTrue('name' in Employee._fields)
self.assertTrue('salary' in Employee._fields)
self.assertIn('name', Employee._fields)
self.assertIn('salary', Employee._fields)
self.assertEqual(Employee._get_collection_name(),
self.Person._get_collection_name())
@@ -189,7 +189,7 @@ class DynamicTest(unittest.TestCase):
self.assertEqual(1, Employee.objects(age=20).count())
joe_bloggs = self.Person.objects.first()
self.assertTrue(isinstance(joe_bloggs, Employee))
self.assertIsInstance(joe_bloggs, Employee)
def test_embedded_dynamic_document(self):
"""Test dynamic embedded documents"""

85
tests/document/indexes.py
View File

@@ -2,14 +2,14 @@
import unittest
import sys
import pymongo
from nose.plugins.skip import SkipTest
from datetime import datetime
import pymongo
from mongoengine import *
from mongoengine.connection import get_db, get_connection
from mongoengine.connection import get_db
from tests.utils import get_mongodb_version, needs_mongodb_v26
__all__ = ("IndexesTest", )
@@ -70,7 +70,7 @@ class IndexesTest(unittest.TestCase):
self.assertEqual(len(info), 4)
info = [value['key'] for key, value in info.iteritems()]
for expected in expected_specs:
self.assertTrue(expected['fields'] in info)
self.assertIn(expected['fields'], info)
def _index_test_inheritance(self, InheritFrom):
@@ -102,7 +102,7 @@ class IndexesTest(unittest.TestCase):
self.assertEqual(len(info), 4)
info = [value['key'] for key, value in info.iteritems()]
for expected in expected_specs:
self.assertTrue(expected['fields'] in info)
self.assertIn(expected['fields'], info)
class ExtendedBlogPost(BlogPost):
title = StringField()
@@ -117,7 +117,7 @@ class IndexesTest(unittest.TestCase):
info = ExtendedBlogPost.objects._collection.index_information()
info = [value['key'] for key, value in info.iteritems()]
for expected in expected_specs:
self.assertTrue(expected['fields'] in info)
self.assertIn(expected['fields'], info)
def test_indexes_document_inheritance(self):
"""Ensure that indexes are used when meta[indexes] is specified for
@@ -226,7 +226,7 @@ class IndexesTest(unittest.TestCase):
list(Person.objects)
info = Person.objects._collection.index_information()
info = [value['key'] for key, value in info.iteritems()]
self.assertTrue([('rank.title', 1)] in info)
self.assertIn([('rank.title', 1)], info)
def test_explicit_geo2d_index(self):
"""Ensure that geo2d indexes work when created via meta[indexes]
@@ -246,7 +246,7 @@ class IndexesTest(unittest.TestCase):
Place.ensure_indexes()
info = Place._get_collection().index_information()
info = [value['key'] for key, value in info.iteritems()]
self.assertTrue([('location.point', '2d')] in info)
self.assertIn([('location.point', '2d')], info)
def test_explicit_geo2d_index_embedded(self):
"""Ensure that geo2d indexes work when created via meta[indexes]
@@ -269,7 +269,7 @@ class IndexesTest(unittest.TestCase):
Place.ensure_indexes()
info = Place._get_collection().index_information()
info = [value['key'] for key, value in info.iteritems()]
self.assertTrue([('current.location.point', '2d')] in info)
self.assertIn([('current.location.point', '2d')], info)
def test_explicit_geosphere_index(self):
"""Ensure that geosphere indexes work when created via meta[indexes]
@@ -289,7 +289,7 @@ class IndexesTest(unittest.TestCase):
Place.ensure_indexes()
info = Place._get_collection().index_information()
info = [value['key'] for key, value in info.iteritems()]
self.assertTrue([('location.point', '2dsphere')] in info)
self.assertIn([('location.point', '2dsphere')], info)
def test_explicit_geohaystack_index(self):
"""Ensure that geohaystack indexes work when created via meta[indexes]
@@ -311,7 +311,7 @@ class IndexesTest(unittest.TestCase):
Place.ensure_indexes()
info = Place._get_collection().index_information()
info = [value['key'] for key, value in info.iteritems()]
self.assertTrue([('location.point', 'geoHaystack')] in info)
self.assertIn([('location.point', 'geoHaystack')], info)
def test_create_geohaystack_index(self):
"""Ensure that geohaystack indexes can be created
@@ -323,7 +323,7 @@ class IndexesTest(unittest.TestCase):
Place.create_index({'fields': (')location.point', 'name')}, bucketSize=10)
info = Place._get_collection().index_information()
info = [value['key'] for key, value in info.iteritems()]
self.assertTrue([('location.point', 'geoHaystack'), ('name', 1)] in info)
self.assertIn([('location.point', 'geoHaystack'), ('name', 1)], info)
def test_dictionary_indexes(self):
"""Ensure that indexes are used when meta[indexes] contains
@@ -356,7 +356,7 @@ class IndexesTest(unittest.TestCase):
value.get('unique', False),
value.get('sparse', False))
for key, value in info.iteritems()]
self.assertTrue(([('addDate', -1)], True, True) in info)
self.assertIn(([('addDate', -1)], True, True), info)
BlogPost.drop_collection()
@@ -412,7 +412,6 @@ class IndexesTest(unittest.TestCase):
User.ensure_indexes()
info = User.objects._collection.index_information()
self.assertEqual(sorted(info.keys()), ['_cls_1_user_guid_1', '_id_'])
User.drop_collection()
def test_embedded_document_index(self):
"""Tests settings an index on an embedded document
@@ -434,7 +433,6 @@ class IndexesTest(unittest.TestCase):
info = BlogPost.objects._collection.index_information()
self.assertEqual(sorted(info.keys()), ['_id_', 'date.yr_-1'])
BlogPost.drop_collection()
def test_list_embedded_document_index(self):
"""Ensure list embedded documents can be indexed
@@ -461,7 +459,6 @@ class IndexesTest(unittest.TestCase):
post1 = BlogPost(title="Embedded Indexes tests in place",
tags=[Tag(name="about"), Tag(name="time")])
post1.save()
BlogPost.drop_collection()
def test_recursive_embedded_objects_dont_break_indexes(self):
@@ -494,8 +491,7 @@ class IndexesTest(unittest.TestCase):
obj = Test(a=1)
obj.save()
connection = get_connection()
IS_MONGODB_3 = connection.server_info()['versionArray'][0] >= 3
IS_MONGODB_3 = get_mongodb_version()[0] >= 3
# Need to be explicit about covered indexes as mongoDB doesn't know if
# the documents returned might have more keys in that here.
@@ -623,8 +619,6 @@ class IndexesTest(unittest.TestCase):
post3 = BlogPost(title='test3', date=Date(year=2010), slug='test')
self.assertRaises(OperationError, post3.save)
BlogPost.drop_collection()
def test_unique_embedded_document(self):
"""Ensure that uniqueness constraints are applied to fields on embedded documents.
"""
@@ -652,8 +646,6 @@ class IndexesTest(unittest.TestCase):
sub=SubDocument(year=2010, slug='test'))
self.assertRaises(NotUniqueError, post3.save)
BlogPost.drop_collection()
def test_unique_embedded_document_in_list(self):
"""
Ensure that the uniqueness constraints are applied to fields in
@@ -684,8 +676,6 @@ class IndexesTest(unittest.TestCase):
self.assertRaises(NotUniqueError, post2.save)
BlogPost.drop_collection()
def test_unique_with_embedded_document_and_embedded_unique(self):
"""Ensure that uniqueness constraints are applied to fields on
embedded documents. And work with unique_with as well.
@@ -719,8 +709,6 @@ class IndexesTest(unittest.TestCase):
sub=SubDocument(year=2009, slug='test-1'))
self.assertRaises(NotUniqueError, post3.save)
BlogPost.drop_collection()
def test_ttl_indexes(self):
class Log(Document):
@@ -733,14 +721,6 @@ class IndexesTest(unittest.TestCase):
Log.drop_collection()
if pymongo.version_tuple[0] < 2 and pymongo.version_tuple[1] < 3:
raise SkipTest('pymongo needs to be 2.3 or higher for this test')
connection = get_connection()
version_array = connection.server_info()['versionArray']
if version_array[0] < 2 and version_array[1] < 2:
raise SkipTest('MongoDB needs to be 2.2 or higher for this test')
# Indexes are lazy so use list() to perform query
list(Log.objects)
info = Log.objects._collection.index_information()
@@ -768,13 +748,11 @@ class IndexesTest(unittest.TestCase):
raise AssertionError("We saved a dupe!")
except NotUniqueError:
pass
Customer.drop_collection()
def test_unique_and_primary(self):
"""If you set a field as primary, then unexpected behaviour can occur.
You won't create a duplicate but you will update an existing document.
"""
class User(Document):
name = StringField(primary_key=True, unique=True)
password = StringField()
@@ -790,8 +768,23 @@ class IndexesTest(unittest.TestCase):
self.assertEqual(User.objects.count(), 1)
self.assertEqual(User.objects.get().password, 'secret2')
def test_unique_and_primary_create(self):
"""Create a new record with a duplicate primary key
throws an exception
"""
class User(Document):
name = StringField(primary_key=True)
password = StringField()
User.drop_collection()
User.objects.create(name='huangz', password='secret')
with self.assertRaises(NotUniqueError):
User.objects.create(name='huangz', password='secret2')
self.assertEqual(User.objects.count(), 1)
self.assertEqual(User.objects.get().password, 'secret')
def test_index_with_pk(self):
"""Ensure you can use `pk` as part of a query"""
@@ -810,7 +803,7 @@ class IndexesTest(unittest.TestCase):
info = BlogPost.objects._collection.index_information()
info = [value['key'] for key, value in info.iteritems()]
index_item = [('_id', 1), ('comments.comment_id', 1)]
self.assertTrue(index_item in info)
self.assertIn(index_item, info)
def test_compound_key_embedded(self):
@@ -857,8 +850,8 @@ class IndexesTest(unittest.TestCase):
info = MyDoc.objects._collection.index_information()
info = [value['key'] for key, value in info.iteritems()]
self.assertTrue([('provider_ids.foo', 1)] in info)
self.assertTrue([('provider_ids.bar', 1)] in info)
self.assertIn([('provider_ids.foo', 1)], info)
self.assertIn([('provider_ids.bar', 1)], info)
def test_sparse_compound_indexes(self):
@@ -874,8 +867,8 @@ class IndexesTest(unittest.TestCase):
info['provider_ids.foo_1_provider_ids.bar_1']['key'])
self.assertTrue(info['provider_ids.foo_1_provider_ids.bar_1']['sparse'])
@needs_mongodb_v26
def test_text_indexes(self):
class Book(Document):
title = DictField()
meta = {
@@ -883,9 +876,9 @@ class IndexesTest(unittest.TestCase):
}
indexes = Book.objects._collection.index_information()
self.assertTrue("title_text" in indexes)
self.assertIn("title_text", indexes)
key = indexes["title_text"]["key"]
self.assertTrue(('_fts', 'text') in key)
self.assertIn(('_fts', 'text'), key)
def test_hashed_indexes(self):
@@ -896,8 +889,8 @@ class IndexesTest(unittest.TestCase):
}
indexes = Book.objects._collection.index_information()
self.assertTrue("ref_id_hashed" in indexes)
self.assertTrue(('ref_id', 'hashed') in indexes["ref_id_hashed"]["key"])
self.assertIn("ref_id_hashed", indexes)
self.assertIn(('ref_id', 'hashed'), indexes["ref_id_hashed"]["key"])
def test_indexes_after_database_drop(self):
"""
@@ -1020,7 +1013,7 @@ class IndexesTest(unittest.TestCase):
TestDoc.ensure_indexes()
index_info = TestDoc._get_collection().index_information()
self.assertTrue('shard_1_1__cls_1_txt_1_1' in index_info)
self.assertIn('shard_1_1__cls_1_txt_1_1', index_info)
if __name__ == '__main__':

12
tests/document/inheritance.py
View File

@@ -268,7 +268,7 @@ class InheritanceTest(unittest.TestCase):
collection = self.db[Animal._get_collection_name()]
obj = collection.find_one()
self.assertFalse('_cls' in obj)
self.assertNotIn('_cls', obj)
def test_cant_turn_off_inheritance_on_subclass(self):
"""Ensure if inheritance is on in a subclass you cant turn it off.
@@ -298,7 +298,7 @@ class InheritanceTest(unittest.TestCase):
# Check that _cls isn't present in simple documents
doc = Animal(name='dog')
self.assertFalse('_cls' in doc.to_mongo())
self.assertNotIn('_cls', doc.to_mongo())
def test_abstract_handle_ids_in_metaclass_properly(self):
@@ -374,14 +374,14 @@ class InheritanceTest(unittest.TestCase):
pass
doc = Comment(content='test')
self.assertFalse('_cls' in doc.to_mongo())
self.assertNotIn('_cls', doc.to_mongo())
class Comment(EmbeddedDocument):
content = StringField()
meta = {'allow_inheritance': True}
doc = Comment(content='test')
self.assertTrue('_cls' in doc.to_mongo())
self.assertIn('_cls', doc.to_mongo())
def test_document_inheritance(self):
"""Ensure mutliple inheritance of abstract documents
@@ -434,8 +434,8 @@ class InheritanceTest(unittest.TestCase):
for cls in [Animal, Fish, Guppy]:
self.assertEqual(cls._meta[k], v)
self.assertFalse('collection' in Animal._meta)
self.assertFalse('collection' in Mammal._meta)
self.assertNotIn('collection', Animal._meta)
self.assertNotIn('collection', Mammal._meta)
self.assertEqual(Animal._get_collection_name(), None)
self.assertEqual(Mammal._get_collection_name(), None)

619
tests/document/instance.py
View File

File diff suppressed because it is too large Load Diff

28
tests/document/validation.py
View File

@@ -20,16 +20,16 @@ class ValidatorErrorTest(unittest.TestCase):
# 1st level error schema
error.errors = {'1st': ValidationError('bad 1st'), }
self.assertTrue('1st' in error.to_dict())
self.assertIn('1st', error.to_dict())
self.assertEqual(error.to_dict()['1st'], 'bad 1st')
# 2nd level error schema
error.errors = {'1st': ValidationError('bad 1st', errors={
'2nd': ValidationError('bad 2nd'),
})}
self.assertTrue('1st' in error.to_dict())
self.assertTrue(isinstance(error.to_dict()['1st'], dict))
self.assertTrue('2nd' in error.to_dict()['1st'])
self.assertIn('1st', error.to_dict())
self.assertIsInstance(error.to_dict()['1st'], dict)
self.assertIn('2nd', error.to_dict()['1st'])
self.assertEqual(error.to_dict()['1st']['2nd'], 'bad 2nd')
# moar levels
@@ -40,10 +40,10 @@ class ValidatorErrorTest(unittest.TestCase):
}),
}),
})}
self.assertTrue('1st' in error.to_dict())
self.assertTrue('2nd' in error.to_dict()['1st'])
self.assertTrue('3rd' in error.to_dict()['1st']['2nd'])
self.assertTrue('4th' in error.to_dict()['1st']['2nd']['3rd'])
self.assertIn('1st', error.to_dict())
self.assertIn('2nd', error.to_dict()['1st'])
self.assertIn('3rd', error.to_dict()['1st']['2nd'])
self.assertIn('4th', error.to_dict()['1st']['2nd']['3rd'])
self.assertEqual(error.to_dict()['1st']['2nd']['3rd']['4th'],
'Inception')
@@ -58,7 +58,7 @@ class ValidatorErrorTest(unittest.TestCase):
try:
User().validate()
except ValidationError as e:
self.assertTrue("User:None" in e.message)
self.assertIn("User:None", e.message)
self.assertEqual(e.to_dict(), {
'username': 'Field is required',
'name': 'Field is required'})
@@ -68,7 +68,7 @@ class ValidatorErrorTest(unittest.TestCase):
try:
user.save()
except ValidationError as e:
self.assertTrue("User:RossC0" in e.message)
self.assertIn("User:RossC0", e.message)
self.assertEqual(e.to_dict(), {
'name': 'Field is required'})
@@ -116,7 +116,7 @@ class ValidatorErrorTest(unittest.TestCase):
try:
Doc(id="bad").validate()
except ValidationError as e:
self.assertTrue("SubDoc:None" in e.message)
self.assertIn("SubDoc:None", e.message)
self.assertEqual(e.to_dict(), {
"e": {'val': 'OK could not be converted to int'}})
@@ -127,14 +127,14 @@ class ValidatorErrorTest(unittest.TestCase):
doc = Doc.objects.first()
keys = doc._data.keys()
self.assertEqual(2, len(keys))
self.assertTrue('e' in keys)
self.assertTrue('id' in keys)
self.assertIn('e', keys)
self.assertIn('id', keys)
doc.e.val = "OK"
try:
doc.save()
except ValidationError as e:
self.assertTrue("Doc:test" in e.message)
self.assertIn("Doc:test", e.message)
self.assertEqual(e.to_dict(), {
"e": {'val': 'OK could not be converted to int'}})

6
tests/fields/__init__.py
View File

@@ -1,3 +1,3 @@
from fields import *
from file_tests import *
from geo import *
from .fields import *
from .file_tests import *
from .geo import *

2858
tests/fields/fields.py
View File

File diff suppressed because it is too large Load Diff

28
tests/fields/file_tests.py
View File

@@ -18,15 +18,13 @@ try:
except ImportError:
HAS_PIL = False
from tests.utils import MongoDBTestCase
TEST_IMAGE_PATH = os.path.join(os.path.dirname(__file__), 'mongoengine.png')
TEST_IMAGE2_PATH = os.path.join(os.path.dirname(__file__), 'mongodb_leaf.png')
class FileTest(unittest.TestCase):
def setUp(self):
connect(db='mongoenginetest')
self.db = get_db()
class FileTest(MongoDBTestCase):
def tearDown(self):
self.db.drop_collection('fs.files')
@@ -55,8 +53,8 @@ class FileTest(unittest.TestCase):
putfile.save()
result = PutFile.objects.first()
self.assertTrue(putfile == result)
self.assertEqual("%s" % result.the_file, "<GridFSProxy: hello>")
self.assertEqual(putfile, result)
self.assertEqual("%s" % result.the_file, "<GridFSProxy: hello (%s)>" % result.the_file.grid_id)
self.assertEqual(result.the_file.read(), text)
self.assertEqual(result.the_file.content_type, content_type)
result.the_file.delete() # Remove file from GridFS
@@ -73,7 +71,7 @@ class FileTest(unittest.TestCase):
putfile.save()
result = PutFile.objects.first()
self.assertTrue(putfile == result)
self.assertEqual(putfile, result)
self.assertEqual(result.the_file.read(), text)
self.assertEqual(result.the_file.content_type, content_type)
result.the_file.delete()
@@ -98,7 +96,7 @@ class FileTest(unittest.TestCase):
streamfile.save()
result = StreamFile.objects.first()
self.assertTrue(streamfile == result)
self.assertEqual(streamfile, result)
self.assertEqual(result.the_file.read(), text + more_text)
self.assertEqual(result.the_file.content_type, content_type)
result.the_file.seek(0)
@@ -134,7 +132,7 @@ class FileTest(unittest.TestCase):
streamfile.save()
result = StreamFile.objects.first()
self.assertTrue(streamfile == result)
self.assertEqual(streamfile, result)
self.assertEqual(result.the_file.read(), text + more_text)
# self.assertEqual(result.the_file.content_type, content_type)
result.the_file.seek(0)
@@ -163,7 +161,7 @@ class FileTest(unittest.TestCase):
setfile.save()
result = SetFile.objects.first()
self.assertTrue(setfile == result)
self.assertEqual(setfile, result)
self.assertEqual(result.the_file.read(), text)
# Try replacing file with new one
@@ -171,7 +169,7 @@ class FileTest(unittest.TestCase):
result.save()
result = SetFile.objects.first()
self.assertTrue(setfile == result)
self.assertEqual(setfile, result)
self.assertEqual(result.the_file.read(), more_text)
result.the_file.delete()
@@ -233,8 +231,8 @@ class FileTest(unittest.TestCase):
test_file_dupe = TestFile()
data = test_file_dupe.the_file.read() # Should be None
self.assertTrue(test_file.name != test_file_dupe.name)
self.assertTrue(test_file.the_file.read() != data)
self.assertNotEqual(test_file.name, test_file_dupe.name)
self.assertNotEqual(test_file.the_file.read(), data)
TestFile.drop_collection()
@@ -293,7 +291,7 @@ class FileTest(unittest.TestCase):
the_file = FileField()
test_file = TestFile()
self.assertFalse(test_file.the_file in [{"test": 1}])
self.assertNotIn(test_file.the_file, [{"test": 1}])
def test_file_disk_space(self):
""" Test disk space usage when we delete/replace a file """

16
tests/fields/geo.py
View File

@@ -298,9 +298,9 @@ class GeoFieldTest(unittest.TestCase):
polygon = PolygonField()
geo_indicies = Event._geo_indices()
self.assertTrue({'fields': [('line', '2dsphere')]} in geo_indicies)
self.assertTrue({'fields': [('polygon', '2dsphere')]} in geo_indicies)
self.assertTrue({'fields': [('point', '2dsphere')]} in geo_indicies)
self.assertIn({'fields': [('line', '2dsphere')]}, geo_indicies)
self.assertIn({'fields': [('polygon', '2dsphere')]}, geo_indicies)
self.assertIn({'fields': [('point', '2dsphere')]}, geo_indicies)
def test_indexes_2dsphere_embedded(self):
"""Ensure that indexes are created automatically for GeoPointFields.
@@ -316,9 +316,9 @@ class GeoFieldTest(unittest.TestCase):
venue = EmbeddedDocumentField(Venue)
geo_indicies = Event._geo_indices()
self.assertTrue({'fields': [('venue.line', '2dsphere')]} in geo_indicies)
self.assertTrue({'fields': [('venue.polygon', '2dsphere')]} in geo_indicies)
self.assertTrue({'fields': [('venue.point', '2dsphere')]} in geo_indicies)
self.assertIn({'fields': [('venue.line', '2dsphere')]}, geo_indicies)
self.assertIn({'fields': [('venue.polygon', '2dsphere')]}, geo_indicies)
self.assertIn({'fields': [('venue.point', '2dsphere')]}, geo_indicies)
def test_geo_indexes_recursion(self):
@@ -335,9 +335,9 @@ class GeoFieldTest(unittest.TestCase):
Parent(name='Berlin').save()
info = Parent._get_collection().index_information()
self.assertFalse('location_2d' in info)
self.assertNotIn('location_2d', info)
info = Location._get_collection().index_information()
self.assertTrue('location_2d' in info)
self.assertIn('location_2d', info)
self.assertEqual(len(Parent._geo_indices()), 0)
self.assertEqual(len(Location._geo_indices()), 1)

12
tests/queryset/__init__.py
View File

@@ -1,6 +1,6 @@
from transform import *
from field_list import *
from queryset import *
from visitor import *
from geo import *
from modify import *
from .transform import *
from .field_list import *
from .queryset import *
from .visitor import *
from .geo import *
from .modify import *

21
tests/queryset/field_list.py
View File

@@ -141,6 +141,16 @@ class OnlyExcludeAllTest(unittest.TestCase):
self.assertEqual(qs._loaded_fields.as_dict(),
{'b': {'$slice': 5}})
def test_mix_slice_with_other_fields(self):
class MyDoc(Document):
a = ListField()
b = ListField()
c = ListField()
qs = MyDoc.objects.fields(a=1, b=0, slice__c=2)
self.assertEqual(qs._loaded_fields.as_dict(),
{'c': {'$slice': 2}, 'a': 1})
def test_only(self):
"""Ensure that QuerySet.only only returns the requested fields.
"""
@@ -171,7 +181,7 @@ class OnlyExcludeAllTest(unittest.TestCase):
employee.save()
obj = self.Person.objects(id=employee.id).only('age').get()
self.assertTrue(isinstance(obj, Employee))
self.assertIsInstance(obj, Employee)
# Check field names are looked up properly
obj = Employee.objects(id=employee.id).only('salary').get()
@@ -187,14 +197,18 @@ class OnlyExcludeAllTest(unittest.TestCase):
title = StringField()
text = StringField()
class VariousData(EmbeddedDocument):
some = BooleanField()
class BlogPost(Document):
content = StringField()
author = EmbeddedDocumentField(User)
comments = ListField(EmbeddedDocumentField(Comment))
various = MapField(field=EmbeddedDocumentField(VariousData))
BlogPost.drop_collection()
post = BlogPost(content='Had a good coffee today...')
post = BlogPost(content='Had a good coffee today...', various={'test_dynamic':{'some': True}})
post.author = User(name='Test User')
post.comments = [Comment(title='I aggree', text='Great post!'), Comment(title='Coffee', text='I hate coffee')]
post.save()
@@ -205,6 +219,9 @@ class OnlyExcludeAllTest(unittest.TestCase):
self.assertEqual(obj.author.name, 'Test User')
self.assertEqual(obj.comments, [])
obj = BlogPost.objects.only('various.test_dynamic.some').get()
self.assertEqual(obj.various["test_dynamic"].some, True)
obj = BlogPost.objects.only('content', 'comments.title',).get()
self.assertEqual(obj.content, 'Had a good coffee today...')
self.assertEqual(obj.author, None)

460
tests/queryset/geo.py
View File

@@ -1,105 +1,139 @@
from datetime import datetime, timedelta
import datetime
import unittest
from pymongo.errors import OperationFailure
from mongoengine import *
from mongoengine.connection import get_connection
from nose.plugins.skip import SkipTest
from tests.utils import MongoDBTestCase, needs_mongodb_v3
__all__ = ("GeoQueriesTest",)
class GeoQueriesTest(unittest.TestCase):
class GeoQueriesTest(MongoDBTestCase):
def setUp(self):
connect(db='mongoenginetest')
def test_geospatial_operators(self):
"""Ensure that geospatial queries are working.
"""
def _create_event_data(self, point_field_class=GeoPointField):
"""Create some sample data re-used in many of the tests below."""
class Event(Document):
title = StringField()
date = DateTimeField()
location = GeoPointField()
location = point_field_class()
def __unicode__(self):
return self.title
self.Event = Event
Event.drop_collection()
event1 = Event(title="Coltrane Motion @ Double Door",
date=datetime.now() - timedelta(days=1),
location=[-87.677137, 41.909889]).save()
event2 = Event(title="Coltrane Motion @ Bottom of the Hill",
date=datetime.now() - timedelta(days=10),
location=[-122.4194155, 37.7749295]).save()
event3 = Event(title="Coltrane Motion @ Empty Bottle",
date=datetime.now(),
location=[-87.686638, 41.900474]).save()
event1 = Event.objects.create(
title="Coltrane Motion @ Double Door",
date=datetime.datetime.now() - datetime.timedelta(days=1),
location=[-87.677137, 41.909889])
event2 = Event.objects.create(
title="Coltrane Motion @ Bottom of the Hill",
date=datetime.datetime.now() - datetime.timedelta(days=10),
location=[-122.4194155, 37.7749295])
event3 = Event.objects.create(
title="Coltrane Motion @ Empty Bottle",
date=datetime.datetime.now(),
location=[-87.686638, 41.900474])
return event1, event2, event3
def test_near(self):
"""Make sure the "near" operator works."""
event1, event2, event3 = self._create_event_data()
# find all events "near" pitchfork office, chicago.
# note that "near" will show the san francisco event, too,
# although it sorts to last.
events = Event.objects(location__near=[-87.67892, 41.9120459])
events = self.Event.objects(location__near=[-87.67892, 41.9120459])
self.assertEqual(events.count(), 3)
self.assertEqual(list(events), [event1, event3, event2])
# find events within 5 degrees of pitchfork office, chicago
point_and_distance = [[-87.67892, 41.9120459], 5]
events = Event.objects(location__within_distance=point_and_distance)
self.assertEqual(events.count(), 2)
events = list(events)
self.assertTrue(event2 not in events)
self.assertTrue(event1 in events)
self.assertTrue(event3 in events)
# ensure ordering is respected by "near"
events = Event.objects(location__near=[-87.67892, 41.9120459])
events = self.Event.objects(location__near=[-87.67892, 41.9120459])
events = events.order_by("-date")
self.assertEqual(events.count(), 3)
self.assertEqual(list(events), [event3, event1, event2])
def test_near_and_max_distance(self):
"""Ensure the "max_distance" operator works alongside the "near"
operator.
"""
event1, event2, event3 = self._create_event_data()
# find events within 10 degrees of san francisco
point = [-122.415579, 37.7566023]
events = Event.objects(location__near=point, location__max_distance=10)
events = self.Event.objects(location__near=point,
location__max_distance=10)
self.assertEqual(events.count(), 1)
self.assertEqual(events[0], event2)
# $minDistance was added in MongoDB v2.6, but continued being buggy
# until v3.0; skip for older versions
@needs_mongodb_v3
def test_near_and_min_distance(self):
"""Ensure the "min_distance" operator works alongside the "near"
operator.
"""
event1, event2, event3 = self._create_event_data()
# find events at least 10 degrees away of san francisco
point = [-122.415579, 37.7566023]
events = Event.objects(location__near=point, location__min_distance=10)
# The following real test passes on MongoDB 3 but minDistance seems
# buggy on older MongoDB versions
if get_connection().server_info()['versionArray'][0] > 2:
self.assertEqual(events.count(), 2)
else:
self.assertTrue(events.count() >= 2)
events = self.Event.objects(location__near=point,
location__min_distance=10)
self.assertEqual(events.count(), 2)
def test_within_distance(self):
"""Make sure the "within_distance" operator works."""
event1, event2, event3 = self._create_event_data()
# find events within 5 degrees of pitchfork office, chicago
point_and_distance = [[-87.67892, 41.9120459], 5]
events = self.Event.objects(
location__within_distance=point_and_distance)
self.assertEqual(events.count(), 2)
events = list(events)
self.assertNotIn(event2, events)
self.assertIn(event1, events)
self.assertIn(event3, events)
# find events within 10 degrees of san francisco
point_and_distance = [[-122.415579, 37.7566023], 10]
events = Event.objects(location__within_distance=point_and_distance)
events = self.Event.objects(
location__within_distance=point_and_distance)
self.assertEqual(events.count(), 1)
self.assertEqual(events[0], event2)
# find events within 1 degree of greenpoint, broolyn, nyc, ny
point_and_distance = [[-73.9509714, 40.7237134], 1]
events = Event.objects(location__within_distance=point_and_distance)
events = self.Event.objects(
location__within_distance=point_and_distance)
self.assertEqual(events.count(), 0)
# ensure ordering is respected by "within_distance"
point_and_distance = [[-87.67892, 41.9120459], 10]
events = Event.objects(location__within_distance=point_and_distance)
events = self.Event.objects(
location__within_distance=point_and_distance)
events = events.order_by("-date")
self.assertEqual(events.count(), 2)
self.assertEqual(events[0], event3)
def test_within_box(self):
"""Ensure the "within_box" operator works."""
event1, event2, event3 = self._create_event_data()
# check that within_box works
box = [(-125.0, 35.0), (-100.0, 40.0)]
events = Event.objects(location__within_box=box)
events = self.Event.objects(location__within_box=box)
self.assertEqual(events.count(), 1)
self.assertEqual(events[0].id, event2.id)
def test_within_polygon(self):
"""Ensure the "within_polygon" operator works."""
event1, event2, event3 = self._create_event_data()
polygon = [
(-87.694445, 41.912114),
(-87.69084, 41.919395),
@@ -107,7 +141,7 @@ class GeoQueriesTest(unittest.TestCase):
(-87.654276, 41.911731),
(-87.656164, 41.898061),
]
events = Event.objects(location__within_polygon=polygon)
events = self.Event.objects(location__within_polygon=polygon)
self.assertEqual(events.count(), 1)
self.assertEqual(events[0].id, event1.id)
@@ -116,13 +150,151 @@ class GeoQueriesTest(unittest.TestCase):
(-1.225891, 52.792797),
(-4.40094, 53.389881)
]
events = Event.objects(location__within_polygon=polygon2)
events = self.Event.objects(location__within_polygon=polygon2)
self.assertEqual(events.count(), 0)
def test_geo_spatial_embedded(self):
def test_2dsphere_near(self):
"""Make sure the "near" operator works with a PointField, which
corresponds to a 2dsphere index.
"""
event1, event2, event3 = self._create_event_data(
point_field_class=PointField
)
# find all events "near" pitchfork office, chicago.
# note that "near" will show the san francisco event, too,
# although it sorts to last.
events = self.Event.objects(location__near=[-87.67892, 41.9120459])
self.assertEqual(events.count(), 3)
self.assertEqual(list(events), [event1, event3, event2])
# ensure ordering is respected by "near"
events = self.Event.objects(location__near=[-87.67892, 41.9120459])
events = events.order_by("-date")
self.assertEqual(events.count(), 3)
self.assertEqual(list(events), [event3, event1, event2])
def test_2dsphere_near_and_max_distance(self):
"""Ensure the "max_distance" operator works alongside the "near"
operator with a 2dsphere index.
"""
event1, event2, event3 = self._create_event_data(
point_field_class=PointField
)
# find events within 10km of san francisco
point = [-122.415579, 37.7566023]
events = self.Event.objects(location__near=point,
location__max_distance=10000)
self.assertEqual(events.count(), 1)
self.assertEqual(events[0], event2)
# find events within 1km of greenpoint, broolyn, nyc, ny
events = self.Event.objects(location__near=[-73.9509714, 40.7237134],
location__max_distance=1000)
self.assertEqual(events.count(), 0)
# ensure ordering is respected by "near"
events = self.Event.objects(
location__near=[-87.67892, 41.9120459],
location__max_distance=10000
).order_by("-date")
self.assertEqual(events.count(), 2)
self.assertEqual(events[0], event3)
def test_2dsphere_geo_within_box(self):
"""Ensure the "geo_within_box" operator works with a 2dsphere
index.
"""
event1, event2, event3 = self._create_event_data(
point_field_class=PointField
)
# check that within_box works
box = [(-125.0, 35.0), (-100.0, 40.0)]
events = self.Event.objects(location__geo_within_box=box)
self.assertEqual(events.count(), 1)
self.assertEqual(events[0].id, event2.id)
def test_2dsphere_geo_within_polygon(self):
"""Ensure the "geo_within_polygon" operator works with a
2dsphere index.
"""
event1, event2, event3 = self._create_event_data(
point_field_class=PointField
)
polygon = [
(-87.694445, 41.912114),
(-87.69084, 41.919395),
(-87.681742, 41.927186),
(-87.654276, 41.911731),
(-87.656164, 41.898061),
]
events = self.Event.objects(location__geo_within_polygon=polygon)
self.assertEqual(events.count(), 1)
self.assertEqual(events[0].id, event1.id)
polygon2 = [
(-1.742249, 54.033586),
(-1.225891, 52.792797),
(-4.40094, 53.389881)
]
events = self.Event.objects(location__geo_within_polygon=polygon2)
self.assertEqual(events.count(), 0)
# $minDistance was added in MongoDB v2.6, but continued being buggy
# until v3.0; skip for older versions
@needs_mongodb_v3
def test_2dsphere_near_and_min_max_distance(self):
"""Ensure "min_distace" and "max_distance" operators work well
together with the "near" operator in a 2dsphere index.
"""
event1, event2, event3 = self._create_event_data(
point_field_class=PointField
)
# ensure min_distance and max_distance combine well
events = self.Event.objects(
location__near=[-87.67892, 41.9120459],
location__min_distance=1000,
location__max_distance=10000
).order_by("-date")
self.assertEqual(events.count(), 1)
self.assertEqual(events[0], event3)
# ensure ordering is respected by "near" with "min_distance"
events = self.Event.objects(
location__near=[-87.67892, 41.9120459],
location__min_distance=10000
).order_by("-date")
self.assertEqual(events.count(), 1)
self.assertEqual(events[0], event2)
def test_2dsphere_geo_within_center(self):
"""Make sure the "geo_within_center" operator works with a
2dsphere index.
"""
event1, event2, event3 = self._create_event_data(
point_field_class=PointField
)
# find events within 5 degrees of pitchfork office, chicago
point_and_distance = [[-87.67892, 41.9120459], 2]
events = self.Event.objects(
location__geo_within_center=point_and_distance)
self.assertEqual(events.count(), 2)
events = list(events)
self.assertNotIn(event2, events)
self.assertIn(event1, events)
self.assertIn(event3, events)
def _test_embedded(self, point_field_class):
"""Helper test method ensuring given point field class works
well in an embedded document.
"""
class Venue(EmbeddedDocument):
location = GeoPointField()
location = point_field_class()
name = StringField()
class Event(Document):
@@ -148,16 +320,18 @@ class GeoQueriesTest(unittest.TestCase):
self.assertEqual(events.count(), 3)
self.assertEqual(list(events), [event1, event3, event2])
def test_spherical_geospatial_operators(self):
"""Ensure that spherical geospatial queries are working
"""
# Needs MongoDB > 2.6.4 https://jira.mongodb.org/browse/SERVER-14039
connection = get_connection()
info = connection.test.command('buildInfo')
mongodb_version = tuple([int(i) for i in info['version'].split('.')])
if mongodb_version < (2, 6, 4):
raise SkipTest("Need MongoDB version 2.6.4+")
def test_geo_spatial_embedded(self):
"""Make sure GeoPointField works properly in an embedded document."""
self._test_embedded(point_field_class=GeoPointField)
def test_2dsphere_point_embedded(self):
"""Make sure PointField works properly in an embedded document."""
self._test_embedded(point_field_class=PointField)
# Needs MongoDB > 2.6.4 https://jira.mongodb.org/browse/SERVER-14039
@needs_mongodb_v3
def test_spherical_geospatial_operators(self):
"""Ensure that spherical geospatial queries are working."""
class Point(Document):
location = GeoPointField()
@@ -177,7 +351,10 @@ class GeoQueriesTest(unittest.TestCase):
# Same behavior for _within_spherical_distance
points = Point.objects(
location__within_spherical_distance=[[-122, 37.5], 60 / earth_radius]
location__within_spherical_distance=[
[-122, 37.5],
60 / earth_radius
]
)
self.assertEqual(points.count(), 2)
@@ -194,14 +371,9 @@ class GeoQueriesTest(unittest.TestCase):
# Test query works with min_distance, being farer from one point
points = Point.objects(location__near_sphere=[-122, 37.8],
location__min_distance=60 / earth_radius)
# The following real test passes on MongoDB 3 but minDistance seems
# buggy on older MongoDB versions
if get_connection().server_info()['versionArray'][0] > 2:
self.assertEqual(points.count(), 1)
far_point = points.first()
self.assertNotEqual(close_point, far_point)
else:
self.assertTrue(points.count() >= 1)
self.assertEqual(points.count(), 1)
far_point = points.first()
self.assertNotEqual(close_point, far_point)
# Finds both points, but orders the north point first because it's
# closer to the reference point to the north.
@@ -220,141 +392,15 @@ class GeoQueriesTest(unittest.TestCase):
# Finds only one point because only the first point is within 60km of
# the reference point to the south.
points = Point.objects(
location__within_spherical_distance=[[-122, 36.5], 60/earth_radius])
location__within_spherical_distance=[
[-122, 36.5],
60 / earth_radius
]
)
self.assertEqual(points.count(), 1)
self.assertEqual(points[0].id, south_point.id)
def test_2dsphere_point(self):
class Event(Document):
title = StringField()
date = DateTimeField()
location = PointField()
def __unicode__(self):
return self.title
Event.drop_collection()
event1 = Event(title="Coltrane Motion @ Double Door",
date=datetime.now() - timedelta(days=1),
location=[-87.677137, 41.909889])
event1.save()
event2 = Event(title="Coltrane Motion @ Bottom of the Hill",
date=datetime.now() - timedelta(days=10),
location=[-122.4194155, 37.7749295]).save()
event3 = Event(title="Coltrane Motion @ Empty Bottle",
date=datetime.now(),
location=[-87.686638, 41.900474]).save()
# find all events "near" pitchfork office, chicago.
# note that "near" will show the san francisco event, too,
# although it sorts to last.
events = Event.objects(location__near=[-87.67892, 41.9120459])
self.assertEqual(events.count(), 3)
self.assertEqual(list(events), [event1, event3, event2])
# find events within 5 degrees of pitchfork office, chicago
point_and_distance = [[-87.67892, 41.9120459], 2]
events = Event.objects(location__geo_within_center=point_and_distance)
self.assertEqual(events.count(), 2)
events = list(events)
self.assertTrue(event2 not in events)
self.assertTrue(event1 in events)
self.assertTrue(event3 in events)
# ensure ordering is respected by "near"
events = Event.objects(location__near=[-87.67892, 41.9120459])
events = events.order_by("-date")
self.assertEqual(events.count(), 3)
self.assertEqual(list(events), [event3, event1, event2])
# find events within 10km of san francisco
point = [-122.415579, 37.7566023]
events = Event.objects(location__near=point, location__max_distance=10000)
self.assertEqual(events.count(), 1)
self.assertEqual(events[0], event2)
# find events within 1km of greenpoint, broolyn, nyc, ny
events = Event.objects(location__near=[-73.9509714, 40.7237134], location__max_distance=1000)
self.assertEqual(events.count(), 0)
# ensure ordering is respected by "near"
events = Event.objects(location__near=[-87.67892, 41.9120459],
location__max_distance=10000).order_by("-date")
self.assertEqual(events.count(), 2)
self.assertEqual(events[0], event3)
# ensure min_distance and max_distance combine well
events = Event.objects(location__near=[-87.67892, 41.9120459],
location__min_distance=1000,
location__max_distance=10000).order_by("-date")
self.assertEqual(events.count(), 1)
self.assertEqual(events[0], event3)
# ensure ordering is respected by "near"
events = Event.objects(location__near=[-87.67892, 41.9120459],
# location__min_distance=10000
location__min_distance=10000).order_by("-date")
self.assertEqual(events.count(), 1)
self.assertEqual(events[0], event2)
# check that within_box works
box = [(-125.0, 35.0), (-100.0, 40.0)]
events = Event.objects(location__geo_within_box=box)
self.assertEqual(events.count(), 1)
self.assertEqual(events[0].id, event2.id)
polygon = [
(-87.694445, 41.912114),
(-87.69084, 41.919395),
(-87.681742, 41.927186),
(-87.654276, 41.911731),
(-87.656164, 41.898061),
]
events = Event.objects(location__geo_within_polygon=polygon)
self.assertEqual(events.count(), 1)
self.assertEqual(events[0].id, event1.id)
polygon2 = [
(-1.742249, 54.033586),
(-1.225891, 52.792797),
(-4.40094, 53.389881)
]
events = Event.objects(location__geo_within_polygon=polygon2)
self.assertEqual(events.count(), 0)
def test_2dsphere_point_embedded(self):
class Venue(EmbeddedDocument):
location = GeoPointField()
name = StringField()
class Event(Document):
title = StringField()
venue = EmbeddedDocumentField(Venue)
Event.drop_collection()
venue1 = Venue(name="The Rock", location=[-87.677137, 41.909889])
venue2 = Venue(name="The Bridge", location=[-122.4194155, 37.7749295])
event1 = Event(title="Coltrane Motion @ Double Door",
venue=venue1).save()
event2 = Event(title="Coltrane Motion @ Bottom of the Hill",
venue=venue2).save()
event3 = Event(title="Coltrane Motion @ Empty Bottle",
venue=venue1).save()
# find all events "near" pitchfork office, chicago.
# note that "near" will show the san francisco event, too,
# although it sorts to last.
events = Event.objects(venue__location__near=[-87.67892, 41.9120459])
self.assertEqual(events.count(), 3)
self.assertEqual(list(events), [event1, event3, event2])
def test_linestring(self):
class Road(Document):
name = StringField()
line = LineStringField()
@@ -410,7 +456,6 @@ class GeoQueriesTest(unittest.TestCase):
self.assertEqual(1, roads)
def test_polygon(self):
class Road(Document):
name = StringField()
poly = PolygonField()
@@ -465,6 +510,24 @@ class GeoQueriesTest(unittest.TestCase):
roads = Road.objects.filter(poly__geo_intersects={"$geometry": polygon}).count()
self.assertEqual(1, roads)
def test_aspymongo_with_only(self):
"""Ensure as_pymongo works with only"""
class Place(Document):
location = PointField()
Place.drop_collection()
p = Place(location=[24.946861267089844, 60.16311983618494])
p.save()
qs = Place.objects().only('location')
self.assertDictEqual(
qs.as_pymongo()[0]['location'],
{u'type': u'Point',
u'coordinates': [
24.946861267089844,
60.16311983618494]
}
)
def test_2dsphere_point_sets_correctly(self):
class Location(Document):
loc = PointField()
@@ -507,5 +570,6 @@ class GeoQueriesTest(unittest.TestCase):
loc = Location.objects.as_pymongo()[0]
self.assertEqual(loc["poly"], {"type": "Polygon", "coordinates": [[[40, 4], [40, 6], [41, 6], [40, 4]]]})
if __name__ == '__main__':
unittest.main()

35
tests/queryset/modify.py
View File

@@ -1,6 +1,8 @@
import unittest
from mongoengine import connect, Document, IntField
from mongoengine import connect, Document, IntField, StringField, ListField
from tests.utils import needs_mongodb_v26
__all__ = ("FindAndModifyTest",)
@@ -94,6 +96,37 @@ class FindAndModifyTest(unittest.TestCase):
self.assertEqual(old_doc.to_mongo(), {"_id": 1})
self.assertDbEqual([{"_id": 0, "value": 0}, {"_id": 1, "value": -1}])
@needs_mongodb_v26
def test_modify_with_push(self):
class BlogPost(Document):
tags = ListField(StringField())
BlogPost.drop_collection()
blog = BlogPost.objects.create()
# Push a new tag via modify with new=False (default).
BlogPost(id=blog.id).modify(push__tags='code')
self.assertEqual(blog.tags, [])
blog.reload()
self.assertEqual(blog.tags, ['code'])
# Push a new tag via modify with new=True.
blog = BlogPost.objects(id=blog.id).modify(push__tags='java', new=True)
self.assertEqual(blog.tags, ['code', 'java'])
# Push a new tag with a positional argument.
blog = BlogPost.objects(id=blog.id).modify(
push__tags__0='python',
new=True)
self.assertEqual(blog.tags, ['python', 'code', 'java'])
# Push multiple new tags with a positional argument.
blog = BlogPost.objects(id=blog.id).modify(
push__tags__1=['go', 'rust'],
new=True)
self.assertEqual(blog.tags, ['python', 'go', 'rust', 'code', 'java'])
if __name__ == '__main__':
unittest.main()

850
tests/queryset/queryset.py
View File

File diff suppressed because it is too large Load Diff

60
tests/queryset/transform.py
View File

@@ -1,5 +1,7 @@
import unittest
from bson.son import SON
from mongoengine import *
from mongoengine.queryset import Q, transform
@@ -28,12 +30,16 @@ class TransformTest(unittest.TestCase):
{'name': {'$exists': True}})
def test_transform_update(self):
class LisDoc(Document):
foo = ListField(StringField())
class DicDoc(Document):
dictField = DictField()
class Doc(Document):
pass
LisDoc.drop_collection()
DicDoc.drop_collection()
Doc.drop_collection()
@@ -42,14 +48,28 @@ class TransformTest(unittest.TestCase):
for k, v in (("set", "$set"), ("set_on_insert", "$setOnInsert"), ("push", "$push")):
update = transform.update(DicDoc, **{"%s__dictField__test" % k: doc})
self.assertTrue(isinstance(update[v]["dictField.test"], dict))
self.assertIsInstance(update[v]["dictField.test"], dict)
# Update special cases
update = transform.update(DicDoc, unset__dictField__test=doc)
self.assertEqual(update["$unset"]["dictField.test"], 1)
update = transform.update(DicDoc, pull__dictField__test=doc)
self.assertTrue(isinstance(update["$pull"]["dictField"]["test"], dict))
self.assertIsInstance(update["$pull"]["dictField"]["test"], dict)
update = transform.update(LisDoc, pull__foo__in=['a'])
self.assertEqual(update, {'$pull': {'foo': {'$in': ['a']}}})
def test_transform_update_push(self):
"""Ensure the differences in behvaior between 'push' and 'push_all'"""
class BlogPost(Document):
tags = ListField(StringField())
update = transform.update(BlogPost, push__tags=['mongo', 'db'])
self.assertEqual(update, {'$push': {'tags': ['mongo', 'db']}})
update = transform.update(BlogPost, push_all__tags=['mongo', 'db'])
self.assertEqual(update, {'$push': {'tags': {'$each': ['mongo', 'db']}}})
def test_query_field_name(self):
"""Ensure that the correct field name is used when querying.
@@ -68,17 +88,15 @@ class TransformTest(unittest.TestCase):
post = BlogPost(**data)
post.save()
self.assertTrue('postTitle' in
BlogPost.objects(title=data['title'])._query)
self.assertIn('postTitle', BlogPost.objects(title=data['title'])._query)
self.assertFalse('title' in
BlogPost.objects(title=data['title'])._query)
self.assertEqual(BlogPost.objects(title=data['title']).count(), 1)
self.assertTrue('_id' in BlogPost.objects(pk=post.id)._query)
self.assertIn('_id', BlogPost.objects(pk=post.id)._query)
self.assertEqual(BlogPost.objects(pk=post.id).count(), 1)
self.assertTrue('postComments.commentContent' in
BlogPost.objects(comments__content='test')._query)
self.assertIn('postComments.commentContent', BlogPost.objects(comments__content='test')._query)
self.assertEqual(BlogPost.objects(comments__content='test').count(), 1)
BlogPost.drop_collection()
@@ -96,8 +114,8 @@ class TransformTest(unittest.TestCase):
post = BlogPost(**data)
post.save()
self.assertTrue('_id' in BlogPost.objects(pk=data['title'])._query)
self.assertTrue('_id' in BlogPost.objects(title=data['title'])._query)
self.assertIn('_id', BlogPost.objects(pk=data['title'])._query)
self.assertIn('_id', BlogPost.objects(title=data['title'])._query)
self.assertEqual(BlogPost.objects(pk=data['title']).count(), 1)
BlogPost.drop_collection()
@@ -241,6 +259,30 @@ class TransformTest(unittest.TestCase):
with self.assertRaises(InvalidQueryError):
events.count()
def test_update_pull_for_list_fields(self):
"""
Test added to check pull operation in update for
EmbeddedDocumentListField which is inside a EmbeddedDocumentField
"""
class Word(EmbeddedDocument):
word = StringField()
index = IntField()
class SubDoc(EmbeddedDocument):
heading = ListField(StringField())
text = EmbeddedDocumentListField(Word)
class MainDoc(Document):
title = StringField()
content = EmbeddedDocumentField(SubDoc)
word = Word(word='abc', index=1)
update = transform.update(MainDoc, pull__content__text=word)
self.assertEqual(update, {'$pull': {'content.text': SON([('word', u'abc'), ('index', 1)])}})
update = transform.update(MainDoc, pull__content__heading='xyz')
self.assertEqual(update, {'$pull': {'content.heading': 'xyz'}})
if __name__ == '__main__':
unittest.main()

2
tests/queryset/visitor.py
View File

@@ -196,7 +196,7 @@ class QTest(unittest.TestCase):
test2 = test.clone()
self.assertEqual(test2.count(), 3)
self.assertFalse(test2 == test)
self.assertNotEqual(test2, test)
test3 = test2.filter(x=6)
self.assertEqual(test3.count(), 1)

162
tests/test_connection.py
View File

@@ -35,20 +35,19 @@ class ConnectionTest(unittest.TestCase):
mongoengine.connection._dbs = {}
def test_connect(self):
"""Ensure that the connect() method works properly.
"""
"""Ensure that the connect() method works properly."""
connect('mongoenginetest')
conn = get_connection()
self.assertTrue(isinstance(conn, pymongo.mongo_client.MongoClient))
self.assertIsInstance(conn, pymongo.mongo_client.MongoClient)
db = get_db()
self.assertTrue(isinstance(db, pymongo.database.Database))
self.assertIsInstance(db, pymongo.database.Database)
self.assertEqual(db.name, 'mongoenginetest')
connect('mongoenginetest2', alias='testdb')
conn = get_connection('testdb')
self.assertTrue(isinstance(conn, pymongo.mongo_client.MongoClient))
self.assertIsInstance(conn, pymongo.mongo_client.MongoClient)
def test_connect_in_mocking(self):
"""Ensure that the connect() method works properly in mocking.
@@ -60,31 +59,31 @@ class ConnectionTest(unittest.TestCase):
connect('mongoenginetest', host='mongomock://localhost')
conn = get_connection()
self.assertTrue(isinstance(conn, mongomock.MongoClient))
self.assertIsInstance(conn, mongomock.MongoClient)
connect('mongoenginetest2', host='mongomock://localhost', alias='testdb2')
conn = get_connection('testdb2')
self.assertTrue(isinstance(conn, mongomock.MongoClient))
self.assertIsInstance(conn, mongomock.MongoClient)
connect('mongoenginetest3', host='mongodb://localhost', is_mock=True, alias='testdb3')
conn = get_connection('testdb3')
self.assertTrue(isinstance(conn, mongomock.MongoClient))
self.assertIsInstance(conn, mongomock.MongoClient)
connect('mongoenginetest4', is_mock=True, alias='testdb4')
conn = get_connection('testdb4')
self.assertTrue(isinstance(conn, mongomock.MongoClient))
self.assertIsInstance(conn, mongomock.MongoClient)
connect(host='mongodb://localhost:27017/mongoenginetest5', is_mock=True, alias='testdb5')
conn = get_connection('testdb5')
self.assertTrue(isinstance(conn, mongomock.MongoClient))
self.assertIsInstance(conn, mongomock.MongoClient)
connect(host='mongomock://localhost:27017/mongoenginetest6', alias='testdb6')
conn = get_connection('testdb6')
self.assertTrue(isinstance(conn, mongomock.MongoClient))
self.assertIsInstance(conn, mongomock.MongoClient)
connect(host='mongomock://localhost:27017/mongoenginetest7', is_mock=True, alias='testdb7')
conn = get_connection('testdb7')
self.assertTrue(isinstance(conn, mongomock.MongoClient))
self.assertIsInstance(conn, mongomock.MongoClient)
def test_connect_with_host_list(self):
"""Ensure that the connect() method works when host is a list
@@ -98,27 +97,27 @@ class ConnectionTest(unittest.TestCase):
connect(host=['mongomock://localhost'])
conn = get_connection()
self.assertTrue(isinstance(conn, mongomock.MongoClient))
self.assertIsInstance(conn, mongomock.MongoClient)
connect(host=['mongodb://localhost'], is_mock=True, alias='testdb2')
conn = get_connection('testdb2')
self.assertTrue(isinstance(conn, mongomock.MongoClient))
self.assertIsInstance(conn, mongomock.MongoClient)
connect(host=['localhost'], is_mock=True, alias='testdb3')
conn = get_connection('testdb3')
self.assertTrue(isinstance(conn, mongomock.MongoClient))
self.assertIsInstance(conn, mongomock.MongoClient)
connect(host=['mongomock://localhost:27017', 'mongomock://localhost:27018'], alias='testdb4')
conn = get_connection('testdb4')
self.assertTrue(isinstance(conn, mongomock.MongoClient))
self.assertIsInstance(conn, mongomock.MongoClient)
connect(host=['mongodb://localhost:27017', 'mongodb://localhost:27018'], is_mock=True, alias='testdb5')
conn = get_connection('testdb5')
self.assertTrue(isinstance(conn, mongomock.MongoClient))
self.assertIsInstance(conn, mongomock.MongoClient)
connect(host=['localhost:27017', 'localhost:27018'], is_mock=True, alias='testdb6')
conn = get_connection('testdb6')
self.assertTrue(isinstance(conn, mongomock.MongoClient))
self.assertIsInstance(conn, mongomock.MongoClient)
def test_disconnect(self):
"""Ensure that the disconnect() method works properly
@@ -146,8 +145,7 @@ class ConnectionTest(unittest.TestCase):
self.assertEqual(expected_connection, actual_connection)
def test_connect_uri(self):
"""Ensure that the connect() method works properly with uri's
"""
"""Ensure that the connect() method works properly with URIs."""
c = connect(db='mongoenginetest', alias='admin')
c.admin.system.users.remove({})
c.mongoenginetest.system.users.remove({})
@@ -165,10 +163,10 @@ class ConnectionTest(unittest.TestCase):
connect("testdb_uri", host='mongodb://username:password@localhost/mongoenginetest')
conn = get_connection()
self.assertTrue(isinstance(conn, pymongo.mongo_client.MongoClient))
self.assertIsInstance(conn, pymongo.mongo_client.MongoClient)
db = get_db()
self.assertTrue(isinstance(db, pymongo.database.Database))
self.assertIsInstance(db, pymongo.database.Database)
self.assertEqual(db.name, 'mongoenginetest')
c.admin.system.users.remove({})
@@ -181,10 +179,10 @@ class ConnectionTest(unittest.TestCase):
connect("mongoenginetest", host='mongodb://localhost/')
conn = get_connection()
self.assertTrue(isinstance(conn, pymongo.mongo_client.MongoClient))
self.assertIsInstance(conn, pymongo.mongo_client.MongoClient)
db = get_db()
self.assertTrue(isinstance(db, pymongo.database.Database))
self.assertIsInstance(db, pymongo.database.Database)
self.assertEqual(db.name, 'mongoenginetest')
def test_connect_uri_default_db(self):
@@ -194,10 +192,10 @@ class ConnectionTest(unittest.TestCase):
connect(host='mongodb://localhost/')
conn = get_connection()
self.assertTrue(isinstance(conn, pymongo.mongo_client.MongoClient))
self.assertIsInstance(conn, pymongo.mongo_client.MongoClient)
db = get_db()
self.assertTrue(isinstance(db, pymongo.database.Database))
self.assertIsInstance(db, pymongo.database.Database)
self.assertEqual(db.name, 'test')
def test_uri_without_credentials_doesnt_override_conn_settings(self):
@@ -214,9 +212,8 @@ class ConnectionTest(unittest.TestCase):
self.assertRaises(OperationFailure, get_db)
def test_connect_uri_with_authsource(self):
"""Ensure that the connect() method works well with
the option `authSource` in URI.
This feature was introduced in MongoDB 2.4 and removed in 2.6
"""Ensure that the connect() method works well with `authSource`
option in the URI.
"""
# Create users
c = connect('mongoenginetest')
@@ -225,30 +222,31 @@ class ConnectionTest(unittest.TestCase):
# Authentication fails without "authSource"
if IS_PYMONGO_3:
test_conn = connect('mongoenginetest', alias='test1',
host='mongodb://username2:password@localhost/mongoenginetest')
test_conn = connect(
'mongoenginetest', alias='test1',
host='mongodb://username2:password@localhost/mongoenginetest'
)
self.assertRaises(OperationFailure, test_conn.server_info)
else:
self.assertRaises(
MongoEngineConnectionError, connect, 'mongoenginetest',
alias='test1',
MongoEngineConnectionError,
connect, 'mongoenginetest', alias='test1',
host='mongodb://username2:password@localhost/mongoenginetest'
)
self.assertRaises(MongoEngineConnectionError, get_db, 'test1')
# Authentication succeeds with "authSource"
connect(
authd_conn = connect(
'mongoenginetest', alias='test2',
host=('mongodb://username2:password@localhost/'
'mongoenginetest?authSource=admin')
)
# This will fail starting from MongoDB 2.6+
db = get_db('test2')
self.assertTrue(isinstance(db, pymongo.database.Database))
self.assertIsInstance(db, pymongo.database.Database)
self.assertEqual(db.name, 'mongoenginetest')
# Clear all users
c.admin.system.users.remove({})
authd_conn.admin.system.users.remove({})
def test_register_connection(self):
"""Ensure that connections with different aliases may be registered.
@@ -257,10 +255,10 @@ class ConnectionTest(unittest.TestCase):
self.assertRaises(MongoEngineConnectionError, get_connection)
conn = get_connection('testdb')
self.assertTrue(isinstance(conn, pymongo.mongo_client.MongoClient))
self.assertIsInstance(conn, pymongo.mongo_client.MongoClient)
db = get_db('testdb')
self.assertTrue(isinstance(db, pymongo.database.Database))
self.assertIsInstance(db, pymongo.database.Database)
self.assertEqual(db.name, 'mongoenginetest2')
def test_register_connection_defaults(self):
@@ -269,11 +267,10 @@ class ConnectionTest(unittest.TestCase):
register_connection('testdb', 'mongoenginetest', host=None, port=None)
conn = get_connection('testdb')
self.assertTrue(isinstance(conn, pymongo.mongo_client.MongoClient))
self.assertIsInstance(conn, pymongo.mongo_client.MongoClient)
def test_connection_kwargs(self):
"""Ensure that connection kwargs get passed to pymongo.
"""
"""Ensure that connection kwargs get passed to pymongo."""
connect('mongoenginetest', alias='t1', tz_aware=True)
conn = get_connection('t1')
@@ -283,6 +280,77 @@ class ConnectionTest(unittest.TestCase):
conn = get_connection('t2')
self.assertFalse(get_tz_awareness(conn))
def test_connection_pool_via_kwarg(self):
"""Ensure we can specify a max connection pool size using
a connection kwarg.
"""
# Use "max_pool_size" or "maxpoolsize" depending on PyMongo version
# (former was changed to the latter as described in
# https://jira.mongodb.org/browse/PYTHON-854).
# TODO remove once PyMongo < 3.0 support is dropped
if pymongo.version_tuple[0] >= 3:
pool_size_kwargs = {'maxpoolsize': 100}
else:
pool_size_kwargs = {'max_pool_size': 100}
conn = connect('mongoenginetest', alias='max_pool_size_via_kwarg', **pool_size_kwargs)
self.assertEqual(conn.max_pool_size, 100)
def test_connection_pool_via_uri(self):
"""Ensure we can specify a max connection pool size using
an option in a connection URI.
"""
if pymongo.version_tuple[0] == 2 and pymongo.version_tuple[1] < 9:
raise SkipTest('maxpoolsize as a URI option is only supported in PyMongo v2.9+')
conn = connect(host='mongodb://localhost/test?maxpoolsize=100', alias='max_pool_size_via_uri')
self.assertEqual(conn.max_pool_size, 100)
def test_write_concern(self):
"""Ensure write concern can be specified in connect() via
a kwarg or as part of the connection URI.
"""
conn1 = connect(alias='conn1', host='mongodb://localhost/testing?w=1&j=true')
conn2 = connect('testing', alias='conn2', w=1, j=True)
if IS_PYMONGO_3:
self.assertEqual(conn1.write_concern.document, {'w': 1, 'j': True})
self.assertEqual(conn2.write_concern.document, {'w': 1, 'j': True})
else:
self.assertEqual(dict(conn1.write_concern), {'w': 1, 'j': True})
self.assertEqual(dict(conn2.write_concern), {'w': 1, 'j': True})
def test_connect_with_replicaset_via_uri(self):
"""Ensure connect() works when specifying a replicaSet via the
MongoDB URI.
"""
if IS_PYMONGO_3:
c = connect(host='mongodb://localhost/test?replicaSet=local-rs')
db = get_db()
self.assertIsInstance(db, pymongo.database.Database)
self.assertEqual(db.name, 'test')
else:
# PyMongo < v3.x raises an exception:
# "localhost:27017 is not a member of replica set local-rs"
with self.assertRaises(MongoEngineConnectionError):
c = connect(host='mongodb://localhost/test?replicaSet=local-rs')
def test_connect_with_replicaset_via_kwargs(self):
"""Ensure connect() works when specifying a replicaSet via the
connection kwargs
"""
if IS_PYMONGO_3:
c = connect(replicaset='local-rs')
self.assertEqual(c._MongoClient__options.replica_set_name,
'local-rs')
db = get_db()
self.assertIsInstance(db, pymongo.database.Database)
self.assertEqual(db.name, 'test')
else:
# PyMongo < v3.x raises an exception:
# "localhost:27017 is not a member of replica set local-rs"
with self.assertRaises(MongoEngineConnectionError):
c = connect(replicaset='local-rs')
def test_datetime(self):
connect('mongoenginetest', tz_aware=True)
d = datetime.datetime(2010, 5, 5, tzinfo=utc)
@@ -296,6 +364,12 @@ class ConnectionTest(unittest.TestCase):
date_doc = DateDoc.objects.first()
self.assertEqual(d, date_doc.the_date)
def test_read_preference_from_parse(self):
if IS_PYMONGO_3:
from pymongo import ReadPreference
conn = connect(host="mongodb://a1.vpc,a2.vpc,a3.vpc/prod?readPreference=secondaryPreferred")
self.assertEqual(conn.read_preference, ReadPreference.SECONDARY_PREFERRED)
def test_multiple_connection_settings(self):
connect('mongoenginetest', alias='t1', host="localhost")
@@ -303,8 +377,8 @@ class ConnectionTest(unittest.TestCase):
mongo_connections = mongoengine.connection._connections
self.assertEqual(len(mongo_connections.items()), 2)
self.assertTrue('t1' in mongo_connections.keys())
self.assertTrue('t2' in mongo_connections.keys())
self.assertIn('t1', mongo_connections.keys())
self.assertIn('t2', mongo_connections.keys())
if not IS_PYMONGO_3:
self.assertEqual(mongo_connections['t1'].host, 'localhost')
self.assertEqual(mongo_connections['t2'].host, '127.0.0.1')

159
tests/test_context_managers.py
View File

@@ -89,15 +89,15 @@ class ContextManagersTest(unittest.TestCase):
with no_dereference(Group) as Group:
group = Group.objects.first()
self.assertTrue(all([not isinstance(m, User)
for m in group.members]))
self.assertFalse(isinstance(group.ref, User))
self.assertFalse(isinstance(group.generic, User))
for m in group.members:
self.assertNotIsInstance(m, User)
self.assertNotIsInstance(group.ref, User)
self.assertNotIsInstance(group.generic, User)
self.assertTrue(all([isinstance(m, User)
for m in group.members]))
self.assertTrue(isinstance(group.ref, User))
self.assertTrue(isinstance(group.generic, User))
for m in group.members:
self.assertIsInstance(m, User)
self.assertIsInstance(group.ref, User)
self.assertIsInstance(group.generic, User)
def test_no_dereference_context_manager_dbref(self):
"""Ensure that DBRef items in ListFields aren't dereferenced.
@@ -129,19 +129,17 @@ class ContextManagersTest(unittest.TestCase):
group = Group.objects.first()
self.assertTrue(all([not isinstance(m, User)
for m in group.members]))
self.assertFalse(isinstance(group.ref, User))
self.assertFalse(isinstance(group.generic, User))
self.assertNotIsInstance(group.ref, User)
self.assertNotIsInstance(group.generic, User)
self.assertTrue(all([isinstance(m, User)
for m in group.members]))
self.assertTrue(isinstance(group.ref, User))
self.assertTrue(isinstance(group.generic, User))
self.assertIsInstance(group.ref, User)
self.assertIsInstance(group.generic, User)
def test_no_sub_classes(self):
class A(Document):
x = IntField()
y = IntField()
meta = {'allow_inheritance': True}
class B(A):
@@ -152,29 +150,29 @@ class ContextManagersTest(unittest.TestCase):
A.drop_collection()
A(x=10, y=20).save()
A(x=15, y=30).save()
B(x=20, y=40).save()
B(x=30, y=50).save()
C(x=40, y=60).save()
A(x=10).save()
A(x=15).save()
B(x=20).save()
B(x=30).save()
C(x=40).save()
self.assertEqual(A.objects.count(), 5)
self.assertEqual(B.objects.count(), 3)
self.assertEqual(C.objects.count(), 1)
with no_sub_classes(A) as A:
with no_sub_classes(A):
self.assertEqual(A.objects.count(), 2)
for obj in A.objects:
self.assertEqual(obj.__class__, A)
with no_sub_classes(B) as B:
with no_sub_classes(B):
self.assertEqual(B.objects.count(), 2)
for obj in B.objects:
self.assertEqual(obj.__class__, B)
with no_sub_classes(C) as C:
with no_sub_classes(C):
self.assertEqual(C.objects.count(), 1)
for obj in C.objects:
@@ -185,18 +183,125 @@ class ContextManagersTest(unittest.TestCase):
self.assertEqual(B.objects.count(), 3)
self.assertEqual(C.objects.count(), 1)
def test_no_sub_classes_modification_to_document_class_are_temporary(self):
class A(Document):
x = IntField()
meta = {'allow_inheritance': True}
class B(A):
z = IntField()
self.assertEqual(A._subclasses, ('A', 'A.B'))
with no_sub_classes(A):
self.assertEqual(A._subclasses, ('A',))
self.assertEqual(A._subclasses, ('A', 'A.B'))
self.assertEqual(B._subclasses, ('A.B',))
with no_sub_classes(B):
self.assertEqual(B._subclasses, ('A.B',))
self.assertEqual(B._subclasses, ('A.B',))
def test_no_subclass_context_manager_does_not_swallow_exception(self):
class User(Document):
name = StringField()
with self.assertRaises(TypeError):
with no_sub_classes(User):
raise TypeError()
def test_query_counter_does_not_swallow_exception(self):
with self.assertRaises(TypeError):
with query_counter() as q:
raise TypeError()
def test_query_counter_temporarily_modifies_profiling_level(self):
connect('mongoenginetest')
db = get_db()
initial_profiling_level = db.profiling_level()
try:
NEW_LEVEL = 1
db.set_profiling_level(NEW_LEVEL)
self.assertEqual(db.profiling_level(), NEW_LEVEL)
with query_counter() as q:
self.assertEqual(db.profiling_level(), 2)
self.assertEqual(db.profiling_level(), NEW_LEVEL)
except Exception:
db.set_profiling_level(initial_profiling_level) # Ensures it gets reseted no matter the outcome of the test
raise
def test_query_counter(self):
connect('mongoenginetest')
db = get_db()
db.test.find({})
collection = db.query_counter
collection.drop()
def issue_1_count_query():
collection.find({}).count()
def issue_1_insert_query():
collection.insert_one({'test': 'garbage'})
def issue_1_find_query():
collection.find_one()
counter = 0
with query_counter() as q:
self.assertEqual(q, counter)
self.assertEqual(q, counter) # Ensures previous count query did not get counted
for _ in range(10):
issue_1_insert_query()
counter += 1
self.assertEqual(q, counter)
for _ in range(4):
issue_1_find_query()
counter += 1
self.assertEqual(q, counter)
for _ in range(3):
issue_1_count_query()
counter += 1
self.assertEqual(q, counter)
def test_query_counter_counts_getmore_queries(self):
connect('mongoenginetest')
db = get_db()
collection = db.query_counter
collection.drop()
many_docs = [{'test': 'garbage %s' % i} for i in range(150)]
collection.insert_many(many_docs) # first batch of documents contains 101 documents
with query_counter() as q:
self.assertEqual(0, q)
self.assertEqual(q, 0)
list(collection.find())
self.assertEqual(q, 2) # 1st select + 1 getmore
for i in range(1, 51):
db.test.find({}).count()
def test_query_counter_ignores_particular_queries(self):
connect('mongoenginetest')
db = get_db()
self.assertEqual(50, q)
collection = db.query_counter
collection.insert_many([{'test': 'garbage %s' % i} for i in range(10)])
with query_counter() as q:
self.assertEqual(q, 0)
cursor = collection.find()
self.assertEqual(q, 0) # cursor wasn't opened yet
_ = next(cursor) # opens the cursor and fires the find query
self.assertEqual(q, 1)
cursor.close() # issues a `killcursors` query that is ignored by the context
self.assertEqual(q, 1)
_ = db.system.indexes.find_one() # queries on db.system.indexes are ignored as well
self.assertEqual(q, 1)
if __name__ == '__main__':
unittest.main()

56
tests/test_datastructures.py
View File

@@ -1,6 +1,21 @@
import unittest
from mongoengine.base.datastructures import StrictDict, SemiStrictDict
from mongoengine.base.datastructures import StrictDict, BaseList
class TestBaseList(unittest.TestCase):
def test_iter_simple(self):
values = [True, False, True, False]
base_list = BaseList(values, instance=None, name='my_name')
self.assertEqual(values, list(base_list))
def test_iter_allow_modification_while_iterating_withou_error(self):
# regular list allows for this, thus this subclass must comply to that
base_list = BaseList([True, False, True, False], instance=None, name='my_name')
for idx, val in enumerate(base_list):
if val:
base_list.pop(idx)
class TestStrictDict(unittest.TestCase):
@@ -76,44 +91,5 @@ class TestStrictDict(unittest.TestCase):
assert dict(**d) == {'a': 1, 'b': 2}
class TestSemiSrictDict(TestStrictDict):
def strict_dict_class(self, *args, **kwargs):
return SemiStrictDict.create(*args, **kwargs)
def test_init_fails_on_nonexisting_attrs(self):
# disable irrelevant test
pass
def test_setattr_raises_on_nonexisting_attr(self):
# disable irrelevant test
pass
def test_setattr_getattr_nonexisting_attr_succeeds(self):
d = self.dtype()
d.x = 1
self.assertEqual(d.x, 1)
def test_init_succeeds_with_nonexisting_attrs(self):
d = self.dtype(a=1, b=1, c=1, x=2)
self.assertEqual((d.a, d.b, d.c, d.x), (1, 1, 1, 2))
def test_iter_with_nonexisting_attrs(self):
d = self.dtype(a=1, b=1, c=1, x=2)
self.assertEqual(list(d), ['a', 'b', 'c', 'x'])
def test_iteritems_with_nonexisting_attrs(self):
d = self.dtype(a=1, b=1, c=1, x=2)
self.assertEqual(list(d.iteritems()), [('a', 1), ('b', 1), ('c', 1), ('x', 2)])
def tets_cmp_with_strict_dicts(self):
d = self.dtype(a=1, b=1, c=1)
dd = StrictDict.create(("a", "b", "c"))(a=1, b=1, c=1)
self.assertEqual(d, dd)
def test_cmp_with_strict_dict_with_nonexisting_attrs(self):
d = self.dtype(a=1, b=1, c=1, x=2)
dd = StrictDict.create(("a", "b", "c", "x"))(a=1, b=1, c=1, x=2)
self.assertEqual(d, dd)
if __name__ == '__main__':
unittest.main()

48
tests/test_dereference.py
View File

@@ -200,8 +200,8 @@ class FieldTest(unittest.TestCase):
group = Group(author=user, members=[user]).save()
raw_data = Group._get_collection().find_one()
self.assertTrue(isinstance(raw_data['author'], DBRef))
self.assertTrue(isinstance(raw_data['members'][0], DBRef))
self.assertIsInstance(raw_data['author'], DBRef)
self.assertIsInstance(raw_data['members'][0], DBRef)
group = Group.objects.first()
self.assertEqual(group.author, user)
@@ -224,8 +224,8 @@ class FieldTest(unittest.TestCase):
self.assertEqual(group.members, [user])
raw_data = Group._get_collection().find_one()
self.assertTrue(isinstance(raw_data['author'], ObjectId))
self.assertTrue(isinstance(raw_data['members'][0], ObjectId))
self.assertIsInstance(raw_data['author'], ObjectId)
self.assertIsInstance(raw_data['members'][0], ObjectId)
def test_recursive_reference(self):
"""Ensure that ReferenceFields can reference their own documents.
@@ -469,7 +469,7 @@ class FieldTest(unittest.TestCase):
self.assertEqual(q, 4)
for m in group_obj.members:
self.assertTrue('User' in m.__class__.__name__)
self.assertIn('User', m.__class__.__name__)
# Document select_related
with query_counter() as q:
@@ -485,7 +485,7 @@ class FieldTest(unittest.TestCase):
self.assertEqual(q, 4)
for m in group_obj.members:
self.assertTrue('User' in m.__class__.__name__)
self.assertIn('User', m.__class__.__name__)
# Queryset select_related
with query_counter() as q:
@@ -502,7 +502,7 @@ class FieldTest(unittest.TestCase):
self.assertEqual(q, 4)
for m in group_obj.members:
self.assertTrue('User' in m.__class__.__name__)
self.assertIn('User', m.__class__.__name__)
UserA.drop_collection()
UserB.drop_collection()
@@ -560,7 +560,7 @@ class FieldTest(unittest.TestCase):
self.assertEqual(q, 4)
for m in group_obj.members:
self.assertTrue('User' in m.__class__.__name__)
self.assertIn('User', m.__class__.__name__)
# Document select_related
with query_counter() as q:
@@ -576,7 +576,7 @@ class FieldTest(unittest.TestCase):
self.assertEqual(q, 4)
for m in group_obj.members:
self.assertTrue('User' in m.__class__.__name__)
self.assertIn('User', m.__class__.__name__)
# Queryset select_related
with query_counter() as q:
@@ -593,7 +593,7 @@ class FieldTest(unittest.TestCase):
self.assertEqual(q, 4)
for m in group_obj.members:
self.assertTrue('User' in m.__class__.__name__)
self.assertIn('User', m.__class__.__name__)
UserA.drop_collection()
UserB.drop_collection()
@@ -633,7 +633,7 @@ class FieldTest(unittest.TestCase):
self.assertEqual(q, 2)
for k, m in group_obj.members.iteritems():
self.assertTrue(isinstance(m, User))
self.assertIsInstance(m, User)
# Document select_related
with query_counter() as q:
@@ -646,7 +646,7 @@ class FieldTest(unittest.TestCase):
self.assertEqual(q, 2)
for k, m in group_obj.members.iteritems():
self.assertTrue(isinstance(m, User))
self.assertIsInstance(m, User)
# Queryset select_related
with query_counter() as q:
@@ -660,7 +660,7 @@ class FieldTest(unittest.TestCase):
self.assertEqual(q, 2)
for k, m in group_obj.members.iteritems():
self.assertTrue(isinstance(m, User))
self.assertIsInstance(m, User)
User.drop_collection()
Group.drop_collection()
@@ -715,7 +715,7 @@ class FieldTest(unittest.TestCase):
self.assertEqual(q, 4)
for k, m in group_obj.members.iteritems():
self.assertTrue('User' in m.__class__.__name__)
self.assertIn('User', m.__class__.__name__)
# Document select_related
with query_counter() as q:
@@ -731,7 +731,7 @@ class FieldTest(unittest.TestCase):
self.assertEqual(q, 4)
for k, m in group_obj.members.iteritems():
self.assertTrue('User' in m.__class__.__name__)
self.assertIn('User', m.__class__.__name__)
# Queryset select_related
with query_counter() as q:
@@ -748,7 +748,7 @@ class FieldTest(unittest.TestCase):
self.assertEqual(q, 4)
for k, m in group_obj.members.iteritems():
self.assertTrue('User' in m.__class__.__name__)
self.assertIn('User', m.__class__.__name__)
Group.objects.delete()
Group().save()
@@ -806,7 +806,7 @@ class FieldTest(unittest.TestCase):
self.assertEqual(q, 2)
for k, m in group_obj.members.iteritems():
self.assertTrue(isinstance(m, UserA))
self.assertIsInstance(m, UserA)
# Document select_related
with query_counter() as q:
@@ -822,7 +822,7 @@ class FieldTest(unittest.TestCase):
self.assertEqual(q, 2)
for k, m in group_obj.members.iteritems():
self.assertTrue(isinstance(m, UserA))
self.assertIsInstance(m, UserA)
# Queryset select_related
with query_counter() as q:
@@ -839,7 +839,7 @@ class FieldTest(unittest.TestCase):
self.assertEqual(q, 2)
for k, m in group_obj.members.iteritems():
self.assertTrue(isinstance(m, UserA))
self.assertIsInstance(m, UserA)
UserA.drop_collection()
Group.drop_collection()
@@ -894,7 +894,7 @@ class FieldTest(unittest.TestCase):
self.assertEqual(q, 4)
for k, m in group_obj.members.iteritems():
self.assertTrue('User' in m.__class__.__name__)
self.assertIn('User', m.__class__.__name__)
# Document select_related
with query_counter() as q:
@@ -910,7 +910,7 @@ class FieldTest(unittest.TestCase):
self.assertEqual(q, 4)
for k, m in group_obj.members.iteritems():
self.assertTrue('User' in m.__class__.__name__)
self.assertIn('User', m.__class__.__name__)
# Queryset select_related
with query_counter() as q:
@@ -927,7 +927,7 @@ class FieldTest(unittest.TestCase):
self.assertEqual(q, 4)
for k, m in group_obj.members.iteritems():
self.assertTrue('User' in m.__class__.__name__)
self.assertIn('User', m.__class__.__name__)
Group.objects.delete()
Group().save()
@@ -1209,10 +1209,10 @@ class FieldTest(unittest.TestCase):
# Can't use query_counter across databases - so test the _data object
book = Book.objects.first()
self.assertFalse(isinstance(book._data['author'], User))
self.assertNotIsInstance(book._data['author'], User)
book.select_related()
self.assertTrue(isinstance(book._data['author'], User))
self.assertIsInstance(book._data['author'], User)
def test_non_ascii_pk(self):
"""

38
tests/test_utils.py Normal file
View File

@@ -0,0 +1,38 @@
import unittest
import re
from mongoengine.base.utils import LazyRegexCompiler
signal_output = []
class LazyRegexCompilerTest(unittest.TestCase):
def test_lazy_regex_compiler_verify_laziness_of_descriptor(self):
class UserEmail(object):
EMAIL_REGEX = LazyRegexCompiler('@', flags=32)
descriptor = UserEmail.__dict__['EMAIL_REGEX']
self.assertIsNone(descriptor._compiled_regex)
regex = UserEmail.EMAIL_REGEX
self.assertEqual(regex, re.compile('@', flags=32))
self.assertEqual(regex.search('user@domain.com').group(), '@')
user_email = UserEmail()
self.assertIs(user_email.EMAIL_REGEX, UserEmail.EMAIL_REGEX)
def test_lazy_regex_compiler_verify_cannot_set_descriptor_on_instance(self):
class UserEmail(object):
EMAIL_REGEX = LazyRegexCompiler('@')
user_email = UserEmail()
with self.assertRaises(AttributeError):
user_email.EMAIL_REGEX = re.compile('@')
def test_lazy_regex_compiler_verify_can_override_class_attr(self):
class UserEmail(object):
EMAIL_REGEX = LazyRegexCompiler('@')
UserEmail.EMAIL_REGEX = re.compile('cookies')
self.assertEqual(UserEmail.EMAIL_REGEX.search('Cake & cookies').group(), 'cookies')

82
tests/utils.py Normal file
View File

@@ -0,0 +1,82 @@
import unittest
from nose.plugins.skip import SkipTest
from mongoengine import connect
from mongoengine.connection import get_db, get_connection
from mongoengine.python_support import IS_PYMONGO_3
MONGO_TEST_DB = 'mongoenginetest' # standard name for the test database
class MongoDBTestCase(unittest.TestCase):
"""Base class for tests that need a mongodb connection
It ensures that the db is clean at the beginning and dropped at the end automatically
"""
@classmethod
def setUpClass(cls):
cls._connection = connect(db=MONGO_TEST_DB)
cls._connection.drop_database(MONGO_TEST_DB)
cls.db = get_db()
@classmethod
def tearDownClass(cls):
cls._connection.drop_database(MONGO_TEST_DB)
def get_mongodb_version():
"""Return the version tuple of the MongoDB server that the default
connection is connected to.
"""
return tuple(get_connection().server_info()['versionArray'])
def _decorated_with_ver_requirement(func, ver_tuple):
"""Return a given function decorated with the version requirement
for a particular MongoDB version tuple.
"""
def _inner(*args, **kwargs):
mongodb_ver = get_mongodb_version()
if mongodb_ver >= ver_tuple:
return func(*args, **kwargs)
raise SkipTest('Needs MongoDB v{}+'.format(
'.'.join([str(v) for v in ver_tuple])
))
_inner.__name__ = func.__name__
_inner.__doc__ = func.__doc__
return _inner
def needs_mongodb_v26(func):
"""Raise a SkipTest exception if we're working with MongoDB version
lower than v2.6.
"""
return _decorated_with_ver_requirement(func, (2, 6))
def needs_mongodb_v3(func):
"""Raise a SkipTest exception if we're working with MongoDB version
lower than v3.0.
"""
return _decorated_with_ver_requirement(func, (3, 0))
def skip_pymongo3(f):
"""Raise a SkipTest exception if we're running a test against
PyMongo v3.x.
"""
def _inner(*args, **kwargs):
if IS_PYMONGO_3:
raise SkipTest("Useless with PyMongo 3+")
return f(*args, **kwargs)
_inner.__name__ = f.__name__
_inner.__doc__ = f.__doc__
return _inner