Misago/misago/threads/tests/test_thread_postsplit_api.py

import json

from django.urls import reverse

from misago.categories.models import Category
from misago.readtracker import poststracker
from misago.threads import testutils
from misago.threads.models import Post, Thread
from misago.threads.serializers.moderation import POSTS_LIMIT
from misago.threads.test import patch_category_acl, patch_other_category_acl
from misago.users.testutils import AuthenticatedUserTestCase


class ThreadPostSplitApiTestCase(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).pk,
            testutils.reply_thread(self.thread).pk,
        ]

        self.api_link = reverse(
            "misago:api:thread-post-split", kwargs={"thread_pk": self.thread.pk}
        )

        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")

    def test_anonymous_user(self):
        """you need to authenticate to split 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_move_posts": False})
    def test_no_permission(self):
        """api validates permission to split"""
        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 split posts from this thread."}
        )

    @patch_category_acl({"can_move_posts": True})
    def test_empty_data(self):
        """api handles empty data"""
        response = self.client.post(self.api_link)
        self.assertEqual(response.status_code, 400)
        self.assertEqual(
            response.json(),
            {"detail": "You have to specify at least one post to split."},
        )

    @patch_category_acl({"can_move_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(),
            {
                "non_field_errors": [
                    "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(),
            {"non_field_errors": ["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(),
            {"non_field_errors": ["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_move_posts": True})
    def test_no_posts_ids(self):
        """api rejects no posts ids"""
        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 specify at least one post to split."},
        )

    @patch_category_acl({"can_move_posts": True})
    def test_empty_posts_ids(self):
        """api rejects empty posts ids list"""
        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 specify at least one post to split."},
        )

    @patch_category_acl({"can_move_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_move_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_move_posts": True})
    def test_split_limit(self):
        """api rejects more posts than split 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 split at single time."
                % POSTS_LIMIT
            },
        )

    @patch_category_acl({"can_move_posts": True})
    def test_split_invisible(self):
        """api validates posts visibility"""
        response = self.client.post(
            self.api_link,
            json.dumps(
                {"posts": [testutils.reply_thread(self.thread, is_unapproved=True).pk]}
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 400)
        self.assertEqual(
            response.json(),
            {"detail": "One or more posts to split could not be found."},
        )

    @patch_category_acl({"can_move_posts": True})
    def test_split_event(self):
        """api rejects events split"""
        response = self.client.post(
            self.api_link,
            json.dumps(
                {"posts": [testutils.reply_thread(self.thread, is_event=True).pk]}
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 400)
        self.assertEqual(response.json(), {"detail": "Events can't be split."})

    @patch_category_acl({"can_move_posts": True})
    def test_split_first_post(self):
        """api rejects first post split"""
        response = self.client.post(
            self.api_link,
            json.dumps({"posts": [self.thread.first_post_id]}),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 400)
        self.assertEqual(
            response.json(), {"detail": "You can't split thread's first post."}
        )

    @patch_category_acl({"can_move_posts": True})
    def test_split_hidden_posts(self):
        """api recjects attempt to split urneadable hidden post"""
        response = self.client.post(
            self.api_link,
            json.dumps(
                {"posts": [testutils.reply_thread(self.thread, is_hidden=True).pk]}
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 400)
        self.assertEqual(
            response.json(),
            {"detail": "You can't split posts the content you can't see."},
        )

    @patch_category_acl({"can_move_posts": True, "can_close_threads": False})
    def test_split_posts_closed_thread_no_permission(self):
        """api recjects attempt to split posts from closed thread"""
        self.thread.is_closed = True
        self.thread.save()

        response = self.client.post(
            self.api_link,
            json.dumps({"posts": [testutils.reply_thread(self.thread).pk]}),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 400)
        self.assertEqual(
            response.json(),
            {"detail": "This thread is closed. You can't split posts in it."},
        )

    @patch_category_acl({"can_move_posts": True, "can_close_threads": False})
    def test_split_posts_closed_category_no_permission(self):
        """api recjects attempt to split posts from closed thread"""
        self.category.is_closed = True
        self.category.save()

        response = self.client.post(
            self.api_link,
            json.dumps({"posts": [testutils.reply_thread(self.thread).pk]}),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 400)
        self.assertEqual(
            response.json(),
            {"detail": "This category is closed. You can't split posts in it."},
        )

    @patch_category_acl({"can_move_posts": True})
    def test_split_other_thread_posts(self):
        """api recjects attempt to split other thread's post"""
        other_thread = testutils.post_thread(self.category)

        response = self.client.post(
            self.api_link,
            json.dumps(
                {"posts": [testutils.reply_thread(other_thread, is_hidden=True).pk]}
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 400)
        self.assertEqual(
            response.json(),
            {"detail": "One or more posts to split could not be found."},
        )

    @patch_category_acl({"can_move_posts": True})
    def test_split_empty_new_thread_data(self):
        """api handles empty form data"""
        response = self.client.post(
            self.api_link,
            json.dumps({"posts": self.posts}),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 400)

        response_json = response.json()
        self.assertEqual(
            response_json,
            {
                "title": ["This field is required."],
                "category": ["This field is required."],
            },
        )

    @patch_category_acl({"can_move_posts": True})
    def test_split_invalid_final_title(self):
        """api rejects split because final thread title was invalid"""
        response = self.client.post(
            self.api_link,
            json.dumps(
                {"posts": self.posts, "title": "$$$", "category": self.category.id}
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 400)

        response_json = response.json()
        self.assertEqual(
            response_json,
            {
                "title": [
                    "Thread title should be at least 5 characters long (it has 3)."
                ]
            },
        )

    @patch_other_category_acl({"can_see": False})
    @patch_category_acl({"can_move_posts": True})
    def test_split_invalid_category(self):
        """api rejects split because final category was invalid"""
        response = self.client.post(
            self.api_link,
            json.dumps(
                {
                    "posts": self.posts,
                    "title": "Valid thread title",
                    "category": self.other_category.id,
                }
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 400)

        response_json = response.json()
        self.assertEqual(
            response_json, {"category": ["Requested category could not be found."]}
        )

    @patch_category_acl({"can_move_posts": True, "can_start_threads": False})
    def test_split_unallowed_start_thread(self):
        """api rejects split because category isn't allowing starting threads"""
        response = self.client.post(
            self.api_link,
            json.dumps(
                {
                    "posts": self.posts,
                    "title": "Valid thread title",
                    "category": self.category.id,
                }
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 400)

        response_json = response.json()
        self.assertEqual(
            response_json,
            {"category": ["You can't create new threads in selected category."]},
        )

    @patch_category_acl({"can_move_posts": True})
    def test_split_invalid_weight(self):
        """api rejects split because final weight was invalid"""
        response = self.client.post(
            self.api_link,
            json.dumps(
                {
                    "posts": self.posts,
                    "title": "Valid thread title",
                    "category": self.category.id,
                    "weight": 4,
                }
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 400)

        response_json = response.json()
        self.assertEqual(
            response_json, {"weight": ["Ensure this value is less than or equal to 2."]}
        )

    @patch_category_acl({"can_move_posts": True})
    def test_split_unallowed_global_weight(self):
        """api rejects split because global weight was unallowed"""
        response = self.client.post(
            self.api_link,
            json.dumps(
                {
                    "posts": self.posts,
                    "title": "Valid thread title",
                    "category": self.category.id,
                    "weight": 2,
                }
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 400)

        response_json = response.json()
        self.assertEqual(
            response_json,
            {
                "weight": [
                    "You don't have permission to pin threads globally in this category."
                ]
            },
        )

    @patch_category_acl({"can_move_posts": True, "can_pin_threads": 0})
    def test_split_unallowed_local_weight(self):
        """api rejects split because local weight was unallowed"""
        response = self.client.post(
            self.api_link,
            json.dumps(
                {
                    "posts": self.posts,
                    "title": "Valid thread title",
                    "category": self.category.id,
                    "weight": 1,
                }
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 400)

        response_json = response.json()
        self.assertEqual(
            response_json,
            {"weight": ["You don't have permission to pin threads in this category."]},
        )

    @patch_category_acl({"can_move_posts": True, "can_pin_threads": 1})
    def test_split_allowed_local_weight(self):
        """api allows local weight"""
        response = self.client.post(
            self.api_link,
            json.dumps(
                {
                    "posts": self.posts,
                    "title": "$$$",
                    "category": self.category.id,
                    "weight": 1,
                }
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 400)

        response_json = response.json()
        self.assertEqual(
            response_json,
            {
                "title": [
                    "Thread title should be at least 5 characters long (it has 3)."
                ]
            },
        )

    @patch_category_acl({"can_move_posts": True, "can_pin_threads": 2})
    def test_split_allowed_global_weight(self):
        """api allows global weight"""
        response = self.client.post(
            self.api_link,
            json.dumps(
                {
                    "posts": self.posts,
                    "title": "$$$",
                    "category": self.category.id,
                    "weight": 2,
                }
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 400)

        response_json = response.json()
        self.assertEqual(
            response_json,
            {
                "title": [
                    "Thread title should be at least 5 characters long (it has 3)."
                ]
            },
        )

    @patch_category_acl({"can_move_posts": True, "can_close_threads": False})
    def test_split_unallowed_close(self):
        """api rejects split because closing thread was unallowed"""
        response = self.client.post(
            self.api_link,
            json.dumps(
                {
                    "posts": self.posts,
                    "title": "Valid thread title",
                    "category": self.category.id,
                    "is_closed": True,
                }
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 400)

        response_json = response.json()
        self.assertEqual(
            response_json,
            {
                "is_closed": [
                    "You don't have permission to close threads in this category."
                ]
            },
        )

    @patch_category_acl({"can_move_posts": True, "can_close_threads": True})
    def test_split_with_close(self):
        """api allows for closing thread"""
        response = self.client.post(
            self.api_link,
            json.dumps(
                {
                    "posts": self.posts,
                    "title": "$$$",
                    "category": self.category.id,
                    "weight": 0,
                    "is_closed": True,
                }
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 400)

        response_json = response.json()
        self.assertEqual(
            response_json,
            {
                "title": [
                    "Thread title should be at least 5 characters long (it has 3)."
                ]
            },
        )

    @patch_category_acl({"can_move_posts": True, "can_hide_threads": 0})
    def test_split_unallowed_hidden(self):
        """api rejects split because hidden thread was unallowed"""
        response = self.client.post(
            self.api_link,
            json.dumps(
                {
                    "posts": self.posts,
                    "title": "Valid thread title",
                    "category": self.category.id,
                    "is_hidden": True,
                }
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 400)

        response_json = response.json()
        self.assertEqual(
            response_json,
            {
                "is_hidden": [
                    "You don't have permission to hide threads in this category."
                ]
            },
        )

    @patch_category_acl({"can_move_posts": True, "can_hide_threads": 1})
    def test_split_with_hide(self):
        """api allows for hiding thread"""
        response = self.client.post(
            self.api_link,
            json.dumps(
                {
                    "posts": self.posts,
                    "title": "$$$",
                    "category": self.category.id,
                    "weight": 0,
                    "is_hidden": True,
                }
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 400)

        response_json = response.json()
        self.assertEqual(
            response_json,
            {
                "title": [
                    "Thread title should be at least 5 characters long (it has 3)."
                ]
            },
        )

    @patch_category_acl({"can_move_posts": True})
    def test_split(self):
        """api splits posts to new thread"""
        self.thread.refresh_from_db()
        self.assertEqual(self.thread.replies, 2)

        response = self.client.post(
            self.api_link,
            json.dumps(
                {
                    "posts": self.posts,
                    "title": "Split thread.",
                    "category": self.category.id,
                }
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 200)

        # thread was created
        split_thread = self.category.thread_set.get(slug="split-thread")
        self.assertEqual(split_thread.replies, 1)

        # posts were removed from old thread
        self.thread.refresh_from_db()
        self.assertEqual(self.thread.replies, 0)

        # posts were moved to new thread
        self.assertEqual(split_thread.post_set.filter(pk__in=self.posts).count(), 2)

    @patch_category_acl({"can_move_posts": True})
    def test_split_best_answer(self):
        """api splits best answer to new thread"""
        best_answer = testutils.reply_thread(self.thread)

        self.thread.set_best_answer(self.user, best_answer)
        self.thread.synchronize()
        self.thread.save()

        self.thread.refresh_from_db()
        self.assertEqual(self.thread.best_answer, best_answer)
        self.assertEqual(self.thread.replies, 3)

        response = self.client.post(
            self.api_link,
            json.dumps(
                {
                    "posts": [best_answer.pk],
                    "title": "Split thread.",
                    "category": self.category.id,
                }
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 200)

        # best_answer was moved and unmarked
        self.thread.refresh_from_db()
        self.assertEqual(self.thread.replies, 2)
        self.assertIsNone(self.thread.best_answer)

        split_thread = self.category.thread_set.get(slug="split-thread")
        self.assertEqual(split_thread.replies, 0)
        self.assertIsNone(split_thread.best_answer)

    @patch_other_category_acl(
        {
            "can_start_threads": True,
            "can_close_threads": True,
            "can_hide_threads": True,
            "can_pin_threads": 2,
        }
    )
    @patch_category_acl({"can_move_posts": True})
    def test_split_kitchensink(self):
        """api splits posts with kitchensink"""
        self.thread.refresh_from_db()
        self.assertEqual(self.thread.replies, 2)

        poststracker.save_read(self.user, self.thread.first_post)
        for post in self.posts:
            poststracker.save_read(
                self.user, Post.objects.select_related().get(pk=post)
            )

        response = self.client.post(
            self.api_link,
            json.dumps(
                {
                    "posts": self.posts,
                    "title": "Split thread",
                    "category": self.other_category.id,
                    "weight": 2,
                    "is_closed": 1,
                    "is_hidden": 1,
                }
            ),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 200)

        # thread was created
        split_thread = self.other_category.thread_set.get(slug="split-thread")
        self.assertEqual(split_thread.replies, 1)
        self.assertEqual(split_thread.weight, 2)
        self.assertTrue(split_thread.is_closed)
        self.assertTrue(split_thread.is_hidden)

        # posts were removed from old thread
        self.thread.refresh_from_db()
        self.assertEqual(self.thread.replies, 0)

        # posts were moved to new thread
        self.assertEqual(split_thread.post_set.filter(pk__in=self.posts).count(), 2)

        # postreads were removed
        postreads = self.user.postread_set.filter(post__is_event=False).order_by("id")

        postreads_threads = list(postreads.values_list("thread_id", flat=True))
        self.assertEqual(postreads_threads, [self.thread.pk])

        postreads_categories = list(postreads.values_list("category_id", flat=True))
        self.assertEqual(postreads_categories, [self.category.pk])