import json from datetime import timedelta from django.urls import reverse from django.utils import timezone from misago.categories.models import Category from misago.threads import testutils from misago.threads.models import Post, Thread from misago.threads.test import patch_category_acl from misago.users.testutils import AuthenticatedUserTestCase class ThreadPostBulkPatchApiTestCase(AuthenticatedUserTestCase): def setUp(self): super().setUp() self.category = Category.objects.get(slug='first-category') self.thread = testutils.post_thread(category=self.category) self.posts = [ testutils.reply_thread(self.thread, poster=self.user), testutils.reply_thread(self.thread), testutils.reply_thread(self.thread, poster=self.user), ] self.ids = [p.id for p in self.posts] self.api_link = reverse( 'misago:api:thread-post-list', kwargs={ 'thread_pk': self.thread.pk, } ) def patch(self, api_link, ops): return self.client.patch(api_link, json.dumps(ops), content_type="application/json") class BulkPatchSerializerTests(ThreadPostBulkPatchApiTestCase): def test_invalid_input_type(self): """api rejects invalid input type""" response = self.patch(self.api_link, [1, 2, 3]) self.assertEqual(response.status_code, 400) self.assertEqual(response.json(), { 'non_field_errors': [ "Invalid data. Expected a dictionary, but got list.", ], }) def test_missing_input_keys(self): """api rejects input with missing keys""" response = self.patch(self.api_link, {}) self.assertEqual(response.status_code, 400) self.assertEqual(response.json(), { 'ids': [ "This field is required.", ], 'ops': [ "This field is required.", ], }) def test_empty_input_keys(self): """api rejects input with empty keys""" response = self.patch(self.api_link, { 'ids': [], 'ops': [], }) self.assertEqual(response.status_code, 400) self.assertEqual(response.json(), { 'ids': [ "Ensure this field has at least 1 elements.", ], 'ops': [ "Ensure this field has at least 1 elements.", ], }) def test_invalid_input_keys(self): """api rejects input with invalid keys""" response = self.patch(self.api_link, { 'ids': ['a'], 'ops': [1], }) self.assertEqual(response.status_code, 400) self.assertEqual(response.json(), { 'ids': [ "A valid integer is required.", ], 'ops': [ 'Expected a dictionary of items but got type "int".', ], }) def test_too_small_id(self): """api rejects input with implausiple id""" response = self.patch(self.api_link, { 'ids': [0], 'ops': [{}], }) self.assertEqual(response.status_code, 400) self.assertEqual(response.json(), { 'ids': [ "Ensure this value is greater than or equal to 1.", ], }) def test_too_large_input(self): """api rejects too large input""" response = self.patch(self.api_link, { 'ids': [i + 1 for i in range(200)], 'ops': [{} for i in range(200)], }) self.assertEqual(response.status_code, 400) self.assertEqual(response.json(), { 'ids': [ "Ensure this field has no more than 24 elements.", ], 'ops': [ "Ensure this field has no more than 10 elements.", ], }) def test_posts_not_found(self): """api fails to find posts""" posts = [ testutils.reply_thread(self.thread, is_hidden=True), testutils.reply_thread(self.thread, is_unapproved=True), ] response = self.patch(self.api_link, { 'ids': [p.id for p in posts], 'ops': [{}], }) self.assertEqual(response.status_code, 403) self.assertEqual(response.json(), { 'detail': "One or more posts to update could not be found.", }) def test_ops_invalid(self): """api validates descriptions""" response = self.patch(self.api_link, { 'ids': self.ids[:1], 'ops': [{}], }) self.assertEqual(response.status_code, 400) self.assertEqual(response.json(), [ {'id': self.ids[0], 'detail': ['undefined op']}, ]) def test_anonymous_user(self): """anonymous users can't use bulk actions""" self.logout_user() response = self.patch(self.api_link, { 'ids': self.ids[:1], 'ops': [{}], }) self.assertEqual(response.status_code, 403) def test_events(self): """cant use bulk actions for events""" for post in self.posts: post.is_event = True post.save() response = self.patch(self.api_link, { 'ids': self.ids, 'ops': [{}], }) self.assertEqual(response.status_code, 403) self.assertEqual(response.json(), { 'detail': "One or more posts to update could not be found.", }) class PostsAddAclApiTests(ThreadPostBulkPatchApiTestCase): def test_add_acl_true(self): """api adds posts acls to response""" response = self.patch(self.api_link, { 'ids': self.ids, 'ops': [ { 'op': 'add', 'path': 'acl', 'value': True, }, ] }) self.assertEqual(response.status_code, 200) response_json = response.json() for i, post in enumerate(self.posts): self.assertEqual(response_json[i]['id'], post.id) self.assertTrue(response_json[i]['acl']) class BulkPostProtectApiTests(ThreadPostBulkPatchApiTestCase): @patch_category_acl({"can_protect_posts": True, "can_edit_posts": 2}) def test_protect_post(self): """api makes it possible to protect posts""" response = self.patch( self.api_link, { 'ids': self.ids, 'ops': [ { 'op': 'replace', 'path': 'is-protected', 'value': True, }, ] } ) self.assertEqual(response.status_code, 200) response_json = response.json() for i, post in enumerate(self.posts): self.assertEqual(response_json[i]['id'], post.id) self.assertTrue(response_json[i]['is_protected']) for post in Post.objects.filter(id__in=self.ids): self.assertTrue(post.is_protected) @patch_category_acl({"can_protect_posts": False}) def test_protect_post_no_permission(self): """api validates permission to protect posts and returns errors""" response = self.patch( self.api_link, { 'ids': self.ids, 'ops': [ { 'op': 'replace', 'path': 'is-protected', 'value': True, }, ] } ) self.assertEqual(response.status_code, 400) response_json = response.json() for i, post in enumerate(self.posts): self.assertEqual(response_json[i]['id'], post.id) self.assertEqual( response_json[i]['detail'], ["You can't protect posts in this category."], ) for post in Post.objects.filter(id__in=self.ids): self.assertFalse(post.is_protected) class BulkPostsApproveApiTests(ThreadPostBulkPatchApiTestCase): @patch_category_acl({"can_approve_content": True}) def test_approve_post(self): """api resyncs thread and categories on posts approval""" for post in self.posts: post.is_unapproved = True post.save() self.thread.synchronize() self.thread.save() self.assertNotIn(self.thread.last_post_id, self.ids) response = self.patch( self.api_link, { 'ids': self.ids, 'ops': [ { 'op': 'replace', 'path': 'is-unapproved', 'value': False, }, ] } ) self.assertEqual(response.status_code, 200) response_json = response.json() for i, post in enumerate(self.posts): self.assertEqual(response_json[i]['id'], post.id) self.assertFalse(response_json[i]['is_unapproved']) for post in Post.objects.filter(id__in=self.ids): self.assertFalse(post.is_unapproved) thread = Thread.objects.get(pk=self.thread.pk) self.assertIn(thread.last_post_id, self.ids) category = Category.objects.get(pk=self.category.pk) self.assertEqual(category.posts, 4)