Misago/misago/threads/tests/test_threads_api.py

from datetime import timedelta

from django.utils import timezone
from django.urls import reverse

from misago.categories import THREADS_ROOT_NAME
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
from misago.threads.threadtypes import trees_map
from misago.users.testutils import AuthenticatedUserTestCase


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

        threads_tree_id = trees_map.get_tree_id_for_root(THREADS_ROOT_NAME)

        self.root = Category.objects.get(tree_id=threads_tree_id, level=0)
        self.category = Category.objects.get(slug='first-category')

        self.thread = testutils.post_thread(category=self.category)
        self.api_link = self.thread.get_api_url()

    def get_thread_json(self):
        response = self.client.get(self.thread.get_api_url())
        self.assertEqual(response.status_code, 200)

        return response.json()


class ThreadRetrieveApiTests(ThreadsApiTestCase):
    def setUp(self):
        super().setUp()

        self.tested_links = [
            self.api_link,
            '%sposts/' % self.api_link,
            '%sposts/?page=1' % self.api_link,
        ]

    @patch_category_acl()
    def test_api_returns_thread(self):
        """api has no showstoppers"""
        for link in self.tested_links:
            response = self.client.get(link)
            self.assertEqual(response.status_code, 200)

            response_json = response.json()
            self.assertEqual(response_json['id'], self.thread.pk)
            self.assertEqual(response_json['title'], self.thread.title)

            if 'posts' in link:
                self.assertIn('post_set', response_json)

    @patch_category_acl({"can_see_all_threads": False})
    def test_api_shows_owned_thread(self):
        """api handles "owned threads only"""
        for link in self.tested_links:
            response = self.client.get(link)
            self.assertEqual(response.status_code, 404)

        self.thread.starter = self.user
        self.thread.save()

        for link in self.tested_links:
            response = self.client.get(link)
            self.assertEqual(response.status_code, 200)

    @patch_category_acl({"can_see": False})
    def test_api_validates_category_see_permission(self):
        """api validates category visiblity"""
        for link in self.tested_links:
            response = self.client.get(link)
            self.assertEqual(response.status_code, 404)

    @patch_category_acl({"can_browse": False})
    def test_api_validates_category_browse_permission(self):
        """api validates category browsability"""
        for link in self.tested_links:
            response = self.client.get(link)
            self.assertEqual(response.status_code, 404)

    def test_api_validates_posts_visibility(self):
        """api validates posts visiblity"""
        hidden_post = testutils.reply_thread(
            self.thread,
            is_hidden=True,
            message="I'am hidden test message!",
        )

        with patch_category_acl({"can_hide_posts": 0}):
            response = self.client.get(self.tested_links[1])
            self.assertNotContains(response, hidden_post.parsed)  # post's body is hidden

        # add permission to see hidden posts
        with patch_category_acl({"can_hide_posts": 1}):
            response = self.client.get(self.tested_links[1])
            self.assertContains(
                response, hidden_post.parsed
            )  # hidden post's body is visible with permission

        # unapproved posts shouldn't show at all
        unapproved_post = testutils.reply_thread(
            self.thread,
            is_unapproved=True,
        )

        with patch_category_acl({"can_approve_content": False}):
            response = self.client.get(self.tested_links[1])
            self.assertNotContains(response, unapproved_post.get_absolute_url())

        # add permission to see unapproved posts
        with patch_category_acl({"can_approve_content": True}):
            response = self.client.get(self.tested_links[1])
            self.assertContains(response, unapproved_post.get_absolute_url())

    def test_api_validates_has_unapproved_posts_visibility(self):
        """api checks acl before exposing unapproved flag"""
        self.thread.has_unapproved_posts = True
        self.thread.save()

        with patch_category_acl({"can_approve_content": False}):
            for link in self.tested_links:
                response = self.client.get(link)
                self.assertEqual(response.status_code, 200)

                response_json = response.json()
                self.assertEqual(response_json['id'], self.thread.pk)
                self.assertEqual(response_json['title'], self.thread.title)
                self.assertFalse(response_json['has_unapproved_posts'])

        with patch_category_acl({"can_approve_content": True}):
            for link in self.tested_links:
                response = self.client.get(link)
                self.assertEqual(response.status_code, 200)

                response_json = response.json()
                self.assertEqual(response_json['id'], self.thread.pk)
                self.assertEqual(response_json['title'], self.thread.title)
                self.assertTrue(response_json['has_unapproved_posts'])


