wip #884: bulk delete threads api

misago/threads/api/threadendpoints/delete.py

@@ -38,7 +38,7 @@ def delete_bulk(request, viewmodel):
                 'error': text_type(e)
             })
         except Http404:
-            pass # ignore invisible threads
+            pass # skip invisible threads
 
     return Response(errors)
 
@@ -58,6 +58,7 @@ def clean_threads_ids(request):
             DELETE_LIMIT,
         )
         raise PermissionDenied(message % {'limit': DELETE_LIMIT})
+
     return set(threads_ids)
 
 

misago/threads/tests/test_thread_postbulkdelete_api.py

@@ -44,7 +44,10 @@ class PostBulkDeleteApiTests(ThreadsApiTestCase):
 
     def test_validate_ids(self):
         """api validates that ids are list of ints"""
-        self.override_acl({'can_hide_own_posts': 2, 'can_hide_posts': 2})
+        self.override_acl({
+            'can_hide_own_posts': 2,
+            'can_hide_posts': 2,
+        })
 
         response = self.delete(self.api_link, True)
         self.assertContains(response, "One or more post ids received were invalid.", status_code=403)
@@ -57,21 +60,30 @@ class PostBulkDeleteApiTests(ThreadsApiTestCase):
 
     def test_validate_ids_length(self):
         """api validates that ids are list of ints"""
-        self.override_acl({'can_hide_own_posts': 2, 'can_hide_posts': 2})
+        self.override_acl({
+            'can_hide_own_posts': 2,
+            'can_hide_posts': 2,
+        })
 
         response = self.delete(self.api_link, list(range(100)))
         self.assertContains(response, "No more than 24 posts can be deleted at single time.", status_code=403)
 
     def test_validate_posts_exist(self):
         """api validates that ids are visible posts"""
-        self.override_acl({'can_hide_own_posts': 2, 'can_hide_posts': 0})
+        self.override_acl({
+            'can_hide_own_posts': 2,
+            'can_hide_posts': 0,
+        })
 
         response = self.delete(self.api_link, [p.id * 10 for p in self.posts])
         self.assertContains(response, "One or more posts to delete could not be found.", status_code=403)
 
     def test_validate_posts_visibility(self):
         """api validates that ids are visible posts"""
-        self.override_acl({'can_hide_own_posts': 2, 'can_hide_posts': 0})
+        self.override_acl({
+            'can_hide_own_posts': 2,
+            'can_hide_posts': 0,
+        })
 
         self.posts[1].is_unapproved = True
         self.posts[1].save()
@@ -81,7 +93,10 @@ class PostBulkDeleteApiTests(ThreadsApiTestCase):
 
     def test_validate_posts_same_thread(self):
         """api validates that ids are visible posts"""
-        self.override_acl({'can_hide_own_posts': 2, 'can_hide_posts': 2})
+        self.override_acl({
+            'can_hide_own_posts': 2,
+            'can_hide_posts': 2,
+        })
 
         other_thread = testutils.post_thread(category=self.category)
         self.posts.append(testutils.reply_thread(other_thread, poster=self.user))
@@ -91,7 +106,10 @@ class PostBulkDeleteApiTests(ThreadsApiTestCase):
 
     def test_no_permission(self):
         """api validates permission to delete"""
-        self.override_acl({'can_hide_own_posts': 1, 'can_hide_posts': 1})
+        self.override_acl({
+            'can_hide_own_posts': 1,
+            'can_hide_posts': 1,
+        })
 
         response = self.delete(self.api_link, [p.id for p in self.posts])
         self.assertContains(response, "You can't delete posts in this category.", status_code=403)
@@ -173,7 +191,10 @@ class PostBulkDeleteApiTests(ThreadsApiTestCase):
 
     def test_delete_first_post(self):
         """api disallows first post's deletion"""
-        self.override_acl({'can_hide_own_posts': 2, 'can_hide_posts': 2})
+        self.override_acl({
+            'can_hide_own_posts': 2,
+            'can_hide_posts': 2,
+        })
 
         ids = [p.id for p in self.posts]
         ids.append(self.thread.first_post_id)
@@ -215,7 +236,10 @@ class PostBulkDeleteApiTests(ThreadsApiTestCase):
 
     def test_delete_posts(self):
         """api deletes thread posts"""
-        self.override_acl({'can_hide_own_posts': 0, 'can_hide_posts': 2})
+        self.override_acl({
+            'can_hide_own_posts': 0,
+            'can_hide_posts': 2,
+        })
 
         response = self.delete(self.api_link, [p.id for p in self.posts])
         self.assertEqual(response.status_code, 200)

misago/threads/tests/test_threads_bulkdelete_api.py

