Misago/misago/threads/tests/test_threads_editor_api.py

import os

from django.urls import reverse

from .. import test
from ...acl import useracl
from ...acl.objectacl import add_acl_to_obj
from ...categories.models import Category
from ...conftest import get_cache_versions
from ...users.test import AuthenticatedUserTestCase
from ..models import Attachment
from ..serializers import AttachmentSerializer
from ..test import patch_category_acl

TESTFILES_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "testfiles")
TEST_DOCUMENT_PATH = os.path.join(TESTFILES_DIR, "document.pdf")

cache_versions = get_cache_versions()


class EditorApiTestCase(AuthenticatedUserTestCase):
    def setUp(self):
        super().setUp()

        self.category = Category.objects.get(slug="first-category")


class ThreadPostEditorApiTests(EditorApiTestCase):
    def setUp(self):
        super().setUp()

        self.api_link = reverse("misago:api:thread-editor")

    def test_anonymous_user_request(self):
        """endpoint validates if user is authenticated"""
        self.logout_user()

        response = self.client.get(self.api_link)
        self.assertEqual(response.status_code, 403)
        self.assertEqual(
            response.json(), {"detail": "You need to be signed in to start threads."}
        )

    @patch_category_acl({"can_browse": False})
    def test_category_visibility_validation(self):
        """endpoint omits non-browseable categories"""
        response = self.client.get(self.api_link)
        self.assertEqual(response.status_code, 403)
        self.assertEqual(
            response.json(),
            {
                "detail": (
                    "No categories that allow new threads are available "
                    "to you at the moment."
                )
            },
        )

    @patch_category_acl({"can_start_threads": False})
    def test_category_disallowing_new_threads(self):
        """endpoint omits category disallowing starting threads"""
        response = self.client.get(self.api_link)
        self.assertEqual(response.status_code, 403)
        self.assertEqual(
            response.json(),
            {
                "detail": (
                    "No categories that allow new threads are available "
                    "to you at the moment."
                )
            },
        )

    @patch_category_acl({"can_close_threads": False, "can_start_threads": True})
    def test_category_closed_disallowing_new_threads(self):
        """endpoint omits closed category"""
        self.category.is_closed = True
        self.category.save()

        response = self.client.get(self.api_link)
        self.assertEqual(response.status_code, 403)
        self.assertEqual(
            response.json(),
            {
                "detail": (
                    "No categories that allow new threads are available "
                    "to you at the moment."
                )
            },
        )

    @patch_category_acl({"can_close_threads": True, "can_start_threads": True})
    def test_category_closed_allowing_new_threads(self):
        """endpoint adds closed category that allows new threads"""
        self.category.is_closed = True
        self.category.save()

        response = self.client.get(self.api_link)
        self.assertEqual(response.status_code, 200)

        response_json = response.json()
        self.assertEqual(
            response_json[0],
            {
                "id": self.category.pk,
                "name": self.category.name,
                "level": 0,
                "post": {"close": True, "hide": False, "pin": 0},
            },
        )

    @patch_category_acl({"can_start_threads": True})
    def test_category_allowing_new_threads(self):
        """endpoint adds category that allows new threads"""
        response = self.client.get(self.api_link)
        self.assertEqual(response.status_code, 200)

        response_json = response.json()
        self.assertEqual(
            response_json[0],
            {
                "id": self.category.pk,
                "name": self.category.name,
                "level": 0,
                "post": {"close": False, "hide": False, "pin": 0},
            },
        )

    @patch_category_acl({"can_close_threads": True, "can_start_threads": True})
    def test_category_allowing_closing_threads(self):
        """endpoint adds category that allows new closed threads"""
        response = self.client.get(self.api_link)
        self.assertEqual(response.status_code, 200)

        response_json = response.json()
        self.assertEqual(
            response_json[0],
            {
                "id": self.category.pk,
                "name": self.category.name,
                "level": 0,
                "post": {"close": True, "hide": False, "pin": 0},
            },
        )

    @patch_category_acl({"can_start_threads": True, "can_pin_threads": 1})
    def test_category_allowing_locally_pinned_threads(self):
        """endpoint adds category that allows locally pinned threads"""
        response = self.client.get(self.api_link)
        self.assertEqual(response.status_code, 200)

        response_json = response.json()
        self.assertEqual(
            response_json[0],
            {
                "id": self.category.pk,
                "name": self.category.name,
                "level": 0,
                "post": {"close": False, "hide": False, "pin": 1},
            },
        )

    @patch_category_acl({"can_start_threads": True, "can_pin_threads": 2})
    def test_category_allowing_globally_pinned_threads(self):
        """endpoint adds category that allows globally pinned threads"""
        response = self.client.get(self.api_link)
        self.assertEqual(response.status_code, 200)

        response_json = response.json()
        self.assertEqual(
            response_json[0],
            {
                "id": self.category.pk,
                "name": self.category.name,
                "level": 0,
                "post": {"close": False, "hide": False, "pin": 2},
            },
        )

    @patch_category_acl({"can_start_threads": True, "can_hide_threads": 1})
    def test_category_allowing_hidding_threads(self):
        """endpoint adds category that allows hiding threads"""
        response = self.client.get(self.api_link)
        self.assertEqual(response.status_code, 200)

        response_json = response.json()
        self.assertEqual(
            response_json[0],
            {
                "id": self.category.pk,
                "name": self.category.name,
                "level": 0,
                "post": {"close": 0, "hide": 1, "pin": 0},
            },
        )

    @patch_category_acl({"can_start_threads": True, "can_hide_threads": 2})
    def test_category_allowing_hidding_and_deleting_threads(self):
        """endpoint adds category that allows hiding and deleting threads"""
        response = self.client.get(self.api_link)
        self.assertEqual(response.status_code, 200)

        response_json = response.json()
        self.assertEqual(
            response_json[0],
            {
                "id": self.category.pk,
                "name": self.category.name,
                "level": 0,
                "post": {"close": False, "hide": 1, "pin": 0},
            },
        )


