Merge branch 'dev' into pull_124
This commit is contained in:
Ross Lawley
@@ -5,8 +5,9 @@ import unittest
|
||||
import pymongo
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from mongoengine.queryset import (QuerySet, MultipleObjectsReturned,
|
||||
DoesNotExist, QueryFieldList)
|
||||
from mongoengine.queryset import (QuerySet, QuerySetManager,
|
||||
MultipleObjectsReturned, DoesNotExist,
|
||||
QueryFieldList)
|
||||
from mongoengine import *
|
||||
|
||||
|
||||
@@ -105,6 +106,10 @@ class QuerySetTest(unittest.TestCase):
|
||||
people = list(self.Person.objects[1:1])
|
||||
self.assertEqual(len(people), 0)
|
||||
|
||||
# Test slice out of range
|
||||
people = list(self.Person.objects[80000:80001])
|
||||
self.assertEqual(len(people), 0)
|
||||
|
||||
def test_find_one(self):
|
||||
"""Ensure that a query using find_one returns a valid result.
|
||||
"""
|
||||
@@ -162,7 +167,7 @@ class QuerySetTest(unittest.TestCase):
|
||||
|
||||
person = self.Person.objects.get(age__lt=30)
|
||||
self.assertEqual(person.name, "User A")
|
||||
|
||||
|
||||
def test_find_array_position(self):
|
||||
"""Ensure that query by array position works.
|
||||
"""
|
||||
@@ -177,7 +182,7 @@ class QuerySetTest(unittest.TestCase):
|
||||
posts = ListField(EmbeddedDocumentField(Post))
|
||||
|
||||
Blog.drop_collection()
|
||||
|
||||
|
||||
Blog.objects.create(tags=['a', 'b'])
|
||||
self.assertEqual(len(Blog.objects(tags__0='a')), 1)
|
||||
self.assertEqual(len(Blog.objects(tags__0='b')), 0)
|
||||
@@ -207,6 +212,55 @@ class QuerySetTest(unittest.TestCase):
|
||||
|
||||
Blog.drop_collection()
|
||||
|
||||
def test_update_array_position(self):
|
||||
"""Ensure that updating by array position works.
|
||||
|
||||
Check update() and update_one() can take syntax like:
|
||||
set__posts__1__comments__1__name="testc"
|
||||
Check that it only works for ListFields.
|
||||
"""
|
||||
class Comment(EmbeddedDocument):
|
||||
name = StringField()
|
||||
|
||||
class Post(EmbeddedDocument):
|
||||
comments = ListField(EmbeddedDocumentField(Comment))
|
||||
|
||||
class Blog(Document):
|
||||
tags = ListField(StringField())
|
||||
posts = ListField(EmbeddedDocumentField(Post))
|
||||
|
||||
Blog.drop_collection()
|
||||
|
||||
comment1 = Comment(name='testa')
|
||||
comment2 = Comment(name='testb')
|
||||
post1 = Post(comments=[comment1, comment2])
|
||||
post2 = Post(comments=[comment2, comment2])
|
||||
blog1 = Blog.objects.create(posts=[post1, post2])
|
||||
blog2 = Blog.objects.create(posts=[post2, post1])
|
||||
|
||||
# Update all of the first comments of second posts of all blogs
|
||||
blog = Blog.objects().update(set__posts__1__comments__0__name="testc")
|
||||
testc_blogs = Blog.objects(posts__1__comments__0__name="testc")
|
||||
self.assertEqual(len(testc_blogs), 2)
|
||||
|
||||
Blog.drop_collection()
|
||||
|
||||
blog1 = Blog.objects.create(posts=[post1, post2])
|
||||
blog2 = Blog.objects.create(posts=[post2, post1])
|
||||
|
||||
# Update only the first blog returned by the query
|
||||
blog = Blog.objects().update_one(
|
||||
set__posts__1__comments__1__name="testc")
|
||||
testc_blogs = Blog.objects(posts__1__comments__1__name="testc")
|
||||
self.assertEqual(len(testc_blogs), 1)
|
||||
|
||||
# Check that using this indexing syntax on a non-list fails
|
||||
def non_list_indexing():
|
||||
Blog.objects().update(set__posts__1__comments__0__name__1="asdf")
|
||||
self.assertRaises(InvalidQueryError, non_list_indexing)
|
||||
|
||||
Blog.drop_collection()
|
||||
|
||||
def test_get_or_create(self):
|
||||
"""Ensure that ``get_or_create`` returns one result or creates a new
|
||||
document.
|
||||
@@ -226,16 +280,16 @@ class QuerySetTest(unittest.TestCase):
|
||||
person, created = self.Person.objects.get_or_create(age=30)
|
||||
self.assertEqual(person.name, "User B")
|
||||
self.assertEqual(created, False)
|
||||
|
||||
|
||||
person, created = self.Person.objects.get_or_create(age__lt=30)
|
||||
self.assertEqual(person.name, "User A")
|
||||
self.assertEqual(created, False)
|
||||
|
||||
|
||||
# Try retrieving when no objects exists - new doc should be created
|
||||
kwargs = dict(age=50, defaults={'name': 'User C'})
|
||||
person, created = self.Person.objects.get_or_create(**kwargs)
|
||||
self.assertEqual(created, True)
|
||||
|
||||
|
||||
person = self.Person.objects.get(age=50)
|
||||
self.assertEqual(person.name, "User C")
|
||||
|
||||
@@ -328,7 +382,7 @@ class QuerySetTest(unittest.TestCase):
|
||||
self.assertEqual(obj, person)
|
||||
obj = self.Person.objects(Q(name__iexact='gUIDO VAN rOSSU')).first()
|
||||
self.assertEqual(obj, None)
|
||||
|
||||
|
||||
# Test unsafe expressions
|
||||
person = self.Person(name='Guido van Rossum [.\'Geek\']')
|
||||
person.save()
|
||||
@@ -593,6 +647,81 @@ class QuerySetTest(unittest.TestCase):
|
||||
|
||||
Email.drop_collection()
|
||||
|
||||
def test_slicing_fields(self):
|
||||
"""Ensure that query slicing an array works.
|
||||
"""
|
||||
class Numbers(Document):
|
||||
n = ListField(IntField())
|
||||
|
||||
Numbers.drop_collection()
|
||||
|
||||
numbers = Numbers(n=[0,1,2,3,4,5,-5,-4,-3,-2,-1])
|
||||
numbers.save()
|
||||
|
||||
# first three
|
||||
numbers = Numbers.objects.fields(slice__n=3).get()
|
||||
self.assertEquals(numbers.n, [0, 1, 2])
|
||||
|
||||
# last three
|
||||
numbers = Numbers.objects.fields(slice__n=-3).get()
|
||||
self.assertEquals(numbers.n, [-3, -2, -1])
|
||||
|
||||
# skip 2, limit 3
|
||||
numbers = Numbers.objects.fields(slice__n=[2, 3]).get()
|
||||
self.assertEquals(numbers.n, [2, 3, 4])
|
||||
|
||||
# skip to fifth from last, limit 4
|
||||
numbers = Numbers.objects.fields(slice__n=[-5, 4]).get()
|
||||
self.assertEquals(numbers.n, [-5, -4, -3, -2])
|
||||
|
||||
# skip to fifth from last, limit 10
|
||||
numbers = Numbers.objects.fields(slice__n=[-5, 10]).get()
|
||||
self.assertEquals(numbers.n, [-5, -4, -3, -2, -1])
|
||||
|
||||
# skip to fifth from last, limit 10 dict method
|
||||
numbers = Numbers.objects.fields(n={"$slice": [-5, 10]}).get()
|
||||
self.assertEquals(numbers.n, [-5, -4, -3, -2, -1])
|
||||
|
||||
def test_slicing_nested_fields(self):
|
||||
"""Ensure that query slicing an embedded array works.
|
||||
"""
|
||||
|
||||
class EmbeddedNumber(EmbeddedDocument):
|
||||
n = ListField(IntField())
|
||||
|
||||
class Numbers(Document):
|
||||
embedded = EmbeddedDocumentField(EmbeddedNumber)
|
||||
|
||||
Numbers.drop_collection()
|
||||
|
||||
numbers = Numbers()
|
||||
numbers.embedded = EmbeddedNumber(n=[0,1,2,3,4,5,-5,-4,-3,-2,-1])
|
||||
numbers.save()
|
||||
|
||||
# first three
|
||||
numbers = Numbers.objects.fields(slice__embedded__n=3).get()
|
||||
self.assertEquals(numbers.embedded.n, [0, 1, 2])
|
||||
|
||||
# last three
|
||||
numbers = Numbers.objects.fields(slice__embedded__n=-3).get()
|
||||
self.assertEquals(numbers.embedded.n, [-3, -2, -1])
|
||||
|
||||
# skip 2, limit 3
|
||||
numbers = Numbers.objects.fields(slice__embedded__n=[2, 3]).get()
|
||||
self.assertEquals(numbers.embedded.n, [2, 3, 4])
|
||||
|
||||
# skip to fifth from last, limit 4
|
||||
numbers = Numbers.objects.fields(slice__embedded__n=[-5, 4]).get()
|
||||
self.assertEquals(numbers.embedded.n, [-5, -4, -3, -2])
|
||||
|
||||
# skip to fifth from last, limit 10
|
||||
numbers = Numbers.objects.fields(slice__embedded__n=[-5, 10]).get()
|
||||
self.assertEquals(numbers.embedded.n, [-5, -4, -3, -2, -1])
|
||||
|
||||
# skip to fifth from last, limit 10 dict method
|
||||
numbers = Numbers.objects.fields(embedded__n={"$slice": [-5, 10]}).get()
|
||||
self.assertEquals(numbers.embedded.n, [-5, -4, -3, -2, -1])
|
||||
|
||||
def test_find_embedded(self):
|
||||
"""Ensure that an embedded document is properly returned from a query.
|
||||
"""
|
||||
@@ -674,7 +803,7 @@ class QuerySetTest(unittest.TestCase):
|
||||
posts = [post.id for post in q]
|
||||
published_posts = (post1, post2, post3, post5, post6)
|
||||
self.assertTrue(all(obj.id in posts for obj in published_posts))
|
||||
|
||||
|
||||
|
||||
# Check Q object combination
|
||||
date = datetime(2010, 1, 10)
|
||||
@@ -714,7 +843,7 @@ class QuerySetTest(unittest.TestCase):
|
||||
|
||||
obj = self.Person.objects(Q(name__not=re.compile('^bob'))).first()
|
||||
self.assertEqual(obj, person)
|
||||
|
||||
|
||||
obj = self.Person.objects(Q(name__not=re.compile('^Gui'))).first()
|
||||
self.assertEqual(obj, None)
|
||||
|
||||
@@ -786,7 +915,7 @@ class QuerySetTest(unittest.TestCase):
|
||||
|
||||
class BlogPost(Document):
|
||||
name = StringField(db_field='doc-name')
|
||||
comments = ListField(EmbeddedDocumentField(Comment),
|
||||
comments = ListField(EmbeddedDocumentField(Comment),
|
||||
db_field='cmnts')
|
||||
|
||||
BlogPost.drop_collection()
|
||||
@@ -958,7 +1087,7 @@ class QuerySetTest(unittest.TestCase):
|
||||
BlogPost.objects.update_one(unset__hits=1)
|
||||
post.reload()
|
||||
self.assertEqual(post.hits, None)
|
||||
|
||||
|
||||
BlogPost.drop_collection()
|
||||
|
||||
def test_update_pull(self):
|
||||
@@ -1027,7 +1156,7 @@ class QuerySetTest(unittest.TestCase):
|
||||
"""
|
||||
|
||||
# run a map/reduce operation spanning all posts
|
||||
results = BlogPost.objects.map_reduce(map_f, reduce_f)
|
||||
results = BlogPost.objects.map_reduce(map_f, reduce_f, "myresults")
|
||||
results = list(results)
|
||||
self.assertEqual(len(results), 4)
|
||||
|
||||
@@ -1038,7 +1167,7 @@ class QuerySetTest(unittest.TestCase):
|
||||
self.assertEqual(film.value, 3)
|
||||
|
||||
BlogPost.drop_collection()
|
||||
|
||||
|
||||
def test_map_reduce_with_custom_object_ids(self):
|
||||
"""Ensure that QuerySet.map_reduce works properly with custom
|
||||
primary keys.
|
||||
@@ -1047,24 +1176,24 @@ class QuerySetTest(unittest.TestCase):
|
||||
class BlogPost(Document):
|
||||
title = StringField(primary_key=True)
|
||||
tags = ListField(StringField())
|
||||
|
||||
|
||||
post1 = BlogPost(title="Post #1", tags=["mongodb", "mongoengine"])
|
||||
post2 = BlogPost(title="Post #2", tags=["django", "mongodb"])
|
||||
post3 = BlogPost(title="Post #3", tags=["hitchcock films"])
|
||||
|
||||
|
||||
post1.save()
|
||||
post2.save()
|
||||
post3.save()
|
||||
|
||||
|
||||
self.assertEqual(BlogPost._fields['title'].db_field, '_id')
|
||||
self.assertEqual(BlogPost._meta['id_field'], 'title')
|
||||
|
||||
|
||||
map_f = """
|
||||
function() {
|
||||
emit(this._id, 1);
|
||||
}
|
||||
"""
|
||||
|
||||
|
||||
# reduce to a list of tag ids and counts
|
||||
reduce_f = """
|
||||
function(key, values) {
|
||||
@@ -1075,10 +1204,10 @@ class QuerySetTest(unittest.TestCase):
|
||||
return total;
|
||||
}
|
||||
"""
|
||||
|
||||
results = BlogPost.objects.map_reduce(map_f, reduce_f)
|
||||
|
||||
results = BlogPost.objects.map_reduce(map_f, reduce_f, "myresults")
|
||||
results = list(results)
|
||||
|
||||
|
||||
self.assertEqual(results[0].object, post1)
|
||||
self.assertEqual(results[1].object, post2)
|
||||
self.assertEqual(results[2].object, post3)
|
||||
@@ -1168,7 +1297,7 @@ class QuerySetTest(unittest.TestCase):
|
||||
|
||||
finalize_f = """
|
||||
function(key, value) {
|
||||
// f(sec_since_epoch,y,z) =
|
||||
// f(sec_since_epoch,y,z) =
|
||||
// log10(z) + ((y*sec_since_epoch) / 45000)
|
||||
z_10 = Math.log(value.z) / Math.log(10);
|
||||
weight = z_10 + ((value.y * value.t_s) / 45000);
|
||||
@@ -1187,6 +1316,7 @@ class QuerySetTest(unittest.TestCase):
|
||||
results = Link.objects.order_by("-value")
|
||||
results = results.map_reduce(map_f,
|
||||
reduce_f,
|
||||
"myresults",
|
||||
finalize_f=finalize_f,
|
||||
scope=scope)
|
||||
results = list(results)
|
||||
@@ -1289,6 +1419,7 @@ class QuerySetTest(unittest.TestCase):
|
||||
class BlogPost(Document):
|
||||
tags = ListField(StringField())
|
||||
deleted = BooleanField(default=False)
|
||||
date = DateTimeField(default=datetime.now)
|
||||
|
||||
@queryset_manager
|
||||
def objects(doc_cls, queryset):
|
||||
@@ -1296,7 +1427,7 @@ class QuerySetTest(unittest.TestCase):
|
||||
|
||||
@queryset_manager
|
||||
def music_posts(doc_cls, queryset):
|
||||
return queryset(tags='music', deleted=False)
|
||||
return queryset(tags='music', deleted=False).order_by('-date')
|
||||
|
||||
BlogPost.drop_collection()
|
||||
|
||||
@@ -1312,7 +1443,7 @@ class QuerySetTest(unittest.TestCase):
|
||||
self.assertEqual([p.id for p in BlogPost.objects],
|
||||
[post1.id, post2.id, post3.id])
|
||||
self.assertEqual([p.id for p in BlogPost.music_posts],
|
||||
[post1.id, post2.id])
|
||||
[post2.id, post1.id])
|
||||
|
||||
BlogPost.drop_collection()
|
||||
|
||||
@@ -1452,10 +1583,12 @@ class QuerySetTest(unittest.TestCase):
|
||||
class Test(Document):
|
||||
testdict = DictField()
|
||||
|
||||
Test.drop_collection()
|
||||
|
||||
t = Test(testdict={'f': 'Value'})
|
||||
t.save()
|
||||
|
||||
self.assertEqual(len(Test.objects(testdict__f__startswith='Val')), 0)
|
||||
self.assertEqual(len(Test.objects(testdict__f__startswith='Val')), 1)
|
||||
self.assertEqual(len(Test.objects(testdict__f='Value')), 1)
|
||||
Test.drop_collection()
|
||||
|
||||
@@ -1514,12 +1647,12 @@ class QuerySetTest(unittest.TestCase):
|
||||
title = StringField()
|
||||
date = DateTimeField()
|
||||
location = GeoPointField()
|
||||
|
||||
|
||||
def __unicode__(self):
|
||||
return self.title
|
||||
|
||||
|
||||
Event.drop_collection()
|
||||
|
||||
|
||||
event1 = Event(title="Coltrane Motion @ Double Door",
|
||||
date=datetime.now() - timedelta(days=1),
|
||||
location=[41.909889, -87.677137])
|
||||
@@ -1529,7 +1662,7 @@ class QuerySetTest(unittest.TestCase):
|
||||
event3 = Event(title="Coltrane Motion @ Empty Bottle",
|
||||
date=datetime.now(),
|
||||
location=[41.900474, -87.686638])
|
||||
|
||||
|
||||
event1.save()
|
||||
event2.save()
|
||||
event3.save()
|
||||
@@ -1541,7 +1674,7 @@ class QuerySetTest(unittest.TestCase):
|
||||
self.assertEqual(events.count(), 3)
|
||||
self.assertEqual(list(events), [event1, event3, event2])
|
||||
|
||||
# find events within 5 miles of pitchfork office, chicago
|
||||
# find events within 5 degrees of pitchfork office, chicago
|
||||
point_and_distance = [[41.9120459, -87.67892], 5]
|
||||
events = Event.objects(location__within_distance=point_and_distance)
|
||||
self.assertEqual(events.count(), 2)
|
||||
@@ -1549,24 +1682,24 @@ class QuerySetTest(unittest.TestCase):
|
||||
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=[41.9120459, -87.67892])
|
||||
events = events.order_by("-date")
|
||||
self.assertEqual(events.count(), 3)
|
||||
self.assertEqual(list(events), [event3, event1, event2])
|
||||
|
||||
# find events around san francisco
|
||||
|
||||
# find events within 10 degrees of san francisco
|
||||
point_and_distance = [[37.7566023, -122.415579], 10]
|
||||
events = Event.objects(location__within_distance=point_and_distance)
|
||||
self.assertEqual(events.count(), 1)
|
||||
self.assertEqual(events[0], event2)
|
||||
|
||||
# find events within 1 mile of greenpoint, broolyn, nyc, ny
|
||||
|
||||
# find events within 1 degree of greenpoint, broolyn, nyc, ny
|
||||
point_and_distance = [[40.7237134, -73.9509714], 1]
|
||||
events = Event.objects(location__within_distance=point_and_distance)
|
||||
self.assertEqual(events.count(), 0)
|
||||
|
||||
|
||||
# ensure ordering is respected by "within_distance"
|
||||
point_and_distance = [[41.9120459, -87.67892], 10]
|
||||
events = Event.objects(location__within_distance=point_and_distance)
|
||||
@@ -1579,9 +1712,61 @@ class QuerySetTest(unittest.TestCase):
|
||||
events = Event.objects(location__within_box=box)
|
||||
self.assertEqual(events.count(), 1)
|
||||
self.assertEqual(events[0].id, event2.id)
|
||||
|
||||
|
||||
Event.drop_collection()
|
||||
|
||||
def test_spherical_geospatial_operators(self):
|
||||
"""Ensure that spherical geospatial queries are working
|
||||
"""
|
||||
class Point(Document):
|
||||
location = GeoPointField()
|
||||
|
||||
Point.drop_collection()
|
||||
|
||||
# These points are one degree apart, which (according to Google Maps)
|
||||
# is about 110 km apart at this place on the Earth.
|
||||
north_point = Point(location=[-122, 38]) # Near Concord, CA
|
||||
south_point = Point(location=[-122, 37]) # Near Santa Cruz, CA
|
||||
north_point.save()
|
||||
south_point.save()
|
||||
|
||||
earth_radius = 6378.009; # in km (needs to be a float for dividing by)
|
||||
|
||||
# Finds both points because they are within 60 km of the reference
|
||||
# point equidistant between them.
|
||||
points = Point.objects(location__near_sphere=[-122, 37.5])
|
||||
self.assertEqual(points.count(), 2)
|
||||
|
||||
# Same behavior for _within_spherical_distance
|
||||
points = Point.objects(
|
||||
location__within_spherical_distance=[[-122, 37.5], 60/earth_radius]
|
||||
);
|
||||
self.assertEqual(points.count(), 2)
|
||||
|
||||
# Finds both points, but orders the north point first because it's
|
||||
# closer to the reference point to the north.
|
||||
points = Point.objects(location__near_sphere=[-122, 38.5])
|
||||
self.assertEqual(points.count(), 2)
|
||||
self.assertEqual(points[0].id, north_point.id)
|
||||
self.assertEqual(points[1].id, south_point.id)
|
||||
|
||||
# Finds both points, but orders the south point first because it's
|
||||
# closer to the reference point to the south.
|
||||
points = Point.objects(location__near_sphere=[-122, 36.5])
|
||||
self.assertEqual(points.count(), 2)
|
||||
self.assertEqual(points[0].id, south_point.id)
|
||||
self.assertEqual(points[1].id, north_point.id)
|
||||
|
||||
# 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]
|
||||
);
|
||||
self.assertEqual(points.count(), 1)
|
||||
self.assertEqual(points[0].id, south_point.id)
|
||||
|
||||
Point.drop_collection()
|
||||
|
||||
def test_custom_querysets(self):
|
||||
"""Ensure that custom QuerySet classes may be used.
|
||||
"""
|
||||
@@ -1602,6 +1787,53 @@ class QuerySetTest(unittest.TestCase):
|
||||
|
||||
Post.drop_collection()
|
||||
|
||||
def test_custom_querysets_set_manager_directly(self):
|
||||
"""Ensure that custom QuerySet classes may be used.
|
||||
"""
|
||||
|
||||
class CustomQuerySet(QuerySet):
|
||||
def not_empty(self):
|
||||
return len(self) > 0
|
||||
|
||||
class CustomQuerySetManager(QuerySetManager):
|
||||
queryset_class = CustomQuerySet
|
||||
|
||||
class Post(Document):
|
||||
objects = CustomQuerySetManager()
|
||||
|
||||
Post.drop_collection()
|
||||
|
||||
self.assertTrue(isinstance(Post.objects, CustomQuerySet))
|
||||
self.assertFalse(Post.objects.not_empty())
|
||||
|
||||
Post().save()
|
||||
self.assertTrue(Post.objects.not_empty())
|
||||
|
||||
Post.drop_collection()
|
||||
|
||||
def test_custom_querysets_managers_directly(self):
|
||||
"""Ensure that custom QuerySet classes may be used.
|
||||
"""
|
||||
|
||||
class CustomQuerySetManager(QuerySetManager):
|
||||
|
||||
@staticmethod
|
||||
def get_queryset(doc_cls, queryset):
|
||||
return queryset(is_published=True)
|
||||
|
||||
class Post(Document):
|
||||
is_published = BooleanField(default=False)
|
||||
published = CustomQuerySetManager()
|
||||
|
||||
Post.drop_collection()
|
||||
|
||||
Post().save()
|
||||
Post(is_published=True).save()
|
||||
self.assertEquals(Post.objects.count(), 2)
|
||||
self.assertEquals(Post.published.count(), 1)
|
||||
|
||||
Post.drop_collection()
|
||||
|
||||
def test_call_after_limits_set(self):
|
||||
"""Ensure that re-filtering after slicing works
|
||||
"""
|
||||
@@ -1637,6 +1869,35 @@ class QuerySetTest(unittest.TestCase):
|
||||
|
||||
Number.drop_collection()
|
||||
|
||||
def test_clone(self):
|
||||
"""Ensure that cloning clones complex querysets
|
||||
"""
|
||||
class Number(Document):
|
||||
n = IntField()
|
||||
|
||||
Number.drop_collection()
|
||||
|
||||
for i in xrange(1, 101):
|
||||
t = Number(n=i)
|
||||
t.save()
|
||||
|
||||
test = Number.objects
|
||||
test2 = test.clone()
|
||||
self.assertFalse(test == test2)
|
||||
self.assertEqual(test.count(), test2.count())
|
||||
|
||||
test = test.filter(n__gt=11)
|
||||
test2 = test.clone()
|
||||
self.assertFalse(test == test2)
|
||||
self.assertEqual(test.count(), test2.count())
|
||||
|
||||
test = test.limit(10)
|
||||
test2 = test.clone()
|
||||
self.assertFalse(test == test2)
|
||||
self.assertEqual(test.count(), test2.count())
|
||||
|
||||
Number.drop_collection()
|
||||
|
||||
def test_unset_reference(self):
|
||||
class Comment(Document):
|
||||
text = StringField()
|
||||
@@ -1658,6 +1919,39 @@ class QuerySetTest(unittest.TestCase):
|
||||
Comment.drop_collection()
|
||||
Post.drop_collection()
|
||||
|
||||
def test_order_works_with_custom_db_field_names(self):
|
||||
class Number(Document):
|
||||
n = IntField(db_field='number')
|
||||
|
||||
Number.drop_collection()
|
||||
|
||||
n2 = Number.objects.create(n=2)
|
||||
n1 = Number.objects.create(n=1)
|
||||
|
||||
self.assertEqual(list(Number.objects), [n2,n1])
|
||||
self.assertEqual(list(Number.objects.order_by('n')), [n1,n2])
|
||||
|
||||
Number.drop_collection()
|
||||
|
||||
def test_order_works_with_primary(self):
|
||||
"""Ensure that order_by and primary work.
|
||||
"""
|
||||
class Number(Document):
|
||||
n = IntField(primary_key=True)
|
||||
|
||||
Number.drop_collection()
|
||||
|
||||
Number(n=1).save()
|
||||
Number(n=2).save()
|
||||
Number(n=3).save()
|
||||
|
||||
numbers = [n.n for n in Number.objects.order_by('-n')]
|
||||
self.assertEquals([3, 2, 1], numbers)
|
||||
|
||||
numbers = [n.n for n in Number.objects.order_by('+n')]
|
||||
self.assertEquals([1, 2, 3], numbers)
|
||||
Number.drop_collection()
|
||||
|
||||
|
||||
class QTest(unittest.TestCase):
|
||||
|
||||
@@ -1679,7 +1973,7 @@ class QTest(unittest.TestCase):
|
||||
|
||||
query = {'age': {'$gte': 18}, 'name': 'test'}
|
||||
self.assertEqual((q1 & q2 & q3 & q4 & q5).to_query(Person), query)
|
||||
|
||||
|
||||
def test_q_with_dbref(self):
|
||||
"""Ensure Q objects handle DBRefs correctly"""
|
||||
connect(db='mongoenginetest')
|
||||
@@ -1721,7 +2015,7 @@ class QTest(unittest.TestCase):
|
||||
query = Q(x__lt=100) & Q(y__ne='NotMyString')
|
||||
query &= Q(y__in=['a', 'b', 'c']) & Q(x__gt=-100)
|
||||
mongo_query = {
|
||||
'x': {'$lt': 100, '$gt': -100},
|
||||
'x': {'$lt': 100, '$gt': -100},
|
||||
'y': {'$ne': 'NotMyString', '$in': ['a', 'b', 'c']},
|
||||
}
|
||||
self.assertEqual(query.to_query(TestDoc), mongo_query)
|
||||
@@ -1795,6 +2089,30 @@ class QTest(unittest.TestCase):
|
||||
for condition in conditions:
|
||||
self.assertTrue(condition in query['$or'])
|
||||
|
||||
|
||||
def test_q_clone(self):
|
||||
|
||||
class TestDoc(Document):
|
||||
x = IntField()
|
||||
|
||||
TestDoc.drop_collection()
|
||||
for i in xrange(1, 101):
|
||||
t = TestDoc(x=i)
|
||||
t.save()
|
||||
|
||||
# Check normal cases work without an error
|
||||
test = TestDoc.objects(Q(x__lt=7) & Q(x__gt=3))
|
||||
|
||||
self.assertEqual(test.count(), 3)
|
||||
|
||||
test2 = test.clone()
|
||||
self.assertEqual(test2.count(), 3)
|
||||
self.assertFalse(test2 == test)
|
||||
|
||||
test2.filter(x=6)
|
||||
self.assertEqual(test2.count(), 1)
|
||||
self.assertEqual(test.count(), 3)
|
||||
|
||||
class QueryFieldListTest(unittest.TestCase):
|
||||
def test_empty(self):
|
||||
q = QueryFieldList()
|
||||
@@ -1805,51 +2123,52 @@ class QueryFieldListTest(unittest.TestCase):
|
||||
|
||||
def test_include_include(self):
|
||||
q = QueryFieldList()
|
||||
q += QueryFieldList(fields=['a', 'b'], direction=QueryFieldList.ONLY)
|
||||
q += QueryFieldList(fields=['a', 'b'], value=QueryFieldList.ONLY)
|
||||
self.assertEqual(q.as_dict(), {'a': True, 'b': True})
|
||||
q += QueryFieldList(fields=['b', 'c'], direction=QueryFieldList.ONLY)
|
||||
q += QueryFieldList(fields=['b', 'c'], value=QueryFieldList.ONLY)
|
||||
self.assertEqual(q.as_dict(), {'b': True})
|
||||
|
||||
def test_include_exclude(self):
|
||||
q = QueryFieldList()
|
||||
q += QueryFieldList(fields=['a', 'b'], direction=QueryFieldList.ONLY)
|
||||
q += QueryFieldList(fields=['a', 'b'], value=QueryFieldList.ONLY)
|
||||
self.assertEqual(q.as_dict(), {'a': True, 'b': True})
|
||||
q += QueryFieldList(fields=['b', 'c'], direction=QueryFieldList.EXCLUDE)
|
||||
q += QueryFieldList(fields=['b', 'c'], value=QueryFieldList.EXCLUDE)
|
||||
self.assertEqual(q.as_dict(), {'a': True})
|
||||
|
||||
def test_exclude_exclude(self):
|
||||
q = QueryFieldList()
|
||||
q += QueryFieldList(fields=['a', 'b'], direction=QueryFieldList.EXCLUDE)
|
||||
q += QueryFieldList(fields=['a', 'b'], value=QueryFieldList.EXCLUDE)
|
||||
self.assertEqual(q.as_dict(), {'a': False, 'b': False})
|
||||
q += QueryFieldList(fields=['b', 'c'], direction=QueryFieldList.EXCLUDE)
|
||||
q += QueryFieldList(fields=['b', 'c'], value=QueryFieldList.EXCLUDE)
|
||||
self.assertEqual(q.as_dict(), {'a': False, 'b': False, 'c': False})
|
||||
|
||||
def test_exclude_include(self):
|
||||
q = QueryFieldList()
|
||||
q += QueryFieldList(fields=['a', 'b'], direction=QueryFieldList.EXCLUDE)
|
||||
q += QueryFieldList(fields=['a', 'b'], value=QueryFieldList.EXCLUDE)
|
||||
self.assertEqual(q.as_dict(), {'a': False, 'b': False})
|
||||
q += QueryFieldList(fields=['b', 'c'], direction=QueryFieldList.ONLY)
|
||||
q += QueryFieldList(fields=['b', 'c'], value=QueryFieldList.ONLY)
|
||||
self.assertEqual(q.as_dict(), {'c': True})
|
||||
|
||||
def test_always_include(self):
|
||||
q = QueryFieldList(always_include=['x', 'y'])
|
||||
q += QueryFieldList(fields=['a', 'b', 'x'], direction=QueryFieldList.EXCLUDE)
|
||||
q += QueryFieldList(fields=['b', 'c'], direction=QueryFieldList.ONLY)
|
||||
q += QueryFieldList(fields=['a', 'b', 'x'], value=QueryFieldList.EXCLUDE)
|
||||
q += QueryFieldList(fields=['b', 'c'], value=QueryFieldList.ONLY)
|
||||
self.assertEqual(q.as_dict(), {'x': True, 'y': True, 'c': True})
|
||||
|
||||
|
||||
def test_reset(self):
|
||||
q = QueryFieldList(always_include=['x', 'y'])
|
||||
q += QueryFieldList(fields=['a', 'b', 'x'], direction=QueryFieldList.EXCLUDE)
|
||||
q += QueryFieldList(fields=['b', 'c'], direction=QueryFieldList.ONLY)
|
||||
q += QueryFieldList(fields=['a', 'b', 'x'], value=QueryFieldList.EXCLUDE)
|
||||
q += QueryFieldList(fields=['b', 'c'], value=QueryFieldList.ONLY)
|
||||
self.assertEqual(q.as_dict(), {'x': True, 'y': True, 'c': True})
|
||||
q.reset()
|
||||
self.assertFalse(q)
|
||||
q += QueryFieldList(fields=['b', 'c'], direction=QueryFieldList.ONLY)
|
||||
q += QueryFieldList(fields=['b', 'c'], value=QueryFieldList.ONLY)
|
||||
self.assertEqual(q.as_dict(), {'x': True, 'y': True, 'b': True, 'c': True})
|
||||
|
||||
|
||||
|
||||
def test_using_a_slice(self):
|
||||
q = QueryFieldList()
|
||||
q += QueryFieldList(fields=['a'], value={"$slice": 5})
|
||||
self.assertEqual(q.as_dict(), {'a': {"$slice": 5}})
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
Reference in New Issue
Block a user