@@ -0,0 +1,173 @@
+import json
+
+from django.urls import reverse
+
+from misago.acl.testutils import override_acl
+from misago.categories import PRIVATE_THREADS_ROOT_NAME
+from misago.categories.models import Category
+from misago.threads import testutils
+from misago.threads.models import Thread
+from misago.threads.threadtypes import trees_map
+
+from .test_threads_api import ThreadsApiTestCase
+
+
+class ThreadsBulkDeleteApiTests(ThreadsApiTestCase):
+    def setUp(self):
+        super(ThreadsBulkDeleteApiTests, self).setUp()
+
+        self.api_link = reverse('misago:api:thread-list')
+
+        self.threads = [
+            testutils.post_thread(
+                category=self.category,
+                poster=self.user,
+            ),
+            testutils.post_thread(category=self.category),
+            testutils.post_thread(
+                category=self.category,
+                poster=self.user,
+            ),
+        ]
+
+    def delete(self, url, data=None):
+        return self.client.delete(url, json.dumps(data), content_type="application/json")
+
+    def test_delete_anonymous(self):
+        """anonymous users can't bulk delete threads"""
+        self.logout_user()
+
+        response = self.delete(self.api_link)
+        self.assertContains(response, "This action is not available to guests.", status_code=403)
+
+    def test_delete_no_ids(self):
+        """api requires ids to delete"""
+        self.override_acl({
+            'can_hide_own_threads': 0,
+            'can_hide_threads': 0,
+        })
+
+        response = self.delete(self.api_link)
+        self.assertContains(response, "You have to specify at least one thread to delete.", status_code=403)
+
+    def test_validate_ids(self):
+        """api validates that ids are list of ints"""
+        self.override_acl({
+            'can_hide_own_threads': 2,
+            'can_hide_threads': 2,
+        })
+
+        response = self.delete(self.api_link, True)
+        self.assertContains(response, "One or more thread ids received were invalid.", status_code=403)
+
+        response = self.delete(self.api_link, 'abbss')
+        self.assertContains(response, "One or more thread ids received were invalid.", status_code=403)
+
+        response = self.delete(self.api_link, [1, 2, 3, 'a', 'b', 'x'])
+        self.assertContains(response, "One or more thread ids received were invalid.", status_code=403)
+
+    def test_validate_ids_length(self):
+        """api validates that ids are list of ints"""
+        self.override_acl({
+            'can_hide_own_threads': 2,
+            'can_hide_threads': 2,
+        })
+
+        response = self.delete(self.api_link, list(range(100)))
+        self.assertContains(response, "No more than 40 threads can be deleted at single time.", status_code=403)
+
+    def test_validate_thread_visibility(self):
+        """api valdiates if user can see deleted thread"""
+        self.override_acl({
+            'can_hide_own_threads': 2,
+            'can_hide_threads': 2,
+        })
+
+        unapproved_thread = self.threads[1]
+
+        unapproved_thread.is_unapproved = True
+        unapproved_thread.save()
+
+        threads_ids = [p.id for p in self.threads]
+
+        response = self.delete(self.api_link, threads_ids)
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.json(), [])
+
+        # unapproved thread wasn't deleted
+        Thread.objects.get(pk=unapproved_thread.pk)
+
+        deleted_threads = [self.threads[0], self.threads[2]]
+        for thread in deleted_threads:
+            with self.assertRaises(Thread.DoesNotExist):
+                Thread.objects.get(pk=thread.pk)
+
+        category = Category.objects.get(pk=self.category.pk)
+        self.assertNotIn(category.last_thread_id, threads_ids)
+
+    def test_delete_other_user_thread_no_permission(self):
+        """api valdiates if user can delete other users threads"""
+        self.override_acl({
+            'can_hide_own_threads': 2,
+            'can_hide_threads': 0,
+        })
+
+        response = self.delete(self.api_link, [p.id for p in self.threads])
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.json(), [
+            {
+                'thread': {
+                    "id": self.threads[1].pk,
+                    "title": self.threads[1].title
+                },
+                'error': "You don't have permission to delete this thread."
+            }
+        ])
+
+        Thread.objects.get(pk=self.threads[1].pk)
+
+        deleted_threads = [self.threads[0], self.threads[2]]
+        for thread in deleted_threads:
+            with self.assertRaises(Thread.DoesNotExist):
+                Thread.objects.get(pk=thread.pk)
+
+        category = Category.objects.get(pk=self.category.pk)
+        self.assertEqual(category.last_thread_id, self.threads[1].pk)
+
+    def test_delete_private_thread(self):
+        """attempt to delete private thread fails"""
+        private_thread = self.threads[0]
+
+        private_thread.category = Category.objects.get(
+            tree_id=trees_map.get_tree_id_for_root(PRIVATE_THREADS_ROOT_NAME),
+        )
+        private_thread.save()
+
+        private_thread.threadparticipant_set.create(
+            user=self.user,
+            is_owner=True,
+        )
+
+        self.override_acl({
+            'can_hide_own_threads': 2,
+            'can_hide_threads': 2,
+        })
+
+        threads_ids = [p.id for p in self.threads]
+
+        response = self.delete(self.api_link, threads_ids)
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.json(), [])
+
+        Thread.objects.get(pk=private_thread.pk)
+
+        deleted_threads = [self.threads[1], self.threads[2]]
+        for thread in deleted_threads:
+            with self.assertRaises(Thread.DoesNotExist):
+                Thread.objects.get(pk=thread.pk)
+
+        category = Category.objects.get(pk=self.category.pk)
+        self.assertNotIn(category.last_thread_id, threads_ids)

misago/threads/utils.py

@@ -72,3 +72,4 @@ def get_thread_id_from_url(request, url):
         return int(resolution.kwargs.get(kwargname))
     except (TypeError, ValueError):
         return None
+