import json from django.urls import reverse from .. import test from ...categories.models import Category from ...readtracker import poststracker from ...users.test import AuthenticatedUserTestCase from ..models import Post from ..serializers.moderation import POSTS_LIMIT from ..test import patch_category_acl class ThreadPostMergeApiTestCase(AuthenticatedUserTestCase): def setUp(self): super().setUp() self.category = Category.objects.get(slug="first-category") self.thread = test.post_thread(category=self.category) self.post = test.reply_thread(self.thread, poster=self.user) self.api_link = reverse( "misago:api:thread-post-merge", kwargs={"thread_pk": self.thread.pk} ) def test_anonymous_user(self): """you need to authenticate to merge posts""" self.logout_user() response = self.client.post( self.api_link, json.dumps({}), content_type="application/json" ) self.assertEqual(response.status_code, 403) self.assertEqual( response.json(), {"detail": "This action is not available to guests."} ) @patch_category_acl({"can_merge_posts": False}) def test_no_permission(self): """api validates permission to merge""" response = self.client.post( self.api_link, json.dumps({}), content_type="application/json" ) self.assertEqual(response.status_code, 403) self.assertEqual( response.json(), {"detail": "You can't merge posts in this thread."} ) @patch_category_acl({"can_merge_posts": True}) def test_empty_data_json(self): """api handles empty json data""" response = self.client.post( self.api_link, json.dumps({}), content_type="application/json" ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), {"detail": "You have to select at least two posts to merge."}, ) @patch_category_acl({"can_merge_posts": True}) def test_empty_data_form(self): """api handles empty form data""" response = self.client.post(self.api_link, {}) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), {"detail": "You have to select at least two posts to merge."}, ) @patch_category_acl({"can_merge_posts": True}) def test_invalid_data(self): """api handles post that is invalid type""" response = self.client.post( self.api_link, "[]", content_type="application/json" ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), {"detail": "Invalid data. Expected a dictionary, but got list."}, ) response = self.client.post( self.api_link, "123", content_type="application/json" ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), {"detail": "Invalid data. Expected a dictionary, but got int."}, ) response = self.client.post( self.api_link, '"string"', content_type="application/json" ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), {"detail": "Invalid data. Expected a dictionary, but got str."}, ) response = self.client.post( self.api_link, "malformed", content_type="application/json" ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), {"detail": "JSON parse error - Expecting value: line 1 column 1 (char 0)"}, ) @patch_category_acl({"can_merge_posts": True}) def test_no_posts_ids(self): """api rejects no posts ids""" response = self.client.post( self.api_link, json.dumps({"posts": []}), content_type="application/json" ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), {"detail": "You have to select at least two posts to merge."}, ) @patch_category_acl({"can_merge_posts": True}) def test_invalid_posts_data(self): """api handles invalid data""" response = self.client.post( self.api_link, json.dumps({"posts": "string"}), content_type="application/json", ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), {"detail": 'Expected a list of items but got type "str".'} ) @patch_category_acl({"can_merge_posts": True}) def test_invalid_posts_ids(self): """api handles invalid post id""" response = self.client.post( self.api_link, json.dumps({"posts": [1, 2, "string"]}), content_type="application/json", ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), {"detail": "One or more post ids received were invalid."} ) @patch_category_acl({"can_merge_posts": True}) def test_one_post_id(self): """api rejects one post id""" response = self.client.post( self.api_link, json.dumps({"posts": [1]}), content_type="application/json" ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), {"detail": "You have to select at least two posts to merge."}, ) @patch_category_acl({"can_merge_posts": True}) def test_merge_limit(self): """api rejects more posts than merge limit""" response = self.client.post( self.api_link, json.dumps({"posts": list(range(POSTS_LIMIT + 1))}), content_type="application/json", ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), { "detail": "No more than %s posts can be merged at single time." % POSTS_LIMIT }, ) @patch_category_acl({"can_merge_posts": True}) def test_merge_event(self): """api recjects events""" event = test.reply_thread(self.thread, is_event=True, poster=self.user) response = self.client.post( self.api_link, json.dumps({"posts": [self.post.pk, event.pk]}), content_type="application/json", ) self.assertEqual(response.status_code, 400) self.assertEqual(response.json(), {"detail": "Events can't be merged."}) @patch_category_acl({"can_merge_posts": True}) def test_merge_notfound_pk(self): """api recjects nonexistant pk's""" response = self.client.post( self.api_link, json.dumps({"posts": [self.post.pk, self.post.pk * 1000]}), content_type="application/json", ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), {"detail": "One or more posts to merge could not be found."}, ) @patch_category_acl({"can_merge_posts": True}) def test_merge_cross_threads(self): """api recjects attempt to merge with post made in other thread""" other_thread = test.post_thread(category=self.category) other_post = test.reply_thread(other_thread, poster=self.user) response = self.client.post( self.api_link, json.dumps({"posts": [self.post.pk, other_post.pk]}), content_type="application/json", ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), {"detail": "One or more posts to merge could not be found."}, ) @patch_category_acl({"can_merge_posts": True}) def test_merge_authenticated_with_guest_post(self): """api recjects attempt to merge with post made by deleted user""" other_post = test.reply_thread(self.thread) response = self.client.post( self.api_link, json.dumps({"posts": [self.post.pk, other_post.pk]}), content_type="application/json", ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), {"detail": "Posts made by different users can't be merged."}, ) @patch_category_acl({"can_merge_posts": True}) def test_merge_guest_with_authenticated_post(self): """api recjects attempt to merge with post made by deleted user""" other_post = test.reply_thread(self.thread) response = self.client.post( self.api_link, json.dumps({"posts": [other_post.pk, self.post.pk]}), content_type="application/json", ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), {"detail": "Posts made by different users can't be merged."}, ) @patch_category_acl({"can_merge_posts": True}) def test_merge_guest_posts_different_usernames(self): """api recjects attempt to merge posts made by different guests""" response = self.client.post( self.api_link, json.dumps( { "posts": [ test.reply_thread(self.thread, poster="Bob").pk, test.reply_thread(self.thread, poster="Miku").pk, ] } ), content_type="application/json", ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), {"detail": "Posts made by different users can't be merged."}, ) @patch_category_acl({"can_merge_posts": True, "can_hide_posts": 1}) def test_merge_different_visibility(self): """api recjects attempt to merge posts with different visibility""" response = self.client.post( self.api_link, json.dumps( { "posts": [ test.reply_thread( self.thread, poster=self.user, is_hidden=True ).pk, test.reply_thread( self.thread, poster=self.user, is_hidden=False ).pk, ] } ), content_type="application/json", ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), {"detail": "Posts with different visibility can't be merged."}, ) @patch_category_acl({"can_merge_posts": True, "can_approve_content": True}) def test_merge_different_approval(self): """api recjects attempt to merge posts with different approval""" response = self.client.post( self.api_link, json.dumps( { "posts": [ test.reply_thread( self.thread, poster=self.user, is_unapproved=True ).pk, test.reply_thread( self.thread, poster=self.user, is_unapproved=False ).pk, ] } ), content_type="application/json", ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), {"detail": "Posts with different visibility can't be merged."}, ) @patch_category_acl({"can_merge_posts": True, "can_close_threads": False}) def test_closed_thread_no_permission(self): """api validates permission to merge in closed thread""" self.thread.is_closed = True self.thread.save() posts = [ test.reply_thread(self.thread, poster=self.user).pk, test.reply_thread(self.thread, poster=self.user).pk, ] response = self.client.post( self.api_link, json.dumps({"posts": posts}), content_type="application/json" ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), {"detail": "This thread is closed. You can't merge posts in it."}, ) @patch_category_acl({"can_merge_posts": True, "can_close_threads": True}) def test_closed_thread(self): """api validates permission to merge in closed thread""" self.thread.is_closed = True self.thread.save() posts = [ test.reply_thread(self.thread, poster=self.user).pk, test.reply_thread(self.thread, poster=self.user).pk, ] response = self.client.post( self.api_link, json.dumps({"posts": posts}), content_type="application/json" ) self.assertEqual(response.status_code, 200) @patch_category_acl({"can_merge_posts": True, "can_close_threads": False}) def test_closed_category_no_permission(self): """api validates permission to merge in closed category""" self.category.is_closed = True self.category.save() posts = [ test.reply_thread(self.thread, poster=self.user).pk, test.reply_thread(self.thread, poster=self.user).pk, ] response = self.client.post( self.api_link, json.dumps({"posts": posts}), content_type="application/json" ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), {"detail": "This category is closed. You can't merge posts in it."}, ) @patch_category_acl({"can_merge_posts": True, "can_close_threads": True}) def test_closed_category(self): """api validates permission to merge in closed category""" self.category.is_closed = True self.category.save() posts = [ test.reply_thread(self.thread, poster=self.user).pk, test.reply_thread(self.thread, poster=self.user).pk, ] response = self.client.post( self.api_link, json.dumps({"posts": posts}), content_type="application/json" ) self.assertEqual(response.status_code, 200) @patch_category_acl({"can_merge_posts": True}) def test_merge_best_answer_first_post(self): """api recjects attempt to merge best_answer with first post""" self.thread.first_post.poster = self.user self.thread.first_post.save() self.post.poster = self.user self.post.save() self.thread.set_best_answer(self.user, self.post) self.thread.save() response = self.client.post( self.api_link, json.dumps({"posts": [self.thread.first_post.pk, self.post.pk]}), content_type="application/json", ) self.assertEqual(response.status_code, 400) self.assertEqual( response.json(), { "detail": "Post marked as best answer can't be merged with thread's first post." }, ) @patch_category_acl({"can_merge_posts": True}) def test_merge_posts(self): """api merges two posts""" post_a = test.reply_thread(self.thread, poster=self.user, message="Battęry") post_b = test.reply_thread(self.thread, poster=self.user, message="Hórse") thread_replies = self.thread.replies response = self.client.post( self.api_link, json.dumps({"posts": [post_a.pk, post_b.pk]}), content_type="application/json", ) self.assertEqual(response.status_code, 200) self.thread.refresh_from_db() self.assertEqual(self.thread.replies, thread_replies - 1) with self.assertRaises(Post.DoesNotExist): Post.objects.get(pk=post_b.pk) merged_post = Post.objects.get(pk=post_a.pk) self.assertEqual(merged_post.parsed, "%s\n%s" % (post_a.parsed, post_b.parsed)) @patch_category_acl({"can_merge_posts": True}) def test_merge_guest_posts(self): """api recjects attempt to merge posts made by same guest""" response = self.client.post( self.api_link, json.dumps( { "posts": [ test.reply_thread(self.thread, poster="Bob").pk, test.reply_thread(self.thread, poster="Bob").pk, ] } ), content_type="application/json", ) self.assertEqual(response.status_code, 200) @patch_category_acl({"can_merge_posts": True, "can_hide_posts": 1}) def test_merge_hidden_posts(self): """api merges two hidden posts""" response = self.client.post( self.api_link, json.dumps( { "posts": [ test.reply_thread( self.thread, poster=self.user, is_hidden=True ).pk, test.reply_thread( self.thread, poster=self.user, is_hidden=True ).pk, ] } ), content_type="application/json", ) self.assertEqual(response.status_code, 200) @patch_category_acl({"can_merge_posts": True, "can_approve_content": True}) def test_merge_unapproved_posts(self): """api merges two unapproved posts""" response = self.client.post( self.api_link, json.dumps( { "posts": [ test.reply_thread( self.thread, poster=self.user, is_unapproved=True ).pk, test.reply_thread( self.thread, poster=self.user, is_unapproved=True ).pk, ] } ), content_type="application/json", ) self.assertEqual(response.status_code, 200) @patch_category_acl({"can_merge_posts": True, "can_hide_threads": True}) def test_merge_with_hidden_thread(self): """api excludes thread's first post from visibility checks""" self.thread.first_post.is_hidden = True self.thread.first_post.poster = self.user self.thread.first_post.save() post_visible = test.reply_thread(self.thread, poster=self.user, is_hidden=False) response = self.client.post( self.api_link, json.dumps({"posts": [self.thread.first_post.pk, post_visible.pk]}), content_type="application/json", ) self.assertEqual(response.status_code, 200) @patch_category_acl({"can_merge_posts": True}) def test_merge_protected(self): """api preserves protected status after merge""" response = self.client.post( self.api_link, json.dumps( { "posts": [ test.reply_thread( self.thread, poster="Bob", is_protected=True ).pk, test.reply_thread( self.thread, poster="Bob", is_protected=False ).pk, ] } ), content_type="application/json", ) self.assertEqual(response.status_code, 200) merged_post = self.thread.post_set.order_by("-id")[0] self.assertTrue(merged_post.is_protected) @patch_category_acl({"can_merge_posts": True}) def test_merge_best_answer(self): """api merges best answer with other post""" best_answer = test.reply_thread(self.thread, poster="Bob") self.thread.set_best_answer(self.user, best_answer) self.thread.save() response = self.client.post( self.api_link, json.dumps( { "posts": [ best_answer.pk, test.reply_thread(self.thread, poster="Bob").pk, ] } ), content_type="application/json", ) self.assertEqual(response.status_code, 200) self.thread.refresh_from_db() self.assertEqual(self.thread.best_answer, best_answer) @patch_category_acl({"can_merge_posts": True}) def test_merge_best_answer_in(self): """api merges best answer into other post""" other_post = test.reply_thread(self.thread, poster="Bob") best_answer = test.reply_thread(self.thread, poster="Bob") self.thread.set_best_answer(self.user, best_answer) self.thread.save() response = self.client.post( self.api_link, json.dumps({"posts": [best_answer.pk, other_post.pk]}), content_type="application/json", ) self.assertEqual(response.status_code, 200) self.thread.refresh_from_db() self.assertEqual(self.thread.best_answer, other_post) @patch_category_acl({"can_merge_posts": True}) def test_merge_best_answer_in_protected(self): """api merges best answer into protected post""" best_answer = test.reply_thread(self.thread, poster="Bob") self.thread.set_best_answer(self.user, best_answer) self.thread.save() response = self.client.post( self.api_link, json.dumps( { "posts": [ best_answer.pk, test.reply_thread( self.thread, poster="Bob", is_protected=True ).pk, ] } ), content_type="application/json", ) self.assertEqual(response.status_code, 200) self.thread.refresh_from_db() self.assertEqual(self.thread.best_answer, best_answer) self.thread.best_answer.refresh_from_db() self.assertTrue(self.thread.best_answer.is_protected) self.assertTrue(self.thread.best_answer_is_protected) @patch_category_acl({"can_merge_posts": True}) def test_merge_remove_reads(self): """two posts merge removes read tracker from post""" post_a = test.reply_thread(self.thread, poster=self.user, message="Battęry") post_b = test.reply_thread(self.thread, poster=self.user, message="Hórse") poststracker.save_read(self.user, post_a) poststracker.save_read(self.user, post_b) response = self.client.post( self.api_link, json.dumps({"posts": [post_a.pk, post_b.pk]}), content_type="application/json", ) self.assertEqual(response.status_code, 200) # both post's were removed from readtracker self.assertEqual(self.user.postread_set.count(), 0)