Merge remote branch 'nvie/dev' into dev

tests/document.py

@@ -502,7 +502,7 @@ class DocumentTest(unittest.TestCase):
         try:
             recipient.save(validate=False)
         except ValidationError:
-            fail()
+            self.fail()
 
     def test_delete(self):
         """Ensure that document may be deleted using the delete method.
@@ -624,6 +624,108 @@ class DocumentTest(unittest.TestCase):
 
         BlogPost.drop_collection()
 
+
+    def test_reverse_delete_rule_cascade_and_nullify(self):
+        """Ensure that a referenced document is also deleted upon deletion.
+        """
+
+        class BlogPost(Document):
+            content = StringField()
+            author = ReferenceField(self.Person, reverse_delete_rule=CASCADE)
+            reviewer = ReferenceField(self.Person, reverse_delete_rule=NULLIFY)
+
+        self.Person.drop_collection()
+        BlogPost.drop_collection()
+
+        author = self.Person(name='Test User')
+        author.save()
+
+        reviewer = self.Person(name='Re Viewer')
+        reviewer.save()
+
+        post = BlogPost(content = 'Watched some TV')
+        post.author = author
+        post.reviewer = reviewer
+        post.save()
+
+        reviewer.delete()
+        self.assertEqual(len(BlogPost.objects), 1)  # No effect on the BlogPost
+        self.assertEqual(BlogPost.objects.get().reviewer, None)
+
+        # Delete the Person, which should lead to deletion of the BlogPost, too
+        author.delete()
+        self.assertEqual(len(BlogPost.objects), 0)
+
+    def test_reverse_delete_rule_cascade_recurs(self):
+        """Ensure that a chain of documents is also deleted upon cascaded
+        deletion.
+        """
+
+        class BlogPost(Document):
+            content = StringField()
+            author = ReferenceField(self.Person, reverse_delete_rule=CASCADE)
+
+        class Comment(Document):
+            text = StringField()
+            post = ReferenceField(BlogPost, reverse_delete_rule=CASCADE)
+
+
+        author = self.Person(name='Test User')
+        author.save()
+
+        post = BlogPost(content = 'Watched some TV')
+        post.author = author
+        post.save()
+
+        comment = Comment(text = 'Kudos.')
+        comment.post = post
+        comment.save()
+
+        # Delete the Person, which should lead to deletion of the BlogPost, and,
+        # recursively to the Comment, too
+        author.delete()
+        self.assertEqual(len(Comment.objects), 0)
+
+        self.Person.drop_collection()
+        BlogPost.drop_collection()
+        Comment.drop_collection()
+
+    def test_reverse_delete_rule_deny(self):
+        """Ensure that a document cannot be referenced if there are still
+        documents referring to it.
+        """
+
+        class BlogPost(Document):
+            content = StringField()
+            author = ReferenceField(self.Person, reverse_delete_rule=DENY)
+
+        self.Person.drop_collection()
+        BlogPost.drop_collection()
+
+        author = self.Person(name='Test User')
+        author.save()
+
+        post = BlogPost(content = 'Watched some TV')
+        post.author = author
+        post.save()
+
+        # Delete the Person should be denied
+        self.assertRaises(OperationError, author.delete)  # Should raise denied error
+        self.assertEqual(len(BlogPost.objects), 1)  # No objects may have been deleted
+        self.assertEqual(len(self.Person.objects), 1)
+
+        # Other users, that don't have BlogPosts must be removable, like normal
+        author = self.Person(name='Another User')
+        author.save()
+
+        self.assertEqual(len(self.Person.objects), 2)
+        author.delete()
+        self.assertEqual(len(self.Person.objects), 1)
+
+        self.Person.drop_collection()
+        BlogPost.drop_collection()
+
+
     def tearDown(self):
         self.Person.drop_collection()
 

tests/fields.py

@@ -523,29 +523,29 @@ class FieldTest(unittest.TestCase):
         Link.drop_collection()
         Post.drop_collection()
         Bookmark.drop_collection()
-    
+
         link_1 = Link(title="Pitchfork")
         link_1.save()
-    
+
         post_1 = Post(title="Behind the Scenes of the Pavement Reunion")
         post_1.save()
-        
+
         bm = Bookmark(bookmark_object=post_1)
         bm.save()
-        
+
         bm = Bookmark.objects(bookmark_object=post_1).first()
-        
+
         self.assertEqual(bm.bookmark_object, post_1)
         self.assertTrue(isinstance(bm.bookmark_object, Post))
-        
+
         bm.bookmark_object = link_1
         bm.save()
-        
+
         bm = Bookmark.objects(bookmark_object=link_1).first()
-        
+
         self.assertEqual(bm.bookmark_object, link_1)
         self.assertTrue(isinstance(bm.bookmark_object, Link))
-    
+
         Link.drop_collection()
         Post.drop_collection()
         Bookmark.drop_collection()
@@ -555,23 +555,23 @@ class FieldTest(unittest.TestCase):
         """
         class Link(Document):
             title = StringField()