class ThreadReplyEditorApiTests(EditorApiTestCase):
    def setUp(self):
        super().setUp()

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

    def test_anonymous_user_request(self):
        """endpoint validates if user is authenticated"""
        self.logout_user()

        response = self.client.get(self.api_link)
        self.assertEqual(response.status_code, 403)
        self.assertEqual(
            response.json(), {"detail": "You have to sign in to reply threads."}
        )

    def test_thread_visibility(self):
        """thread's visibility is validated"""
        with patch_category_acl({"can_see": False}):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 404)

        with patch_category_acl({"can_browse": False}):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 404)

        with patch_category_acl({"can_see_all_threads": False}):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 404)

    @patch_category_acl({"can_reply_threads": False})
    def test_no_reply_permission(self):
        """permssion to reply is validated"""
        response = self.client.get(self.api_link)
        self.assertEqual(response.status_code, 403)
        self.assertEqual(
            response.json(), {"detail": "You can't reply to threads in this category."}
        )

    def test_closed_category(self):
        """permssion to reply in closed category is validated"""
        self.category.is_closed = True
        self.category.save()

        with patch_category_acl(
            {"can_reply_threads": True, "can_close_threads": False}
        ):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 403)
            self.assertEqual(
                response.json(),
                {
                    "detail": (
                        "This category is closed. You can't reply to threads in it."
                    )
                },
            )

        # allow to post in closed category
        with patch_category_acl({"can_reply_threads": True, "can_close_threads": True}):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 200)

    def test_closed_thread(self):
        """permssion to reply in closed thread is validated"""
        self.thread.is_closed = True
        self.thread.save()

        with patch_category_acl(
            {"can_reply_threads": True, "can_close_threads": False}
        ):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 403)
            self.assertEqual(
                response.json(),
                {"detail": "You can't reply to closed threads in this category."},
            )

        # allow to post in closed thread
        with patch_category_acl({"can_reply_threads": True, "can_close_threads": True}):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 200)

    @patch_category_acl({"can_reply_threads": True})
    def test_allow_reply_thread(self):
        """api returns 200 code if thread reply is allowed"""
        response = self.client.get(self.api_link)
        self.assertEqual(response.status_code, 200)

    def test_reply_to_visibility(self):
        """api validates replied post visibility"""

        # unapproved reply can't be replied to
        unapproved_reply = test.reply_thread(self.thread, is_unapproved=True)

        with patch_category_acl({"can_reply_threads": True}):
            response = self.client.get(
                "%s?reply=%s" % (self.api_link, unapproved_reply.pk)
            )
            self.assertEqual(response.status_code, 404)

        # hidden reply can't be replied to
        hidden_reply = test.reply_thread(self.thread, is_hidden=True)

        with patch_category_acl({"can_reply_threads": True}):
            response = self.client.get("%s?reply=%s" % (self.api_link, hidden_reply.pk))
            self.assertEqual(response.status_code, 403)
            self.assertEqual(
                response.json(), {"detail": "You can't reply to hidden posts."}
            )

    def test_reply_to_other_thread_post(self):
        """api validates is replied post belongs to same thread"""
        other_thread = test.post_thread(category=self.category)
        reply_to = test.reply_thread(other_thread)

        response = self.client.get("%s?reply=%s" % (self.api_link, reply_to.pk))
        self.assertEqual(response.status_code, 404)

    @patch_category_acl({"can_reply_threads": True})
    def test_reply_to_event(self):
        """events can't be replied to"""
        reply_to = test.reply_thread(self.thread, is_event=True)

        response = self.client.get("%s?reply=%s" % (self.api_link, reply_to.pk))
        self.assertEqual(response.status_code, 403)
        self.assertEqual(response.json(), {"detail": "You can't reply to events."})

    @patch_category_acl({"can_reply_threads": True})
    def test_reply_to(self):
        """api includes replied to post details in response"""
        reply_to = test.reply_thread(self.thread)

        response = self.client.get("%s?reply=%s" % (self.api_link, reply_to.pk))

        self.assertEqual(response.status_code, 200)
        self.assertEqual(
            response.json(),
            {
                "id": reply_to.pk,
                "post": reply_to.original,
                "poster": reply_to.poster_name,
            },
        )


