# -*- coding: utf-8 -*- import unittest from datetime import datetime from nose.plugins.skip import SkipTest from pymongo.errors import OperationFailure import pymongo from six import iteritems from mongoengine import * from mongoengine.connection import get_db from mongoengine.mongodb_support import get_mongodb_version, MONGODB_32, MONGODB_3 from tests.utils import requires_mongodb_gte_26, requires_mongodb_lte_32, requires_mongodb_gte_34 __all__ = ("IndexesTest", ) class IndexesTest(unittest.TestCase): def setUp(self): self.connection = connect(db='mongoenginetest') self.db = get_db() self.mongodb_version = get_mongodb_version() class Person(Document): name = StringField() age = IntField() non_field = True meta = {"allow_inheritance": True} self.Person = Person def tearDown(self): self.connection.drop_database(self.db) def test_indexes_document(self): """Ensure that indexes are used when meta[indexes] is specified for Documents """ self._index_test(Document) def test_indexes_dynamic_document(self): """Ensure that indexes are used when meta[indexes] is specified for Dynamic Documents """ self._index_test(DynamicDocument) def _index_test(self, InheritFrom): class BlogPost(InheritFrom): date = DateTimeField(db_field='addDate', default=datetime.now) category = StringField() tags = ListField(StringField()) meta = { 'indexes': [ '-date', 'tags', ('category', '-date') ] } expected_specs = [{'fields': [('addDate', -1)]}, {'fields': [('tags', 1)]}, {'fields': [('category', 1), ('addDate', -1)]}] self.assertEqual(expected_specs, BlogPost._meta['index_specs']) BlogPost.ensure_indexes() info = BlogPost.objects._collection.index_information() # _id, '-date', 'tags', ('cat', 'date') self.assertEqual(len(info), 4) info = [value['key'] for key, value in iteritems(info)] for expected in expected_specs: self.assertIn(expected['fields'], info) def _index_test_inheritance(self, InheritFrom): class BlogPost(InheritFrom): date = DateTimeField(db_field='addDate', default=datetime.now) category = StringField() tags = ListField(StringField()) meta = { 'indexes': [ '-date', 'tags', ('category', '-date') ], 'allow_inheritance': True } expected_specs = [{'fields': [('_cls', 1), ('addDate', -1)]}, {'fields': [('_cls', 1), ('tags', 1)]}, {'fields': [('_cls', 1), ('category', 1), ('addDate', -1)]}] self.assertEqual(expected_specs, BlogPost._meta['index_specs']) BlogPost.ensure_indexes() info = BlogPost.objects._collection.index_information() # _id, '-date', 'tags', ('cat', 'date') # NB: there is no index on _cls by itself, since # the indices on -date and tags will both contain # _cls as first element in the key self.assertEqual(len(info), 4) info = [value['key'] for key, value in iteritems(info)] for expected in expected_specs: self.assertIn(expected['fields'], info) class ExtendedBlogPost(BlogPost): title = StringField() meta = {'indexes': ['title']} expected_specs.append({'fields': [('_cls', 1), ('title', 1)]}) self.assertEqual(expected_specs, ExtendedBlogPost._meta['index_specs']) BlogPost.drop_collection() ExtendedBlogPost.ensure_indexes() info = ExtendedBlogPost.objects._collection.index_information() info = [value['key'] for key, value in iteritems(info)] for expected in expected_specs: self.assertIn(expected['fields'], info) def test_indexes_document_inheritance(self): """Ensure that indexes are used when meta[indexes] is specified for Documents """ self._index_test_inheritance(Document) def test_indexes_dynamic_document_inheritance(self): """Ensure that indexes are used when meta[indexes] is specified for Dynamic Documents """ self._index_test_inheritance(DynamicDocument) def test_inherited_index(self): """Ensure index specs are inhertited correctly""" class A(Document): title = StringField() meta = { 'indexes': [ { 'fields': ('title',), }, ], 'allow_inheritance': True, } class B(A): description = StringField() self.assertEqual(A._meta['index_specs'], B._meta['index_specs']) self.assertEqual([{'fields': [('_cls', 1), ('title', 1)]}], A._meta['index_specs']) def test_index_no_cls(self): """Ensure index specs are inhertited correctly""" class A(Document): title = StringField() meta = { 'indexes': [ {'fields': ('title',), 'cls': False}, ], 'allow_inheritance': True, 'index_cls': False } self.assertEqual([('title', 1)], A._meta['index_specs'][0]['fields']) A._get_collection().drop_indexes() A.ensure_indexes() info = A._get_collection().index_information() self.assertEqual(len(info.keys()), 2) class B(A): c = StringField() d = StringField() meta = { 'indexes': [{'fields': ['c']}, {'fields': ['d'], 'cls': True}], 'allow_inheritance': True } self.assertEqual([('c', 1)], B._meta['index_specs'][1]['fields']) self.assertEqual([('_cls', 1), ('d', 1)], B._meta['index_specs'][2]['fields']) def test_build_index_spec_is_not_destructive(self): class MyDoc(Document): keywords = StringField() meta = { 'indexes': ['keywords'], 'allow_inheritance': False } self.assertEqual(MyDoc._meta['index_specs'], [{'fields': [('keywords', 1)]}]) # Force index creation MyDoc.ensure_indexes() self.assertEqual(MyDoc._meta['index_specs'], [{'fields': [('keywords', 1)]}]) def test_embedded_document_index_meta(self): """Ensure that embedded document indexes are created explicitly """ class Rank(EmbeddedDocument): title = StringField(required=True) class Person(Document): name = StringField(required=True) rank = EmbeddedDocumentField(Rank, required=False) meta = { 'indexes': [ 'rank.title', ], 'allow_inheritance': False } self.assertEqual([{'fields': [('rank.title', 1)]}], Person._meta['index_specs']) Person.drop_collection() # Indexes are lazy so use list() to perform query list(Person.objects) info = Person.objects._collection.index_information() info = [value['key'] for key, value in iteritems(info)] self.assertIn([('rank.title', 1)], info) def test_explicit_geo2d_index(self): """Ensure that geo2d indexes work when created via meta[indexes] """ class Place(Document): location = DictField() meta = { 'allow_inheritance': True, 'indexes': [ '*location.point', ] } self.assertEqual([{'fields': [('location.point', '2d')]}], Place._meta['index_specs']) Place.ensure_indexes() info = Place._get_collection().index_information() info = [value['key'] for key, value in iteritems(info)] self.assertIn([('location.point', '2d')], info) def test_explicit_geo2d_index_embedded(self): """Ensure that geo2d indexes work when created via meta[indexes] """ class EmbeddedLocation(EmbeddedDocument): location = DictField() class Place(Document): current = DictField(field=EmbeddedDocumentField('EmbeddedLocation')) meta = { 'allow_inheritance': True, 'indexes': [ '*current.location.point', ] } self.assertEqual([{'fields': [('current.location.point', '2d')]}], Place._meta['index_specs']) Place.ensure_indexes() info = Place._get_collection().index_information() info = [value['key'] for key, value in iteritems(info)] self.assertIn([('current.location.point', '2d')], info) def test_explicit_geosphere_index(self): """Ensure that geosphere indexes work when created via meta[indexes] """ class Place(Document): location = DictField() meta = { 'allow_inheritance': True, 'indexes': [ '(location.point', ] } self.assertEqual([{'fields': [('location.point', '2dsphere')]}], Place._meta['index_specs']) Place.ensure_indexes() info = Place._get_collection().index_information() info = [value['key'] for key, value in iteritems(info)] self.assertIn([('location.point', '2dsphere')], info) def test_explicit_geohaystack_index(self): """Ensure that geohaystack indexes work when created via meta[indexes] """ raise SkipTest('GeoHaystack index creation is not supported for now' 'from meta, as it requires a bucketSize parameter.') class Place(Document): location = DictField() name = StringField() meta = { 'indexes': [ (')location.point', 'name') ] } self.assertEqual([{'fields': [('location.point', 'geoHaystack'), ('name', 1)]}], Place._meta['index_specs']) Place.ensure_indexes() info = Place._get_collection().index_information() info = [value['key'] for key, value in iteritems(info)] self.assertIn([('location.point', 'geoHaystack')], info) def test_create_geohaystack_index(self): """Ensure that geohaystack indexes can be created """ class Place(Document): location = DictField() name = StringField() Place.create_index({'fields': (')location.point', 'name')}, bucketSize=10) info = Place._get_collection().index_information() info = [value['key'] for key, value in iteritems(info)] self.assertIn([('location.point', 'geoHaystack'), ('name', 1)], info) def test_dictionary_indexes(self): """Ensure that indexes are used when meta[indexes] contains dictionaries instead of lists. """ class BlogPost(Document): date = DateTimeField(db_field='addDate', default=datetime.now) category = StringField() tags = ListField(StringField()) meta = { 'indexes': [ {'fields': ['-date'], 'unique': True, 'sparse': True}, ], } self.assertEqual([{'fields': [('addDate', -1)], 'unique': True, 'sparse': True}], BlogPost._meta['index_specs']) BlogPost.drop_collection() info = BlogPost.objects._collection.index_information() # _id, '-date' self.assertEqual(len(info), 2) # Indexes are lazy so use list() to perform query list(BlogPost.objects) info = BlogPost.objects._collection.index_information() info = [(value['key'], value.get('unique', False), value.get('sparse', False)) for key, value in iteritems(info)] self.assertIn(([('addDate', -1)], True, True), info) BlogPost.drop_collection() def test_abstract_index_inheritance(self): class UserBase(Document): user_guid = StringField(required=True) meta = { 'abstract': True, 'indexes': ['user_guid'], 'allow_inheritance': True } class Person(UserBase): name = StringField() meta = { 'indexes': ['name'], } Person.drop_collection() Person(name="test", user_guid='123').save() self.assertEqual(1, Person.objects.count()) info = Person.objects._collection.index_information() self.assertEqual(sorted(info.keys()), ['_cls_1_name_1', '_cls_1_user_guid_1', '_id_']) def test_disable_index_creation(self): """Tests setting auto_create_index to False on the connection will disable any index generation. """ class User(Document): meta = { 'allow_inheritance': True, 'indexes': ['user_guid'], 'auto_create_index': False } user_guid = StringField(required=True) class MongoUser(User): pass User.drop_collection() User(user_guid='123').save() MongoUser(user_guid='123').save() self.assertEqual(2, User.objects.count()) info = User.objects._collection.index_information() self.assertEqual(info.keys(), ['_id_']) User.ensure_indexes() info = User.objects._collection.index_information() self.assertEqual(sorted(info.keys()), ['_cls_1_user_guid_1', '_id_']) def test_embedded_document_index(self): """Tests settings an index on an embedded document """ class Date(EmbeddedDocument): year = IntField(db_field='yr') class BlogPost(Document): title = StringField() date = EmbeddedDocumentField(Date) meta = { 'indexes': [ '-date.year' ], } BlogPost.drop_collection() info = BlogPost.objects._collection.index_information() self.assertEqual(sorted(info.keys()), ['_id_', 'date.yr_-1']) def test_list_embedded_document_index(self): """Ensure list embedded documents can be indexed """ class Tag(EmbeddedDocument): name = StringField(db_field='tag') class BlogPost(Document): title = StringField() tags = ListField(EmbeddedDocumentField(Tag)) meta = { 'indexes': [ 'tags.name' ] } BlogPost.drop_collection() info = BlogPost.objects._collection.index_information() # we don't use _cls in with list fields by default self.assertEqual(sorted(info.keys()), ['_id_', 'tags.tag_1']) post1 = BlogPost(title="Embedded Indexes tests in place", tags=[Tag(name="about"), Tag(name="time")]) post1.save() def test_recursive_embedded_objects_dont_break_indexes(self): class RecursiveObject(EmbeddedDocument): obj = EmbeddedDocumentField('self') class RecursiveDocument(Document): recursive_obj = EmbeddedDocumentField(RecursiveObject) meta = {'allow_inheritance': True} RecursiveDocument.ensure_indexes() info = RecursiveDocument._get_collection().index_information() self.assertEqual(sorted(info.keys()), ['_cls_1', '_id_']) def test_covered_index(self): """Ensure that covered indexes can be used """ IS_MONGODB_3 = get_mongodb_version() >= MONGODB_3 class Test(Document): a = IntField() b = IntField() meta = { 'indexes': ['a'], 'allow_inheritance': False } Test.drop_collection() obj = Test(a=1) obj.save() # Need to be explicit about covered indexes as mongoDB doesn't know if # the documents returned might have more keys in that here. query_plan = Test.objects(id=obj.id).exclude('a').explain() if not IS_MONGODB_3: self.assertFalse(query_plan['indexOnly']) else: self.assertEqual(query_plan.get('queryPlanner').get('winningPlan').get('inputStage').get('stage'), 'IDHACK') query_plan = Test.objects(id=obj.id).only('id').explain() if not IS_MONGODB_3: self.assertTrue(query_plan['indexOnly']) else: self.assertEqual(query_plan.get('queryPlanner').get('winningPlan').get('inputStage').get('stage'), 'IDHACK') query_plan = Test.objects(a=1).only('a').exclude('id').explain() if not IS_MONGODB_3: self.assertTrue(query_plan['indexOnly']) else: self.assertEqual(query_plan.get('queryPlanner').get('winningPlan').get('inputStage').get('stage'), 'IXSCAN') self.assertEqual(query_plan.get('queryPlanner').get('winningPlan').get('stage'), 'PROJECTION') query_plan = Test.objects(a=1).explain() if not IS_MONGODB_3: self.assertFalse(query_plan['indexOnly']) else: self.assertEqual(query_plan.get('queryPlanner').get('winningPlan').get('inputStage').get('stage'), 'IXSCAN') self.assertEqual(query_plan.get('queryPlanner').get('winningPlan').get('stage'), 'FETCH') def test_index_on_id(self): class BlogPost(Document): meta = { 'indexes': [ ['categories', 'id'] ] } title = StringField(required=True) description = StringField(required=True) categories = ListField() BlogPost.drop_collection() indexes = BlogPost.objects._collection.index_information() self.assertEqual(indexes['categories_1__id_1']['key'], [('categories', 1), ('_id', 1)]) def test_hint(self): MONGO_VER = self.mongodb_version TAGS_INDEX_NAME = 'tags_1' class BlogPost(Document): tags = ListField(StringField()) meta = { 'indexes': [ { 'fields': ['tags'], 'name': TAGS_INDEX_NAME } ], } BlogPost.drop_collection() for i in range(10): tags = [("tag %i" % n) for n in range(i % 2)] BlogPost(tags=tags).save() self.assertEqual(BlogPost.objects.count(), 10) self.assertEqual(BlogPost.objects.hint().count(), 10) if MONGO_VER >= MONGODB_32: # Mongo32 throws an error if an index exists (i.e `tags` in our case) # and you use hint on an index name that does not exist with self.assertRaises(OperationFailure): BlogPost.objects.hint([('ZZ', 1)]).count() else: self.assertEqual(BlogPost.objects.hint([('ZZ', 1)]).count(), 10) self.assertEqual(BlogPost.objects.hint(TAGS_INDEX_NAME).count(), 10) with self.assertRaises(Exception): BlogPost.objects.hint(('tags', 1)).next() def test_unique(self): """Ensure that uniqueness constraints are applied to fields. """ class BlogPost(Document): title = StringField() slug = StringField(unique=True) BlogPost.drop_collection() post1 = BlogPost(title='test1', slug='test') post1.save() # Two posts with the same slug is not allowed post2 = BlogPost(title='test2', slug='test') self.assertRaises(NotUniqueError, post2.save) # Ensure backwards compatibilty for errors self.assertRaises(OperationError, post2.save) @requires_mongodb_gte_34 def test_primary_key_unique_not_working_under_mongo_34(self): """Relates to #1445""" class Blog(Document): id = StringField(primary_key=True, unique=True) Blog.drop_collection() with self.assertRaises(OperationFailure) as ctx_err: Blog(id='garbage').save() try: self.assertIn("The field 'unique' is not valid for an _id index specification", str(ctx_err.exception)) except AssertionError: # error is slightly different on python 3.6 self.assertIn("The field 'background' is not valid for an _id index specification", str(ctx_err.exception)) @requires_mongodb_lte_32 def test_primary_key_unique_working_under_mongo_32(self): """Relates to #1445""" class Blog(Document): id = StringField(primary_key=True, unique=True) Blog.drop_collection() Blog(id='garbage').save() def test_unique_with(self): """Ensure that unique_with constraints are applied to fields. """ class Date(EmbeddedDocument): year = IntField(db_field='yr') class BlogPost(Document): title = StringField() date = EmbeddedDocumentField(Date) slug = StringField(unique_with='date.year') BlogPost.drop_collection() post1 = BlogPost(title='test1', date=Date(year=2009), slug='test') post1.save() # day is different so won't raise exception post2 = BlogPost(title='test2', date=Date(year=2010), slug='test') post2.save() # Now there will be two docs with the same slug and the same day: fail post3 = BlogPost(title='test3', date=Date(year=2010), slug='test') self.assertRaises(OperationError, post3.save) def test_unique_embedded_document(self): """Ensure that uniqueness constraints are applied to fields on embedded documents. """ class SubDocument(EmbeddedDocument): year = IntField(db_field='yr') slug = StringField(unique=True) class BlogPost(Document): title = StringField() sub = EmbeddedDocumentField(SubDocument) BlogPost.drop_collection() post1 = BlogPost(title='test1', sub=SubDocument(year=2009, slug="test")) post1.save() # sub.slug is different so won't raise exception post2 = BlogPost(title='test2', sub=SubDocument(year=2010, slug='another-slug')) post2.save() # Now there will be two docs with the same sub.slug post3 = BlogPost(title='test3', sub=SubDocument(year=2010, slug='test')) self.assertRaises(NotUniqueError, post3.save) def test_unique_embedded_document_in_list(self): """ Ensure that the uniqueness constraints are applied to fields in embedded documents, even when the embedded documents in in a list field. """ class SubDocument(EmbeddedDocument): year = IntField(db_field='yr') slug = StringField(unique=True) class BlogPost(Document): title = StringField() subs = ListField(EmbeddedDocumentField(SubDocument)) BlogPost.drop_collection() post1 = BlogPost( title='test1', subs=[ SubDocument(year=2009, slug='conflict'), SubDocument(year=2009, slug='conflict') ] ) post1.save() post2 = BlogPost( title='test2', subs=[SubDocument(year=2014, slug='conflict')] ) self.assertRaises(NotUniqueError, post2.save) 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. """ class SubDocument(EmbeddedDocument): year = IntField(db_field='yr') slug = StringField(unique=True) class BlogPost(Document): title = StringField(unique_with='sub.year') sub = EmbeddedDocumentField(SubDocument) BlogPost.drop_collection() post1 = BlogPost(title='test1', sub=SubDocument(year=2009, slug="test")) post1.save() # sub.slug is different so won't raise exception post2 = BlogPost(title='test2', sub=SubDocument(year=2010, slug='another-slug')) post2.save() # Now there will be two docs with the same sub.slug post3 = BlogPost(title='test3', sub=SubDocument(year=2010, slug='test')) self.assertRaises(NotUniqueError, post3.save) # Now there will be two docs with the same title and year post3 = BlogPost(title='test1', sub=SubDocument(year=2009, slug='test-1')) self.assertRaises(NotUniqueError, post3.save) def test_ttl_indexes(self): class Log(Document): created = DateTimeField(default=datetime.now) meta = { 'indexes': [ {'fields': ['created'], 'expireAfterSeconds': 3600} ] } Log.drop_collection() # Indexes are lazy so use list() to perform query list(Log.objects) info = Log.objects._collection.index_information() self.assertEqual(3600, info['created_1']['expireAfterSeconds']) def test_unique_and_indexes(self): """Ensure that 'unique' constraints aren't overridden by meta.indexes. """ class Customer(Document): cust_id = IntField(unique=True, required=True) meta = { 'indexes': ['cust_id'], 'allow_inheritance': False, } Customer.drop_collection() cust = Customer(cust_id=1) cust.save() cust_dupe = Customer(cust_id=1) try: cust_dupe.save() raise AssertionError("We saved a dupe!") except NotUniqueError: pass def test_primary_save_duplicate_update_existing_object(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) password = StringField() User.drop_collection() user = User(name='huangz', password='secret') user.save() user = User(name='huangz', password='secret2') user.save() 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""" class Comment(EmbeddedDocument): comment_id = IntField(required=True) try: class BlogPost(Document): comments = EmbeddedDocumentField(Comment) meta = {'indexes': [ {'fields': ['pk', 'comments.comment_id'], 'unique': True}]} except UnboundLocalError: self.fail('Unbound local error at index + pk definition') info = BlogPost.objects._collection.index_information() info = [value['key'] for key, value in iteritems(info)] index_item = [('_id', 1), ('comments.comment_id', 1)] self.assertIn(index_item, info) def test_compound_key_embedded(self): class CompoundKey(EmbeddedDocument): name = StringField(required=True) term = StringField(required=True) class ReportEmbedded(Document): key = EmbeddedDocumentField(CompoundKey, primary_key=True) text = StringField() my_key = CompoundKey(name="n", term="ok") report = ReportEmbedded(text="OK", key=my_key).save() self.assertEqual({'text': 'OK', '_id': {'term': 'ok', 'name': 'n'}}, report.to_mongo()) self.assertEqual(report, ReportEmbedded.objects.get(pk=my_key)) def test_compound_key_dictfield(self): class ReportDictField(Document): key = DictField(primary_key=True) text = StringField() my_key = {"name": "n", "term": "ok"} report = ReportDictField(text="OK", key=my_key).save() self.assertEqual({'text': 'OK', '_id': {'term': 'ok', 'name': 'n'}}, report.to_mongo()) # We can't directly call ReportDictField.objects.get(pk=my_key), # because dicts are unordered, and if the order in MongoDB is # different than the one in `my_key`, this test will fail. self.assertEqual(report, ReportDictField.objects.get(pk__name=my_key['name'])) self.assertEqual(report, ReportDictField.objects.get(pk__term=my_key['term'])) def test_string_indexes(self): class MyDoc(Document): provider_ids = DictField() meta = { "indexes": ["provider_ids.foo", "provider_ids.bar"], } info = MyDoc.objects._collection.index_information() info = [value['key'] for key, value in iteritems(info)] self.assertIn([('provider_ids.foo', 1)], info) self.assertIn([('provider_ids.bar', 1)], info) def test_sparse_compound_indexes(self): class MyDoc(Document): provider_ids = DictField() meta = { "indexes": [{'fields': ("provider_ids.foo", "provider_ids.bar"), 'sparse': True}], } info = MyDoc.objects._collection.index_information() self.assertEqual([('provider_ids.foo', 1), ('provider_ids.bar', 1)], info['provider_ids.foo_1_provider_ids.bar_1']['key']) self.assertTrue(info['provider_ids.foo_1_provider_ids.bar_1']['sparse']) @requires_mongodb_gte_26 def test_text_indexes(self): class Book(Document): title = DictField() meta = { "indexes": ["$title"], } indexes = Book.objects._collection.index_information() self.assertIn("title_text", indexes) key = indexes["title_text"]["key"] self.assertIn(('_fts', 'text'), key) def test_hashed_indexes(self): class Book(Document): ref_id = StringField() meta = { "indexes": ["#ref_id"], } indexes = Book.objects._collection.index_information() self.assertIn("ref_id_hashed", indexes) self.assertIn(('ref_id', 'hashed'), indexes["ref_id_hashed"]["key"]) def test_indexes_after_database_drop(self): """ Test to ensure that indexes are re-created on a collection even after the database has been dropped. Issue #812 """ # Use a new connection and database since dropping the database could # cause concurrent tests to fail. connection = connect(db='tempdatabase', alias='test_indexes_after_database_drop') class BlogPost(Document): title = StringField() slug = StringField(unique=True) meta = {'db_alias': 'test_indexes_after_database_drop'} try: BlogPost.drop_collection() # Create Post #1 post1 = BlogPost(title='test1', slug='test') post1.save() # Drop the Database connection.drop_database('tempdatabase') # Re-create Post #1 post1 = BlogPost(title='test1', slug='test') post1.save() # Create Post #2 post2 = BlogPost(title='test2', slug='test') self.assertRaises(NotUniqueError, post2.save) finally: # Drop the temporary database at the end connection.drop_database('tempdatabase') def test_index_dont_send_cls_option(self): """ Ensure that 'cls' option is not sent through ensureIndex. We shouldn't send internal MongoEngine arguments that are not a part of the index spec. This is directly related to the fact that MongoDB doesn't validate the options that are passed to ensureIndex. For more details, see: https://jira.mongodb.org/browse/SERVER-769 """ class TestDoc(Document): txt = StringField() meta = { 'allow_inheritance': True, 'indexes': [ {'fields': ('txt',), 'cls': False} ] } class TestChildDoc(TestDoc): txt2 = StringField() meta = { 'indexes': [ {'fields': ('txt2',), 'cls': False} ] } TestDoc.drop_collection() TestDoc.ensure_indexes() TestChildDoc.ensure_indexes() index_info = TestDoc._get_collection().index_information() for key in index_info: del index_info[key]['v'] # drop the index version - we don't care about that here if 'ns' in index_info[key]: del index_info[key]['ns'] # drop the index namespace - we don't care about that here, MongoDB 3+ if 'dropDups' in index_info[key]: del index_info[key]['dropDups'] # drop the index dropDups - it is deprecated in MongoDB 3+ self.assertEqual(index_info, { 'txt_1': { 'key': [('txt', 1)], 'background': False }, '_id_': { 'key': [('_id', 1)], }, 'txt2_1': { 'key': [('txt2', 1)], 'background': False }, '_cls_1': { 'key': [('_cls', 1)], 'background': False, } }) def test_compound_index_underscore_cls_not_overwritten(self): """ Test that the compound index doesn't get another _cls when it is specified """ class TestDoc(Document): shard_1 = StringField() txt_1 = StringField() meta = { 'collection': 'test', 'allow_inheritance': True, 'sparse': True, 'shard_key': 'shard_1', 'indexes': [ ('shard_1', '_cls', 'txt_1'), ] } TestDoc.drop_collection() TestDoc.ensure_indexes() index_info = TestDoc._get_collection().index_information() self.assertIn('shard_1_1__cls_1_txt_1_1', index_info) if __name__ == '__main__': unittest.main()