class ThreadDeleteApiTests(ThreadsApiTestCase):
    def setUp(self):
        super().setUp()

        self.last_thread = testutils.post_thread(category=self.category)
        self.api_link = self.last_thread.get_api_url()

    def test_delete_thread_no_permission(self):
        """api tests permission to delete threads"""
        with patch_category_acl({"can_hide_threads": 0}):
            response = self.client.delete(self.api_link)
            self.assertEqual(response.status_code, 403)
            self.assertEqual(
                response.json()['detail'], "You can't delete threads in this category."
            )

        with patch_category_acl({"can_hide_threads": 1}):
            response = self.client.delete(self.api_link)
            self.assertEqual(response.status_code, 403)
            self.assertEqual(
                response.json()['detail'], "You can't delete threads in this category."
            )

    @patch_category_acl({'can_hide_threads': 1, 'can_hide_own_threads': 2})
    def test_delete_other_user_thread_no_permission(self):
        """api tests thread owner when deleting own thread"""
        response = self.client.delete(self.api_link)
        self.assertEqual(response.status_code, 403)
        self.assertEqual(
            response.json()['detail'], "You can't delete other users theads in this category."
        )

    @patch_category_acl({
        'can_hide_threads': 2,
        'can_hide_own_threads': 2,
        'can_close_threads': False,
    })
    def test_delete_thread_closed_category_no_permission(self):
        """api tests category's closed state"""
        self.category.is_closed = True
        self.category.save()

        response = self.client.delete(self.api_link)
        self.assertEqual(response.status_code, 403)
        self.assertEqual(
            response.json()['detail'], "This category is closed. You can't delete threads in it."
        )

    @patch_category_acl({
        'can_hide_threads': 2,
        'can_hide_own_threads': 2,
        'can_close_threads': False,
    })
    def test_delete_thread_closed_no_permission(self):
        """api tests thread's closed state"""
        self.last_thread.is_closed = True
        self.last_thread.save()

        response = self.client.delete(self.api_link)
        self.assertEqual(response.status_code, 403)
        self.assertEqual(
            response.json()['detail'], "This thread is closed. You can't delete it."
        )

    @patch_category_acl({
        'can_hide_threads': 1,
        'can_hide_own_threads': 2,
        'thread_edit_time': 1
    })
    def test_delete_owned_thread_no_time(self):
        """api tests permission to delete owned thread within time limit"""
        self.last_thread.starter = self.user
        self.last_thread.started_on = timezone.now() - timedelta(minutes=10)
        self.last_thread.save()

        response = self.client.delete(self.api_link)
        self.assertEqual(response.status_code, 403)
        self.assertEqual(
            response.json()['detail'], "You can't delete threads that are older than 1 minute."
        )

    @patch_category_acl({'can_hide_threads': 2})
    def test_delete_thread(self):
        """DELETE to API link with permission deletes thread"""
        category = Category.objects.get(slug='first-category')
        self.assertEqual(category.last_thread_id, self.last_thread.pk)

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

        with self.assertRaises(Thread.DoesNotExist):
            Thread.objects.get(pk=self.last_thread.pk)

        # category was synchronised after deletion
        category = Category.objects.get(slug='first-category')
        self.assertEqual(category.last_thread_id, self.thread.pk)

        # test that last thread's deletion triggers category sync
        response = self.client.delete(self.thread.get_api_url())
        self.assertEqual(response.status_code, 200)

        with self.assertRaises(Thread.DoesNotExist):
            Thread.objects.get(pk=self.thread.pk)

        category = Category.objects.get(slug='first-category')
        self.assertIsNone(category.last_thread_id)

    @patch_category_acl({
        'can_hide_threads': 1,
        'can_hide_own_threads': 2,
        'thread_edit_time': 30
    })
    def test_delete_owned_thread(self):
        """api lets owner to delete owned thread within time limit"""
        self.last_thread.starter = self.user
        self.last_thread.started_on = timezone.now() - timedelta(minutes=10)
        self.last_thread.save()

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