-    
+
         class Post(Document):
             title = StringField()
-    
+
         class User(Document):
             bookmarks = ListField(GenericReferenceField())
-    
+
         Link.drop_collection()
         Post.drop_collection()
         User.drop_collection()
-    
+
         link_1 = Link(title="Pitchfork")
         link_1.save()
-    
+
         post_1 = Post(title="Behind the Scenes of the Pavement Reunion")
         post_1.save()
-    
+
         user = User(bookmarks=[post_1, link_1])
         user.save()
         

tests/queryset.py

@@ -351,8 +351,6 @@ class QuerySetTest(unittest.TestCase):
     def test_filter_chaining(self):
         """Ensure filters can be chained together.
         """
-        from datetime import datetime
-
         class BlogPost(Document):
             title = StringField()
             is_published = BooleanField()
@@ -848,6 +846,71 @@ class QuerySetTest(unittest.TestCase):
         self.Person.objects.delete()
         self.assertEqual(len(self.Person.objects), 0)
 
+    def test_reverse_delete_rule_cascade(self):
+        """Ensure cascading deletion of referring documents from the database.
+        """
+        class BlogPost(Document):
+            content = StringField()
+            author = ReferenceField(self.Person, reverse_delete_rule=CASCADE)
+        BlogPost.drop_collection()
+
+        me = self.Person(name='Test User')
+        me.save()
+        someoneelse = self.Person(name='Some-one Else')
+        someoneelse.save()
+
+        BlogPost(content='Watching TV', author=me).save()
+        BlogPost(content='Chilling out', author=me).save()
+        BlogPost(content='Pro Testing', author=someoneelse).save()
+
+        self.assertEqual(3, BlogPost.objects.count())
+        self.Person.objects(name='Test User').delete()
+        self.assertEqual(1, BlogPost.objects.count())
+
+    def test_reverse_delete_rule_nullify(self):
+        """Ensure nullification of references to deleted documents.
+        """
+        class Category(Document):
+            name = StringField()
+
+        class BlogPost(Document):
+            content = StringField()
+            category = ReferenceField(Category, reverse_delete_rule=NULLIFY)
+
+        BlogPost.drop_collection()
+        Category.drop_collection()
+
+        lameness = Category(name='Lameness')
+        lameness.save()
+
+        post = BlogPost(content='Watching TV', category=lameness)
+        post.save()
+
+        self.assertEqual(1, BlogPost.objects.count())
+        self.assertEqual('Lameness', BlogPost.objects.first().category.name)
+        Category.objects.delete()
+        self.assertEqual(1, BlogPost.objects.count())
+        self.assertEqual(None, BlogPost.objects.first().category)
+
+    def test_reverse_delete_rule_deny(self):
+        """Ensure deletion gets denied on documents that still have references
+        to them.
+        """
+        class BlogPost(Document):
+            content = StringField()
+            author = ReferenceField(self.Person, reverse_delete_rule=DENY)
+
+        BlogPost.drop_collection()
+        self.Person.drop_collection()
+
+        me = self.Person(name='Test User')
+        me.save()
+
+        post = BlogPost(content='Watching TV', author=me)
+        post.save()
+
+        self.assertRaises(OperationError, self.Person.objects.delete)
+
     def test_update(self):
         """Ensure that atomic updates work properly.
         """