class EditReplyEditorApiTests(EditorApiTestCase):
    def setUp(self):
        super().setUp()

        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-editor",
            kwargs={"thread_pk": self.thread.pk, "pk": self.post.pk},
        )

    def test_anonymous_user_request(self):
        """endpoint validates if user is authenticated"""
        self.logout_user()

        response = self.client.get(self.api_link)
        self.assertEqual(response.status_code, 403)
        self.assertEqual(
            response.json(), {"detail": "You have to sign in to edit posts."}
        )

    def test_thread_visibility(self):
        """thread's visibility is validated"""
        with patch_category_acl({"can_see": False}):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 404)

        with patch_category_acl({"can_browse": False}):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 404)

        with patch_category_acl({"can_see_all_threads": False}):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 404)

    @patch_category_acl({"can_edit_posts": 0})
    def test_no_edit_permission(self):
        """permssion to edit is validated"""
        response = self.client.get(self.api_link)
        self.assertEqual(response.status_code, 403)
        self.assertEqual(
            response.json(), {"detail": "You can't edit posts in this category."}
        )

    def test_closed_category(self):
        """permssion to edit in closed category is validated"""
        self.category.is_closed = True
        self.category.save()

        with patch_category_acl({"can_edit_posts": 1, "can_close_threads": False}):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 403)
            self.assertEqual(
                response.json(),
                {"detail": "This category is closed. You can't edit posts in it."},
            )

        # allow to edit in closed category
        with patch_category_acl({"can_edit_posts": 1, "can_close_threads": True}):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 200)

    def test_closed_thread(self):
        """permssion to edit in closed thread is validated"""
        self.thread.is_closed = True
        self.thread.save()

        with patch_category_acl({"can_edit_posts": 1, "can_close_threads": False}):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 403)
            self.assertEqual(
                response.json(),
                {"detail": "This thread is closed. You can't edit posts in it."},
            )

        # allow to edit in closed thread
        with patch_category_acl({"can_edit_posts": 1, "can_close_threads": True}):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 200)

    def test_protected_post(self):
        """permssion to edit protected post is validated"""
        self.post.is_protected = True
        self.post.save()

        with patch_category_acl({"can_edit_posts": 1, "can_protect_posts": False}):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 403)
            self.assertEqual(
                response.json(),
                {"detail": "This post is protected. You can't edit it."},
            )

        # allow to post in closed thread
        with patch_category_acl({"can_edit_posts": 1, "can_protect_posts": True}):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 200)

    def test_post_visibility(self):
        """edited posts visibility is validated"""
        self.post.is_hidden = True
        self.post.save()

        with patch_category_acl({"can_edit_posts": 1}):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 403)
            self.assertEqual(
                response.json(), {"detail": "This post is hidden, you can't edit it."}
            )

        # allow hidden edition
        with patch_category_acl({"can_edit_posts": 1, "can_hide_posts": 1}):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 200)

        # test unapproved post
        self.post.is_unapproved = True
        self.post.is_hidden = False
        self.post.poster = None
        self.post.save()

        with patch_category_acl({"can_edit_posts": 2, "can_approve_content": 0}):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 404)

        # allow unapproved edition
        with patch_category_acl({"can_edit_posts": 2, "can_approve_content": 1}):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 200)

    @patch_category_acl({"can_edit_posts": 2})
    def test_post_is_event(self):
        """events can't be edited"""
        self.post.is_event = True
        self.post.save()

        response = self.client.get(self.api_link)
        self.assertEqual(response.status_code, 403)
        self.assertEqual(response.json(), {"detail": "Events can't be edited."})

    def test_other_user_post(self):
        """api validates if other user's post can be edited"""
        self.post.poster = None
        self.post.save()

        with patch_category_acl({"can_edit_posts": 1}):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 403)
            self.assertEqual(
                response.json(),
                {"detail": "You can't edit other users posts in this category."},
            )

        # allow other users post edition
        with patch_category_acl({"can_edit_posts": 2}):
            response = self.client.get(self.api_link)
            self.assertEqual(response.status_code, 200)

    @patch_category_acl({"can_hide_threads": 1, "can_edit_posts": 2})
    def test_edit_first_post_hidden(self):
        """
        endpoint returns valid configuration for editor of hidden thread's first post
        """
        self.thread.is_hidden = True
        self.thread.save()
        self.thread.first_post.is_hidden = True
        self.thread.first_post.save()

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

        response = self.client.get(api_link)
        self.assertEqual(response.status_code, 200)

    @patch_category_acl({"can_edit_posts": 1})
    def test_edit(self):
        """endpoint returns valid configuration for editor"""
        with patch_category_acl({"max_attachment_size": 1000}):
            for _ in range(3):
                with open(TEST_DOCUMENT_PATH, "rb") as upload:
                    response = self.client.post(
                        reverse("misago:api:attachment-list"), data={"upload": upload}
                    )
                self.assertEqual(response.status_code, 200)

        attachments = list(Attachment.objects.order_by("id"))

        attachments[0].uploader = None
        attachments[0].save()

        for attachment in attachments[:2]:
            attachment.post = self.post
            attachment.save()

        response = self.client.get(self.api_link)
        user_acl = useracl.get_user_acl(self.user, cache_versions)
        for attachment in attachments:
            add_acl_to_obj(user_acl, attachment)

        self.assertEqual(response.status_code, 200)
        self.assertEqual(
            response.json(),
            {
                "id": self.post.pk,
                "api": self.post.get_api_url(),
                "post": self.post.original,
                "can_protect": False,
                "is_protected": self.post.is_protected,
                "poster": self.post.poster_name,
                "attachments": [
                    AttachmentSerializer(
                        attachments[1], context={"user": self.user}
                    ).data,
                    AttachmentSerializer(
                        attachments[0], context={"user": self.user}
                    ).data,
                ],
            },
        )