Misago/misago/threads/tests/test_thread_patch_api.py

import json

from django.utils import six
from django.utils.encoding import smart_str

from misago.acl.testutils import override_acl
from misago.categories.models import Category

from .test_threads_api import ThreadsApiTestCase


class ThreadPinGloballyApiTests(ThreadsApiTestCase):
    def test_pin_thread(self):
        """api makes it possible to pin globally thread"""
        self.override_acl({
            'can_pin_threads': 2
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'weight', 'value': 2}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 200)

        thread_json = self.get_thread_json()
        self.assertEqual(thread_json['weight'], 2)

    def test_unpin_thread(self):
        """api makes it possible to unpin thread"""
        self.thread.weight = 2
        self.thread.save()

        thread_json = self.get_thread_json()
        self.assertEqual(thread_json['weight'], 2)

        self.override_acl({
            'can_pin_threads': 2
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'weight', 'value': 0}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 200)

        thread_json = self.get_thread_json()
        self.assertEqual(thread_json['weight'], 0)

    def test_pin_thread_no_permission(self):
        """api pin thread globally with no permission fails"""
        self.override_acl({
            'can_pin_threads': 1
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'weight', 'value': 2}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 400)

        response_json = json.loads(smart_str(response.content))
        self.assertEqual(response_json['detail'][0],
            "You don't have permission to pin this thread globally.")

        thread_json = self.get_thread_json()
        self.assertEqual(thread_json['weight'], 0)

    def test_unpin_thread_no_permission(self):
        """api unpin thread with no permission fails"""
        self.thread.weight = 2
        self.thread.save()

        thread_json = self.get_thread_json()
        self.assertEqual(thread_json['weight'], 2)

        self.override_acl({
            'can_pin_threads': 1
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'weight', 'value': 1}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 400)

        response_json = json.loads(smart_str(response.content))
        self.assertEqual(response_json['detail'][0],
            "You don't have permission to change this thread's weight.")

        thread_json = self.get_thread_json()
        self.assertEqual(thread_json['weight'], 2)


class ThreadPinLocallyApiTests(ThreadsApiTestCase):
    def test_pin_thread(self):
        """api makes it possible to pin locally thread"""
        self.override_acl({
            'can_pin_threads': 1
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'weight', 'value': 1}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 200)

        thread_json = self.get_thread_json()
        self.assertEqual(thread_json['weight'], 1)

    def test_unpin_thread(self):
        """api makes it possible to unpin thread"""
        self.thread.weight = 1
        self.thread.save()

        thread_json = self.get_thread_json()
        self.assertEqual(thread_json['weight'], 1)

        self.override_acl({
            'can_pin_threads': 1
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'weight', 'value': 0}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 200)

        thread_json = self.get_thread_json()
        self.assertEqual(thread_json['weight'], 0)

    def test_pin_thread_no_permission(self):
        """api pin thread locally with no permission fails"""
        self.override_acl({
            'can_pin_threads': 0
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'weight', 'value': 1}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 400)

        response_json = json.loads(smart_str(response.content))
        self.assertEqual(response_json['detail'][0],
            "You don't have permission to change this thread's weight.")

        thread_json = self.get_thread_json()
        self.assertEqual(thread_json['weight'], 0)

    def test_unpin_thread_no_permission(self):
        """api unpin thread with no permission fails"""
        self.thread.weight = 1
        self.thread.save()

        thread_json = self.get_thread_json()
        self.assertEqual(thread_json['weight'], 1)

        self.override_acl({
            'can_pin_threads': 0
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'weight', 'value': 0}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 400)

        response_json = json.loads(smart_str(response.content))
        self.assertEqual(response_json['detail'][0],
            "You don't have permission to change this thread's weight.")

        thread_json = self.get_thread_json()
        self.assertEqual(thread_json['weight'], 1)


