import json from django.urls import reverse from misago.categories.models import Category from misago.threads import testutils from misago.threads.models import Thread from misago.threads.test import patch_category_acl, patch_other_category_acl from .test_threads_api import ThreadsApiTestCase class ThreadsBulkPatchApiTestCase(ThreadsApiTestCase): def setUp(self): super().setUp() self.threads = list(reversed([ testutils.post_thread(category=self.category), testutils.post_thread(category=self.category), testutils.post_thread(category=self.category), ])) self.ids = list(reversed([t.id for t in self.threads])) self.api_link = reverse('misago:api:thread-list') def patch(self, api_link, ops): return self.client.patch(api_link, json.dumps(ops), content_type="application/json") class BulkPatchSerializerTests(ThreadsBulkPatchApiTestCase): 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 40 elements.", ], 'ops': [ "Ensure this field has no more than 10 elements.", ], }) def test_threads_not_found(self): """api fails to find threads""" threads = [ testutils.post_thread(category=self.category, is_hidden=True), testutils.post_thread(category=self.category, is_unapproved=True), ] response = self.patch(self.api_link, { 'ids': [t.id for t in threads], 'ops': [{}], }) self.assertEqual(response.status_code, 403) self.assertEqual(response.json(), { 'detail': "One or more threads 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) class ThreadAddAclApiTests(ThreadsBulkPatchApiTestCase): def test_add_acl_true(self): """api adds current threads acl 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, thread in enumerate(self.threads): self.assertEqual(response_json[i]['id'], thread.id) self.assertTrue(response_json[i]['acl']) class BulkThreadChangeTitleApiTests(ThreadsBulkPatchApiTestCase): @patch_category_acl({"can_edit_threads": 2}) def test_change_thread_title(self): """api changes thread title and resyncs the category""" response = self.patch( self.api_link, { 'ids': self.ids, 'ops': [ { 'op': 'replace', 'path': 'title', 'value': 'Changed the title!', }, ] } ) self.assertEqual(response.status_code, 200) response_json = response.json() for i, thread in enumerate(self.threads): self.assertEqual(response_json[i]['id'], thread.id) self.assertEqual(response_json[i]['title'], 'Changed the title!') for thread in Thread.objects.filter(id__in=self.ids): self.assertEqual(thread.title, 'Changed the title!') category = Category.objects.get(pk=self.category.id) self.assertEqual(category.last_thread_title, 'Changed the title!') @patch_category_acl({"can_edit_threads": 0}) def test_change_thread_title_no_permission(self): """api validates permission to change title, returns errors""" response = self.patch( self.api_link, { 'ids': self.ids, 'ops': [ { 'op': 'replace', 'path': 'title', 'value': 'Changed the title!', }, ] } ) self.assertEqual(response.status_code, 400) response_json = response.json() for i, thread in enumerate(self.threads): self.assertEqual(response_json[i]['id'], thread.id) self.assertEqual( response_json[i]['detail'], ["You can't edit threads in this category."], ) class BulkThreadMoveApiTests(ThreadsBulkPatchApiTestCase): def setUp(self): super().setUp() Category( name='Other Category', slug='other-category', ).insert_at( self.category, position='last-child', save=True, ) self.other_category = Category.objects.get(slug='other-category') @patch_category_acl({"can_move_threads": True}) @patch_other_category_acl({"can_start_threads": 2}) def test_move_thread(self): """api moves threads to other category and syncs both categories""" response = self.patch( self.api_link, { 'ids': self.ids, 'ops': [ { 'op': 'replace', 'path': 'category', 'value': self.other_category.id, }, { 'op': 'replace', 'path': 'flatten-categories', 'value': None, }, ] } ) self.assertEqual(response.status_code, 200) response_json = response.json() for i, thread in enumerate(self.threads): self.assertEqual(response_json[i]['id'], thread.id) self.assertEqual(response_json[i]['category'], self.other_category.id) for thread in Thread.objects.filter(id__in=self.ids): self.assertEqual(thread.category_id, self.other_category.id) category = Category.objects.get(pk=self.category.id) self.assertEqual(category.threads, self.category.threads - 3) new_category = Category.objects.get(pk=self.other_category.id) self.assertEqual(new_category.threads, 3) class BulkThreadsHideApiTests(ThreadsBulkPatchApiTestCase): @patch_category_acl({"can_hide_threads": 1}) def test_hide_thread(self): """api makes it possible to hide thread""" response = self.patch( self.api_link, { 'ids': self.ids, 'ops': [ { 'op': 'replace', 'path': 'is-hidden', 'value': True, }, ] } ) self.assertEqual(response.status_code, 200) response_json = response.json() for i, thread in enumerate(self.threads): self.assertEqual(response_json[i]['id'], thread.id) self.assertTrue(response_json[i]['is_hidden']) for thread in Thread.objects.filter(id__in=self.ids): self.assertTrue(thread.is_hidden) category = Category.objects.get(pk=self.category.id) self.assertNotIn(category.last_thread_id, self.ids) class BulkThreadsApproveApiTests(ThreadsBulkPatchApiTestCase): @patch_category_acl({"can_approve_content": True}) def test_approve_thread(self): """api approvse threads and syncs category""" for thread in self.threads: thread.first_post.is_unapproved = True thread.first_post.save() thread.synchronize() thread.save() self.assertTrue(thread.is_unapproved) self.assertTrue(thread.has_unapproved_posts) self.category.synchronize() self.category.save() 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, thread in enumerate(self.threads): self.assertEqual(response_json[i]['id'], thread.id) self.assertFalse(response_json[i]['is_unapproved']) self.assertFalse(response_json[i]['has_unapproved_posts']) for thread in Thread.objects.filter(id__in=self.ids): self.assertFalse(thread.is_unapproved) self.assertFalse(thread.has_unapproved_posts) category = Category.objects.get(pk=self.category.id) self.assertIn(category.last_thread_id, self.ids)