class ThreadMoveApiTests(ThreadsApiTestCase):
    def setUp(self):
        super(ThreadMoveApiTests, self).setUp()

        Category(
            name='Category B',
            slug='category-b',
        ).insert_at(self.category, position='last-child', save=True)
        self.category_b = Category.objects.get(slug='category-b')

    def override_other_acl(self, acl):
        final_acl = {
            'can_see': 1,
            'can_browse': 1,
            'can_see_all_threads': 1,
            'can_see_own_threads': 0,
            'can_hide_threads': 0,
            'can_approve_content': 0,
        }
        final_acl.update(acl)

        categories_acl = self.user.acl['categories']
        categories_acl[self.category_b.pk] = final_acl

        visible_categories = [self.category.pk]
        if final_acl['can_see']:
            visible_categories.append(self.category_b.pk)

        override_acl(self.user, {
            'visible_categories': visible_categories,
            'categories': categories_acl,
        })

    def test_move_thread_no_top(self):
        """api moves thread to other category, sets no top category"""
        self.override_acl({
            'can_move_threads': True
        })
        self.override_other_acl({
            'can_start_threads': 2
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'category', 'value': self.category_b.pk},
            {'op': 'add', 'path': 'top-category', 'value': self.category_b.pk},
            {'op': 'replace', 'path': 'flatten-categories', 'value': None},
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 200)

        self.override_other_acl({})

        thread_json = self.get_thread_json()
        self.assertEqual(thread_json['category']['id'], self.category_b.pk)

        reponse_json = json.loads(smart_str(response.content))
        self.assertEqual(reponse_json['category'], self.category_b.pk)
        self.assertEqual(reponse_json['top_category'], None)

    def test_move_thread_with_top(self):
        """api moves thread to other category, sets top"""
        self.override_acl({
            'can_move_threads': True
        })
        self.override_other_acl({
            'can_start_threads': 2
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'category', 'value': self.category_b.pk},
            {
                'op': 'add',
                'path': 'top-category',
                'value': Category.objects.root_category().pk,
            },
            {'op': 'replace', 'path': 'flatten-categories', 'value': None},
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 200)

        self.override_other_acl({})

        thread_json = self.get_thread_json()
        self.assertEqual(thread_json['category']['id'], self.category_b.pk)

        reponse_json = json.loads(smart_str(response.content))
        self.assertEqual(reponse_json['category'], self.category_b.pk)
        self.assertEqual(reponse_json['top_category'], self.category.pk)

    def test_move_thread_no_permission(self):
        """api move thread to other category with no permission fails"""
        self.override_acl({
            'can_move_threads': False
        })
        self.override_other_acl({})

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'category', 'value': self.category_b.pk}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 400)

        response_json = json.loads(smart_str(response.content))
        self.assertEqual(response_json['detail'][0],
            "You don't have permission to move this thread.")

        self.override_other_acl({})

        thread_json = self.get_thread_json()
        self.assertEqual(thread_json['category']['id'], self.category.pk)

    def test_move_thread_no_category_access(self):
        """api move thread to category with no access fails"""
        self.override_acl({
            'can_move_threads': True
        })
        self.override_other_acl({
            'can_see': False
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'category', 'value': self.category_b.pk}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 400)

        response_json = json.loads(smart_str(response.content))
        self.assertEqual(response_json['detail'][0], 'NOT FOUND')

        self.override_other_acl({})

        thread_json = self.get_thread_json()
        self.assertEqual(thread_json['category']['id'], self.category.pk)

    def test_move_thread_no_category_browse(self):
        """api move thread to category with no browsing access fails"""
        self.override_acl({
            'can_move_threads': True
        })
        self.override_other_acl({
            'can_browse': False
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'category', 'value': self.category_b.pk}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 400)

        response_json = json.loads(smart_str(response.content))
        self.assertEqual(response_json['detail'][0],
            'You don\'t have permission to browse "Category B" contents.')

        self.override_other_acl({})

        thread_json = self.get_thread_json()
        self.assertEqual(thread_json['category']['id'], self.category.pk)

    def test_thread_flatten_categories(self):
        """api flatten thread categories"""
        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'flatten-categories', 'value': None}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 200)

        response_json = json.loads(smart_str(response.content))
        self.assertEqual(response_json['category'], self.category.pk)

    def test_thread_top_flatten_categories(self):
        """api flatten thread with top category"""
        self.thread.category = self.category_b
        self.thread.save()

        self.override_other_acl({})

        response = self.client.patch(self.api_link, json.dumps([
            {
                'op': 'add',
                'path': 'top-category',
                'value': Category.objects.root_category().pk,
            },
            {'op': 'replace', 'path': 'flatten-categories', 'value': None},
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 200)

        response_json = json.loads(smart_str(response.content))
        self.assertEqual(response_json['top_category'], self.category.pk)
        self.assertEqual(response_json['category'], self.category_b.pk)


class ThreadCloseApiTests(ThreadsApiTestCase):
    def test_close_thread(self):
        """api makes it possible to close thread"""
        self.override_acl({
            'can_close_threads': True
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'is-closed', 'value': True}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 200)

        thread_json = self.get_thread_json()
        self.assertTrue(thread_json['is_closed'])

    def test_open_thread(self):
        """api makes it possible to open thread"""
        self.thread.is_closed = True
        self.thread.save()

        thread_json = self.get_thread_json()
        self.assertTrue(thread_json['is_closed'])

        self.override_acl({
            'can_close_threads': True
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'is-closed', 'value': False}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 200)

        thread_json = self.get_thread_json()
        self.assertFalse(thread_json['is_closed'])

    def test_close_thread_no_permission(self):
        """api close thread with no permission fails"""
        self.override_acl({
            'can_close_threads': False
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'is-closed', 'value': True}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 400)

        response_json = json.loads(smart_str(response.content))
        self.assertEqual(response_json['detail'][0],
            "You don't have permission to close this thread.")

        thread_json = self.get_thread_json()
        self.assertFalse(thread_json['is_closed'])

    def test_open_thread_no_permission(self):
        """api open thread with no permission fails"""
        self.thread.is_closed = True
        self.thread.save()

        thread_json = self.get_thread_json()
        self.assertTrue(thread_json['is_closed'])

        self.override_acl({
            'can_close_threads': False
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'is-closed', 'value': False}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 400)

        response_json = json.loads(smart_str(response.content))
        self.assertEqual(response_json['detail'][0],
            "You don't have permission to open this thread.")

        thread_json = self.get_thread_json()
        self.assertTrue(thread_json['is_closed'])


class ThreadApproveApiTests(ThreadsApiTestCase):
    def test_approve_thread(self):
        """api makes it possible to approve thread"""
        self.thread.is_unapproved = True
        self.thread.save()

        self.override_acl({
            'can_approve_content': 1
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'is-unapproved', 'value': False}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 200)

        thread_json = self.get_thread_json()
        self.assertFalse(thread_json['is_unapproved'])

    def test_unapprove_thread(self):
        """api returns permission error on approval removal"""
        self.override_acl({
            'can_approve_content': 1
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'is-unapproved', 'value': True}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 400)

        response_json = json.loads(smart_str(response.content))
        self.assertEqual(response_json['detail'][0],
            "Content approval can't be reversed.")


class ThreadHideApiTests(ThreadsApiTestCase):
    def test_hide_thread(self):
        """api makes it possible to hide thread"""
        self.override_acl({
            'can_hide_threads': 1
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'is-hidden', 'value': True}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 200)

        self.override_acl({
            'can_hide_threads': 1
        })

        thread_json = self.get_thread_json()
        self.assertTrue(thread_json['is_hidden'])

    def test_show_thread(self):
        """api makes it possible to unhide thread"""
        self.thread.is_hidden = True
        self.thread.save()

        self.override_acl({
            'can_hide_threads': 1
        })

        thread_json = self.get_thread_json()
        self.assertTrue(thread_json['is_hidden'])

        self.override_acl({
            'can_hide_threads': 1
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'is-hidden', 'value': False}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 200)

        self.override_acl({
            'can_hide_threads': 1
        })

        thread_json = self.get_thread_json()
        self.assertFalse(thread_json['is_hidden'])

    def test_hide_thread_no_permission(self):
        """api hide thread with no permission fails"""
        self.override_acl({
            'can_hide_threads': 0
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'is-hidden', 'value': True}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 400)

        response_json = json.loads(smart_str(response.content))
        self.assertEqual(response_json['detail'][0],
            "You don't have permission to hide this thread.")

        thread_json = self.get_thread_json()
        self.assertFalse(thread_json['is_hidden'])

    def test_show_thread_no_permission(self):
        """api unhide thread with no permission fails"""
        self.thread.is_hidden = True
        self.thread.save()

        self.override_acl({
            'can_hide_threads': 1
        })

        thread_json = self.get_thread_json()
        self.assertTrue(thread_json['is_hidden'])

        self.override_acl({
            'can_hide_threads': 0
        })

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'is-hidden', 'value': False}
        ]),
        content_type="application/json")
        self.assertEqual(response.status_code, 404)


class ThreadSubscribeApiTests(ThreadsApiTestCase):
    def test_subscribe_thread(self):
        """api makes it possible to subscribe thread"""
        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'subscription', 'value': 'notify'}
        ]),
        content_type="application/json")

        self.assertEqual(response.status_code, 200)

        thread_json = self.get_thread_json()
        self.assertFalse(thread_json['subscription'])

        subscription = self.user.subscription_set.get(thread=self.thread)
        self.assertFalse(subscription.send_email)

    def test_subscribe_thread_with_email(self):
        """api makes it possible to subscribe thread with emails"""
        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'subscription', 'value': 'email'}
        ]),
        content_type="application/json")

        self.assertEqual(response.status_code, 200)

        thread_json = self.get_thread_json()
        self.assertTrue(thread_json['subscription'])

        subscription = self.user.subscription_set.get(thread=self.thread)
        self.assertTrue(subscription.send_email)

    def test_unsubscribe_thread(self):
        """api makes it possible to unsubscribe thread"""
        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'subscription', 'value': 'remove'}
        ]),
        content_type="application/json")

        self.assertEqual(response.status_code, 200)

        thread_json = self.get_thread_json()
        self.assertIsNone(thread_json['subscription'])

        self.assertEqual(self.user.subscription_set.count(), 0)

    def test_subscribe_as_guest(self):
        """api makes it impossible to subscribe thread"""
        self.logout_user()

        response = self.client.patch(self.api_link, json.dumps([
            {'op': 'replace', 'path': 'subscription', 'value': 'email'}
        ]),
        content_type="application/json")

        self.assertEqual(response.status_code, 403)

    def test_subscribe_nonexistant_thread(self):
        """api makes it impossible to subscribe nonexistant thread"""
        bad_api_link = self.api_link.replace(
            six.text_type(self.thread.pk), six.text_type(self.thread.pk + 9))

        response = self.client.patch(bad_api_link, json.dumps([
            {'op': 'replace', 'path': 'subscription', 'value': 'email'}
        ]),
        content_type="application/json")

        self.assertEqual(response.status_code, 404)