Misago/misago/threads/tests/test_threadslists.py

from datetime import timedelta

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

from misago.acl.testutils import override_acl
from misago.api.testutils import ApiTestsMixin
from misago.categories.models import Category
from misago.conf import settings
from misago.readtracker import poststracker
from misago.threads import testutils
from misago.users.models import AnonymousUser
from misago.users.testutils import AuthenticatedUserTestCase


LISTS_URLS = ('', 'my/', 'new/', 'unread/', 'subscribed/', )


class ThreadsListTestCase(AuthenticatedUserTestCase, ApiTestsMixin):
    def setUp(self):
        """
        Create categories tree for test cases:

        First category (created by migration)

        Category A
          + Category B
            + Subcategory C
            + Subcategory D

        Category E
          + Subcategory F
        """
        super(ThreadsListTestCase, self).setUp()

        self.api_link = reverse('misago:api:thread-list')

        self.root = Category.objects.root_category()
        self.first_category = Category.objects.get(slug='first-category')

        Category(
            name='Category A',
            slug='category-a',
            css_class='showing-category-a',
        ).insert_at(
            self.root,
            position='last-child',
            save=True,
        )
        Category(
            name='Category E',
            slug='category-e',
            css_class='showing-category-e',
        ).insert_at(
            self.root,
            position='last-child',
            save=True,
        )

        self.root = Category.objects.root_category()

        self.category_a = Category.objects.get(slug='category-a')

        Category(
            name='Category B',
            slug='category-b',
            css_class='showing-category-b',
        ).insert_at(
            self.category_a,
            position='last-child',
            save=True,
        )

        self.category_b = Category.objects.get(slug='category-b')

        Category(
            name='Category C',
            slug='category-c',
            css_class='showing-category-c',
        ).insert_at(
            self.category_b,
            position='last-child',
            save=True,
        )
        Category(
            name='Category D',
            slug='category-d',
            css_class='showing-category-d',
        ).insert_at(
            self.category_b,
            position='last-child',
            save=True,
        )

        self.category_c = Category.objects.get(slug='category-c')
        self.category_d = Category.objects.get(slug='category-d')

        self.category_e = Category.objects.get(slug='category-e')
        Category(
            name='Category F',
            slug='category-f',
            css_class='showing-category-f',
        ).insert_at(
            self.category_e,
            position='last-child',
            save=True,
        )

        self.category_f = Category.objects.get(slug='category-f')

        self.clear_state()

        Category.objects.partial_rebuild(self.root.tree_id)

        self.root = Category.objects.root_category()
        self.category_a = Category.objects.get(slug='category-a')
        self.category_b = Category.objects.get(slug='category-b')
        self.category_c = Category.objects.get(slug='category-c')
        self.category_d = Category.objects.get(slug='category-d')
        self.category_e = Category.objects.get(slug='category-e')
        self.category_f = Category.objects.get(slug='category-f')

        self.access_all_categories()

    def access_all_categories(self, category_acl=None, base_acl=None):
        self.clear_state()

        categories_acl = {
            'categories': {},
            'visible_categories': [],
            'browseable_categories': [],
            'can_approve_content': [],
        }

        # copy first category's acl to other categories to make base for overrides
        first_category_acl = self.user.acl_cache['categories'][self.first_category.pk].copy()
        for category in Category.objects.all_categories():
            categories_acl['categories'][category.pk] = first_category_acl

        if base_acl:
            categories_acl.update(base_acl)

        for category in Category.objects.all_categories():
            categories_acl['visible_categories'].append(category.pk)
            categories_acl['browseable_categories'].append(category.pk)
            categories_acl['categories'][category.pk].update({
                'can_see': 1,
                'can_browse': 1,
                'can_see_all_threads': 1,
                'can_see_own_threads': 0,
                'can_hide_threads': 0,
                'can_approve_content': 0,
            })

            if category_acl:
                categories_acl['categories'][category.pk].update(category_acl)
                if category_acl.get('can_approve_content'):
                    categories_acl['can_approve_content'].append(category.pk)

        override_acl(self.user, categories_acl)
        return categories_acl

    def assertContainsThread(self, response, thread):
        self.assertContains(response, u' href="{}"'.format(thread.get_absolute_url()))

    def assertNotContainsThread(self, response, thread):
        self.assertNotContains(response, u' href="{}"'.format(thread.get_absolute_url()))


class ApiTests(ThreadsListTestCase):
    def test_root_category(self):
        """its possible to access threads endpoint with category=ROOT_ID"""
        response = self.client.get('%s?category=%s' % (self.api_link, self.root.pk, ))
        self.assertEqual(response.status_code, 200)

    def test_explicit_first_page(self):
        """its possible to access threads endpoint with explicit first page"""
        response = self.client.get('%s?category=%s&page=1' % (self.api_link, self.root.pk, ))
        self.assertEqual(response.status_code, 200)

    def test_invalid_list_type(self):
        """api returns 404 for invalid list type"""
        response = self.client.get('%s?category=%s&list=nope' % (self.api_link, self.root.pk, ))
        self.assertEqual(response.status_code, 404)


class AllThreadsListTests(ThreadsListTestCase):
    def test_list_renders_empty(self):
        """empty threads list renders"""
        for url in LISTS_URLS:
            self.access_all_categories()

            response = self.client.get('/' + url)
            self.assertEqual(response.status_code, 200)
            self.assertContains(response, "empty-message")
            if url:
                self.assertContains(response, "No threads matching specified criteria")
            else:
                self.assertContains(response, "There are no threads on this forum")

            self.access_all_categories()

            response = self.client.get(self.category_b.get_absolute_url() + url)
            self.assertEqual(response.status_code, 200)
            self.assertContains(response, self.category_b.name)
            self.assertContains(response, "empty-message")
            if url:
                self.assertContains(response, "No threads matching specified criteria")
            else:
                self.assertContains(response, "There are no threads in this category")

            self.access_all_categories()

            response = self.client.get('%s?list=%s' % (self.api_link, url.strip('/') or 'all'))
            self.assertApiResultsAreEmpty(response)

        # empty lists render for anonymous user?
        self.logout_user()
        self.user = self.get_anonymous_user()

        self.access_all_categories()

        response = self.client.get('/')
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "empty-message")
        self.assertContains(response, "There are no threads on this forum")

        self.access_all_categories()

        response = self.client.get(self.category_b.get_absolute_url())
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, self.category_b.name)
        self.assertContains(response, "empty-message")
        self.assertContains(response, "There are no threads in this category")

        self.access_all_categories()

        response = self.client.get('%s?list=all' % self.api_link)
        self.assertApiResultsAreEmpty(response)

    def test_list_authenticated_only_views(self):
        """authenticated only views return 403 for guests"""
        for url in LISTS_URLS:
            self.access_all_categories()

            response = self.client.get('/' + url)
            self.assertEqual(response.status_code, 200)

            self.access_all_categories()

            response = self.client.get(self.category_b.get_absolute_url() + url)
            self.assertEqual(response.status_code, 200)
            self.assertContains(response, self.category_b.name)

            self.access_all_categories()

            self.access_all_categories()
            response = self.client.get(
                '%s?category=%s&list=%s' %
                (self.api_link, self.category_b.pk, url.strip('/') or 'all', )
            )
            self.assertEqual(response.status_code, 200)

        self.logout_user()
        self.user = self.get_anonymous_user()
        for url in LISTS_URLS[1:]:
            self.access_all_categories()

            response = self.client.get('/' + url)
            self.assertEqual(response.status_code, 403)

            self.access_all_categories()
            response = self.client.get(self.category_b.get_absolute_url() + url)
            self.assertEqual(response.status_code, 403)

            self.access_all_categories()
            response = self.client.get(
                '%s?category=%s&list=%s' %
                (self.api_link, self.category_b.pk, url.strip('/') or 'all', )
            )
            self.assertEqual(response.status_code, 403)

    def test_list_renders_categories_picker(self):
        """categories picker renders valid categories"""
        Category(
            name='Hidden Category',
            slug='hidden-category',
        ).insert_at(
            self.root,
            position='last-child',
            save=True,
        )
        test_category = Category.objects.get(slug='hidden-category')

        testutils.post_thread(
            category=self.category_b,
        )

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

        self.assertContains(response, 'subcategory-%s' % self.category_a.css_class)

        # readable categories, but non-accessible directly
        self.assertNotContains(response, 'subcategory-%s' % self.category_b.css_class)
        self.assertNotContains(response, 'subcategory-%s' % self.category_c.css_class)
        self.assertNotContains(response, 'subcategory-%s' % self.category_d.css_class)
        self.assertNotContains(response, 'subcategory-%s' % self.category_f.css_class)

        # hidden category
        self.assertNotContains(response, 'subcategory-%s' % test_category.css_class)

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

        response_json = response.json()
        self.assertIn(self.category_a.pk, response_json['subcategories'])
        self.assertNotIn(self.category_b.pk, response_json['subcategories'])

        # test category view
        self.access_all_categories()

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

        self.assertContains(response, 'subcategory-%s' % self.category_b.css_class)

        # readable categories, but non-accessible directly
        self.assertNotContains(response, 'subcategory-%s' % self.category_c.css_class)
        self.assertNotContains(response, 'subcategory-%s' % self.category_d.css_class)
        self.assertNotContains(response, 'subcategory-%s' % self.category_f.css_class)

        self.access_all_categories()
        response = self.client.get('%s?category=%s' % (self.api_link, self.category_a.pk))
        self.assertEqual(response.status_code, 200)

        response_json = response.json()
        self.assertEqual(response_json['subcategories'][0], self.category_b.pk)

    def test_display_pinned_threads(self):
        """
        threads list displays globally pinned threads first
        and locally ones inbetween other
        """
        globally = testutils.post_thread(
            category=self.first_category,
            is_global=True,
        )

        locally = testutils.post_thread(
            category=self.first_category,
            is_pinned=True,
        )

        standard = testutils.post_thread(category=self.first_category)

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

        content = smart_str(response.content)
        positions = {
            'g': content.find(globally.get_absolute_url()),
            'l': content.find(locally.get_absolute_url()),
            's': content.find(standard.get_absolute_url()),
        }

        # global announcement before others
        self.assertTrue(positions['g'] < positions['l'])
        self.assertTrue(positions['g'] < positions['s'])

        # standard in the middle
        self.assertTrue(positions['s'] < positions['l'])
        self.assertTrue(positions['s'] > positions['g'])

        # locally pinned last
        self.assertTrue(positions['l'] > positions['g'])
        self.assertTrue(positions['l'] > positions['s'])

        # API behaviour is identic
        response = self.client.get('/api/threads/')
        self.assertApiResultsEqual(response, [
            globally,  # global announcement before others
            standard,  # standard in the middle
            locally,  # locally pinned last
        ])

    def test_noscript_pagination(self):
        """threads list is paginated for users with js disabled"""
        threads_per_page = settings.MISAGO_THREADS_PER_PAGE

        threads = []
        for _ in range(settings.MISAGO_THREADS_PER_PAGE * 3):
            threads.append(testutils.post_thread(category=self.first_category))

        # secondary page renders
        response = self.client.get('/?page=2')
        self.assertEqual(response.status_code, 200)

        for thread in threads[:threads_per_page]:
            self.assertNotContainsThread(response, thread)
        for thread in threads[threads_per_page:threads_per_page * 2]:
            self.assertContainsThread(response, thread)
        for thread in threads[threads_per_page * 2:]:
            self.assertNotContainsThread(response, thread)

        self.assertNotContains(response, '/?page=1')
        self.assertContains(response, '/?page=3')

        # third page renders
        response = self.client.get('/?page=3')
        self.assertEqual(response.status_code, 200)

        for thread in threads[threads_per_page:]:
            self.assertNotContainsThread(response, thread)
        for thread in threads[:threads_per_page]:
            self.assertContainsThread(response, thread)

        self.assertContains(response, '/?page=2')
        self.assertNotContains(response, '/?page=4')

        # excessive page gives 404
        response = self.client.get('/?page=4')
        self.assertEqual(response.status_code, 404)


class CategoryThreadsListTests(ThreadsListTestCase):
    def test_access_hidden_category(self):
        """hidden category returns 404"""
        Category(
            name='Hidden Category',
            slug='hidden-category',
        ).insert_at(
            self.root,
            position='last-child',
            save=True,
        )
        test_category = Category.objects.get(slug='hidden-category')

        for url in LISTS_URLS:
            response = self.client.get(test_category.get_absolute_url() + url)
            self.assertEqual(response.status_code, 404)

            response = self.client.get('%s?category=%s' % (self.api_link, test_category.pk))
            self.assertEqual(response.status_code, 404)

    def test_access_protected_category(self):
        """protected category returns 403"""
        Category(
            name='Hidden Category',
            slug='hidden-category',
        ).insert_at(
            self.root,
            position='last-child',
            save=True,
        )
        test_category = Category.objects.get(slug='hidden-category')

        for url in LISTS_URLS:
            override_acl(
                self.user, {
                    'visible_categories': [test_category.pk],
                    'browseable_categories': [],
                    'categories': {
                        test_category.pk: {
                            'can_see': 1,
                            'can_browse': 0,
                        },
                    },
                }
            )
            response = self.client.get(test_category.get_absolute_url() + url)
            self.assertEqual(response.status_code, 403)

            override_acl(
                self.user, {
                    'visible_categories': [test_category.pk],
                    'browseable_categories': [],
                    'categories': {
                        test_category.pk: {
                            'can_see': 1,
                            'can_browse': 0,
                        },
                    },
                }
            )
            response = self.client.get(
                '%s?category=%s&list=%s' % (self.api_link, test_category.pk, url.strip('/'), )
            )
            self.assertEqual(response.status_code, 403)

    def test_display_pinned_threads(self):
        """
        category threads list displays globally pinned threads first
        then locally ones and unpinned last
        """
        globally = testutils.post_thread(
            category=self.first_category,
            is_global=True,
        )

        locally = testutils.post_thread(
            category=self.first_category,
            is_pinned=True,
        )

        standard = testutils.post_thread(category=self.first_category)

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

        content = smart_str(response.content)
        positions = {
            'g': content.find(globally.get_absolute_url()),
            'l': content.find(locally.get_absolute_url()),
            's': content.find(standard.get_absolute_url()),
        }

        # global announcement before others
        self.assertTrue(positions['g'] < positions['l'])
        self.assertTrue(positions['g'] < positions['s'])

        # pinned in the middle
        self.assertTrue(positions['l'] < positions['s'])
        self.assertTrue(positions['l'] > positions['g'])

        # standard last
        self.assertTrue(positions['s'] > positions['g'])
        self.assertTrue(positions['s'] > positions['g'])

        # API behaviour is identic
        response = self.client.get('/api/threads/?category=%s' % self.first_category.pk)
        self.assertEqual(response.status_code, 200)
        self.assertApiResultsEqual(response, [
            globally,  # global announcement before others
            locally,  # locally pinned in the middle
            standard,  # standard last
        ])


class ThreadsVisibilityTests(ThreadsListTestCase):
    def test_list_renders_test_thread(self):
        """list renders test thread with valid top category"""
        test_thread = testutils.post_thread(
            category=self.category_c,
        )

        response = self.client.get('/')
        self.assertContainsThread(response, test_thread)

        self.assertContains(response, 'subcategory-%s' % self.category_a.css_class)
        self.assertContains(response, 'subcategory-%s' % self.category_e.css_class)

        self.assertNotContains(response, 'thread-detail-category-%s' % self.category_a.css_class)
        self.assertContains(response, 'thread-detail-category-%s' % self.category_c.css_class)

        # api displays same data
        self.access_all_categories()
        response = self.client.get(self.api_link)
        self.assertInApiResults(response, test_thread)

        response_json = response.json()
        self.assertEqual(len(response_json['subcategories']), 3)
        self.assertIn(self.category_a.pk, response_json['subcategories'])

        # test category view
        self.access_all_categories()
        response = self.client.get(self.category_b.get_absolute_url())
        self.assertEqual(response.status_code, 200)

        # thread displays
        self.assertContainsThread(response, test_thread)

        self.assertNotContains(response, 'thread-detail-category-%s' % self.category_b.css_class)
        self.assertContains(response, 'thread-detail-category-%s' % self.category_c.css_class)

        # api displays same data
        self.access_all_categories()
        response = self.client.get('%s?category=%s' % (self.api_link, self.category_b.pk))
        self.assertInApiResults(response, test_thread)

        response_json = response.json()
        self.assertEqual(len(response_json['subcategories']), 2)
        self.assertEqual(response_json['subcategories'][0], self.category_c.pk)

    def test_list_hides_hidden_thread(self):
        """list renders empty due to no permission to see thread"""
        Category(
            name='Hidden Category',
            slug='hidden-category',
        ).insert_at(
            self.root,
            position='last-child',
            save=True,
        )

        test_category = Category.objects.get(slug='hidden-category')
        test_thread = testutils.post_thread(category=test_category)

        response = self.client.get('/')
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "empty-message")
        self.assertNotContainsThread(response, test_thread)

    def test_api_hides_hidden_thread(self):
        """api returns empty due to no permission to see thread"""
        Category(
            name='Hidden Category',
            slug='hidden-category',
        ).insert_at(
            self.root,
            position='last-child',
            save=True,
        )

        test_category = Category.objects.get(slug='hidden-category')

        testutils.post_thread(
            category=test_category,
        )

        response = self.client.get(self.api_link)
        self.assertApiResultsAreEmpty(response)

    def test_list_user_see_own_unapproved_thread(self):
        """list renders unapproved thread that belongs to viewer"""
        test_thread = testutils.post_thread(
            category=self.category_a,
            poster=self.user,
            is_unapproved=True,
        )

        response = self.client.get('/')
        self.assertEqual(response.status_code, 200)
        self.assertContainsThread(response, test_thread)

        # test api
        self.access_all_categories()
        response = self.client.get(self.api_link)
        self.assertInApiResults(response, test_thread)

    def test_list_user_cant_see_unapproved_thread(self):
        """list hides unapproved thread that belongs to other user"""
        test_thread = testutils.post_thread(
            category=self.category_a,
            is_unapproved=True,
        )

        response = self.client.get('/')
        self.assertEqual(response.status_code, 200)
        self.assertNotContainsThread(response, test_thread)

        # test api
        self.access_all_categories()
        response = self.client.get(self.api_link)
        self.assertApiResultsAreEmpty(response)

    def test_list_user_cant_see_hidden_thread(self):
        """list hides hidden thread that belongs to other user"""
        test_thread = testutils.post_thread(
            category=self.category_a,
            is_hidden=True,
        )

        response = self.client.get('/')
        self.assertEqual(response.status_code, 200)
        self.assertNotContainsThread(response, test_thread)

        # test api
        self.access_all_categories()
        response = self.client.get(self.api_link)
        self.assertApiResultsAreEmpty(response)

    def test_list_user_cant_see_own_hidden_thread(self):
        """list hides hidden thread that belongs to viewer"""
        test_thread = testutils.post_thread(
            category=self.category_a,
            poster=self.user,
            is_hidden=True,
        )

        response = self.client.get('/')
        self.assertEqual(response.status_code, 200)
        self.assertNotContainsThread(response, test_thread)

        # test api
        self.access_all_categories()
        response = self.client.get(self.api_link)
        self.assertApiResultsAreEmpty(response)

    def test_list_user_can_see_own_hidden_thread(self):
        """list shows hidden thread that belongs to viewer due to permission"""
        test_thread = testutils.post_thread(
            category=self.category_a,
            poster=self.user,
            is_hidden=True,
        )

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

        response = self.client.get('/')
        self.assertEqual(response.status_code, 200)
        self.assertContainsThread(response, test_thread)

        # test api
        self.access_all_categories({'can_hide_threads': 1})

        response = self.client.get(self.api_link)
        self.assertInApiResults(response, test_thread)

    def test_list_user_can_see_hidden_thread(self):
        """list shows hidden thread that belongs to other user due to permission"""
        test_thread = testutils.post_thread(
            category=self.category_a,
            is_hidden=True,
        )

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

        response = self.client.get('/')
        self.assertEqual(response.status_code, 200)
        self.assertContainsThread(response, test_thread)

        # test api
        self.access_all_categories({'can_hide_threads': 1})

        response = self.client.get(self.api_link)
        self.assertInApiResults(response, test_thread)

    def test_list_user_can_see_unapproved_thread(self):
        """list shows hidden thread that belongs to other user due to permission"""
        test_thread = testutils.post_thread(
            category=self.category_a,
            is_unapproved=True,
        )

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

        response = self.client.get('/')
        self.assertEqual(response.status_code, 200)
        self.assertContainsThread(response, test_thread)

        # test api
        self.access_all_categories({'can_approve_content': 1})

        response = self.client.get(self.api_link)
        self.assertInApiResults(response, test_thread)


class MyThreadsListTests(ThreadsListTestCase):
    def test_list_renders_empty(self):
        """list renders empty"""
        self.access_all_categories()

        response = self.client.get('/my/')
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "empty-message")

        self.access_all_categories()

        response = self.client.get(self.category_a.get_absolute_url() + 'my/')
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "empty-message")

        # test api
        self.access_all_categories()
        response = self.client.get('%s?list=my' % self.api_link)
        self.assertApiResultsAreEmpty(response)

        self.access_all_categories()
        response = self.client.get('%s?list=my&category=%s' % (self.api_link, self.category_a.pk))
        self.assertApiResultsAreEmpty(response)

    def test_list_renders_test_thread(self):
        """list renders only threads posted by user"""
        test_thread = testutils.post_thread(
            category=self.category_a,
            poster=self.user,
        )

        other_thread = testutils.post_thread(category=self.category_a)

        self.access_all_categories()

        response = self.client.get('/my/')
        self.assertEqual(response.status_code, 200)
        self.assertContainsThread(response, test_thread)
        self.assertNotContainsThread(response, other_thread)

        self.access_all_categories()

        response = self.client.get(self.category_a.get_absolute_url() + 'my/')
        self.assertEqual(response.status_code, 200)
        self.assertContainsThread(response, test_thread)
        self.assertNotContainsThread(response, other_thread)

        # test api
        self.access_all_categories()
        response = self.client.get('%s?list=my' % self.api_link)
        self.assertInApiResults(response, test_thread)

        self.access_all_categories()
        response = self.client.get('%s?list=my&category=%s' % (self.api_link, self.category_a.pk))
        self.assertInApiResults(response, test_thread)


class NewThreadsListTests(ThreadsListTestCase):
    def test_list_renders_empty(self):
        """list renders empty"""
        self.access_all_categories()

        response = self.client.get('/new/')
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "empty-message")

        self.access_all_categories()

        response = self.client.get(self.category_a.get_absolute_url() + 'new/')
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "empty-message")

        # test api
        self.access_all_categories()
        response = self.client.get('%s?list=new' % self.api_link)
        self.assertApiResultsAreEmpty(response)

        self.access_all_categories()
        response = self.client.get('%s?list=new&category=%s' % (self.api_link, self.category_a.pk))
        self.assertApiResultsAreEmpty(response)

    def test_list_renders_new_thread(self):
        """list renders new thread"""
        test_thread = testutils.post_thread(category=self.category_a)

        self.access_all_categories()

        response = self.client.get('/new/')
        self.assertEqual(response.status_code, 200)
        self.assertContainsThread(response, test_thread)

        self.access_all_categories()

        response = self.client.get(self.category_a.get_absolute_url() + 'new/')
        self.assertEqual(response.status_code, 200)
        self.assertContainsThread(response, test_thread)

        # test api
        self.access_all_categories()
        response = self.client.get('%s?list=new' % self.api_link)
        self.assertInApiResults(response, test_thread)

        self.access_all_categories()
        response = self.client.get('%s?list=new&category=%s' % (self.api_link, self.category_a.pk))
        self.assertInApiResults(response, test_thread)

    def test_list_renders_thread_bumped_after_user_cutoff(self):
        """list renders new thread bumped after user cutoff"""
        self.user.joined_on = timezone.now() - timedelta(days=10)
        self.user.save()

        test_thread = testutils.post_thread(
            category=self.category_a,
            started_on=self.user.joined_on - timedelta(days=2),
        )

        testutils.reply_thread(
            test_thread,
            posted_on=self.user.joined_on + timedelta(days=4),
        )

        self.access_all_categories()

        response = self.client.get('/new/')
        self.assertEqual(response.status_code, 200)
        self.assertContainsThread(response, test_thread)

        self.access_all_categories()

        response = self.client.get(self.category_a.get_absolute_url() + 'new/')
        self.assertEqual(response.status_code, 200)
        self.assertContainsThread(response, test_thread)

        # test api
        self.access_all_categories()
        response = self.client.get('%s?list=new' % self.api_link)
        self.assertInApiResults(response, test_thread)

        self.access_all_categories()
        response = self.client.get('%s?list=new&category=%s' % (self.api_link, self.category_a.pk))
        self.assertInApiResults(response, test_thread)

    def test_list_hides_global_cutoff_thread(self):
        """list hides thread started before global cutoff"""
        self.user.joined_on = timezone.now() - timedelta(days=10)
        self.user.save()

        test_thread = testutils.post_thread(
            category=self.category_a,
            started_on=timezone.now() - timedelta(days=settings.MISAGO_READTRACKER_CUTOFF + 1),
        )

        self.access_all_categories()

        response = self.client.get('/new/')
        self.assertEqual(response.status_code, 200)
        self.assertNotContainsThread(response, test_thread)

        self.access_all_categories()

        response = self.client.get(self.category_a.get_absolute_url() + 'new/')
        self.assertEqual(response.status_code, 200)
        self.assertNotContainsThread(response, test_thread)

        # test api
        self.access_all_categories()
        response = self.client.get('%s?list=new' % self.api_link)
        self.assertNotInApiResults(response, test_thread)

        self.access_all_categories()
        response = self.client.get('%s?list=new&category=%s' % (self.api_link, self.category_a.pk))
        self.assertNotInApiResults(response, test_thread)

    def test_list_hides_user_cutoff_thread(self):
        """list hides thread started before users cutoff"""
        self.user.joined_on = timezone.now() - timedelta(days=5)
        self.user.save()

        test_thread = testutils.post_thread(
            category=self.category_a,
            started_on=self.user.joined_on - timedelta(minutes=1),
        )

        self.access_all_categories()

        response = self.client.get('/new/')
        self.assertEqual(response.status_code, 200)
        self.assertNotContainsThread(response, test_thread)

        self.access_all_categories()

        response = self.client.get(self.category_a.get_absolute_url() + 'new/')
        self.assertEqual(response.status_code, 200)
        self.assertNotContainsThread(response, test_thread)

        # test api
        self.access_all_categories()
        response = self.client.get('%s?list=new' % self.api_link)
        self.assertNotInApiResults(response, test_thread)

        self.access_all_categories()
        response = self.client.get('%s?list=new&category=%s' % (self.api_link, self.category_a.pk))
        self.assertNotInApiResults(response, test_thread)

    def test_list_hides_user_read_thread(self):
        """list hides thread already read by user"""
        self.user.joined_on = timezone.now() - timedelta(days=5)
        self.user.save()

        test_thread = testutils.post_thread(category=self.category_a)

        poststracker.save_read(self.user, test_thread.first_post)

        self.access_all_categories()

        response = self.client.get('/new/')
        self.assertEqual(response.status_code, 200)
        self.assertNotContainsThread(response, test_thread)

        self.access_all_categories()

        response = self.client.get(self.category_a.get_absolute_url() + 'new/')
        self.assertEqual(response.status_code, 200)
        self.assertNotContainsThread(response, test_thread)

        # test api
        self.access_all_categories()
        response = self.client.get('%s?list=new' % self.api_link)
        self.assertNotInApiResults(response, test_thread)

        self.access_all_categories()
        response = self.client.get('%s?list=new&category=%s' % (self.api_link, self.category_a.pk))
        self.assertNotInApiResults(response, test_thread)


class UnreadThreadsListTests(ThreadsListTestCase):
    def test_list_renders_empty(self):
        """list renders empty"""
        self.access_all_categories()

        response = self.client.get('/unread/')
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "empty-message")

        self.access_all_categories()

        response = self.client.get(self.category_a.get_absolute_url() + 'unread/')
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "empty-message")

        # test api
        self.access_all_categories()
        response = self.client.get('%s?list=unread' % self.api_link)
        self.assertApiResultsAreEmpty(response)

        self.access_all_categories()
        response = self.client.get(
            '%s?list=unread&category=%s' % (self.api_link, self.category_a.pk)
        )
        self.assertApiResultsAreEmpty(response)

    def test_list_renders_unread_thread(self):
        """list renders thread with unread posts"""
        self.user.joined_on = timezone.now() - timedelta(days=5)
        self.user.save()

        test_thread = testutils.post_thread(category=self.category_a)

        poststracker.save_read(self.user, test_thread.first_post)

        testutils.reply_thread(test_thread)

        self.access_all_categories()

        response = self.client.get('/unread/')
        self.assertEqual(response.status_code, 200)
        self.assertContainsThread(response, test_thread)

        self.access_all_categories()

        response = self.client.get(self.category_a.get_absolute_url() + 'unread/')
        self.assertEqual(response.status_code, 200)
        self.assertContainsThread(response, test_thread)

        # test api
        self.access_all_categories()
        response = self.client.get('%s?list=unread' % self.api_link)
        self.assertInApiResults(response, test_thread)

        self.access_all_categories()
        response = self.client.get(
            '%s?list=unread&category=%s' % (self.api_link, self.category_a.pk)
        )
        self.assertInApiResults(response, test_thread)

    def test_list_hides_never_read_thread(self):
        """list hides never read thread"""
        self.user.joined_on = timezone.now() - timedelta(days=5)
        self.user.save()

        test_thread = testutils.post_thread(category=self.category_a)

        self.access_all_categories()

        response = self.client.get('/unread/')
        self.assertEqual(response.status_code, 200)
        self.assertNotContainsThread(response, test_thread)

        self.access_all_categories()

        response = self.client.get(self.category_a.get_absolute_url() + 'unread/')
        self.assertEqual(response.status_code, 200)
        self.assertNotContainsThread(response, test_thread)

        # test api
        self.access_all_categories()
        response = self.client.get('%s?list=unread' % self.api_link)
        self.assertNotInApiResults(response, test_thread)

        self.access_all_categories()
        response = self.client.get(
            '%s?list=unread&category=%s' % (self.api_link, self.category_a.pk)
        )
        self.assertNotInApiResults(response, test_thread)

    def test_list_hides_read_thread(self):
        """list hides read thread"""
        self.user.joined_on = timezone.now() - timedelta(days=5)
        self.user.save()

        test_thread = testutils.post_thread(category=self.category_a)

        poststracker.save_read(self.user, test_thread.first_post)

        self.access_all_categories()

        response = self.client.get('/unread/')
        self.assertEqual(response.status_code, 200)
        self.assertNotContainsThread(response, test_thread)

        self.access_all_categories()

        response = self.client.get(self.category_a.get_absolute_url() + 'unread/')
        self.assertEqual(response.status_code, 200)
        self.assertNotContainsThread(response, test_thread)

        # test api
        self.access_all_categories()
        response = self.client.get('%s?list=unread' % self.api_link)
        self.assertNotInApiResults(response, test_thread)

        self.access_all_categories()
        response = self.client.get(
            '%s?list=unread&category=%s' % (self.api_link, self.category_a.pk)
        )
        self.assertNotInApiResults(response, test_thread)

    def test_list_hides_global_cutoff_thread(self):
        """list hides thread replied before global cutoff"""
        self.user.joined_on = timezone.now() - timedelta(days=10)
        self.user.save()

        test_thread = testutils.post_thread(
            category=self.category_a,
            started_on=timezone.now() - timedelta(days=settings.MISAGO_READTRACKER_CUTOFF + 5),
        )

        poststracker.save_read(self.user, test_thread.first_post)

        testutils.reply_thread(test_thread, posted_on=test_thread.started_on + timedelta(days=1))

        self.access_all_categories()

        response = self.client.get('/unread/')
        self.assertEqual(response.status_code, 200)
        self.assertNotContainsThread(response, test_thread)

        self.access_all_categories()

        response = self.client.get(self.category_a.get_absolute_url() + 'unread/')
        self.assertEqual(response.status_code, 200)
        self.assertNotContainsThread(response, test_thread)

        # test api
        self.access_all_categories()
        response = self.client.get('%s?list=unread' % self.api_link)
        self.assertNotInApiResults(response, test_thread)

        self.access_all_categories()
        response = self.client.get(
            '%s?list=unread&category=%s' % (self.api_link, self.category_a.pk)
        )
        self.assertNotInApiResults(response, test_thread)

    def test_list_hides_user_cutoff_thread(self):
        """list hides thread replied before user cutoff"""
        self.user.joined_on = timezone.now() - timedelta(days=10)
        self.user.save()

        test_thread = testutils.post_thread(
            category=self.category_a,
            started_on=self.user.joined_on - timedelta(days=2),
        )

        poststracker.save_read(self.user, test_thread.first_post)

        testutils.reply_thread(
            test_thread,
            posted_on=test_thread.started_on + timedelta(days=1),
        )

        self.access_all_categories()

        response = self.client.get('/unread/')
        self.assertEqual(response.status_code, 200)
        self.assertNotContainsThread(response, test_thread)

        self.access_all_categories()

        response = self.client.get(self.category_a.get_absolute_url() + 'unread/')
        self.assertEqual(response.status_code, 200)
        self.assertNotContainsThread(response, test_thread)

        # test api
        self.access_all_categories()
        response = self.client.get('%s?list=unread' % self.api_link)
        self.assertNotInApiResults(response, test_thread)

        self.access_all_categories()
        response = self.client.get(
            '%s?list=unread&category=%s' % (self.api_link, self.category_a.pk)
        )
        self.assertNotInApiResults(response, test_thread)


class SubscribedThreadsListTests(ThreadsListTestCase):
    def test_list_shows_subscribed_thread(self):
        """list shows subscribed thread"""
        test_thread = testutils.post_thread(category=self.category_a)
        self.user.subscription_set.create(
            thread=test_thread,
            category=self.category_a,
            last_read_on=test_thread.last_post_on,
        )

        self.access_all_categories()

        response = self.client.get('/subscribed/')
        self.assertEqual(response.status_code, 200)
        self.assertContainsThread(response, test_thread)

        self.access_all_categories()

        response = self.client.get(self.category_a.get_absolute_url() + 'subscribed/')
        self.assertEqual(response.status_code, 200)
        self.assertContainsThread(response, test_thread)

        # test api
        self.access_all_categories()
        response = self.client.get('%s?list=subscribed' % self.api_link)
        self.assertInApiResults(response, test_thread)

        self.access_all_categories()
        response = self.client.get(
            '%s?list=subscribed&category=%s' % (self.api_link, self.category_a.pk)
        )
        self.assertInApiResults(response, test_thread)

    def test_list_hides_unsubscribed_thread(self):
        """list shows subscribed thread"""
        test_thread = testutils.post_thread(category=self.category_a)

        self.access_all_categories()

        response = self.client.get('/subscribed/')
        self.assertEqual(response.status_code, 200)
        self.assertNotContainsThread(response, test_thread)

        self.access_all_categories()

        response = self.client.get(self.category_a.get_absolute_url() + 'subscribed/')
        self.assertEqual(response.status_code, 200)
        self.assertNotContainsThread(response, test_thread)

        # test api
        self.access_all_categories()
        response = self.client.get('%s?list=subscribed' % self.api_link)
        self.assertApiResultsAreEmpty(response)

        self.access_all_categories()
        response = self.client.get(
            '%s?list=subscribed&category=%s' % (self.api_link, self.category_a.pk)
        )
        self.assertApiResultsAreEmpty(response)

class UnapprovedListTests(ThreadsListTestCase):
    def test_list_errors_without_permission(self):
        """list errors if user has no permission to access it"""
        TEST_URLS = (
            '/unapproved/', self.category_a.get_absolute_url() + 'unapproved/',
            '%s?list=unapproved' % self.api_link,
        )

        for test_url in TEST_URLS:
            self.access_all_categories()
            response = self.client.get(test_url)
            self.assertEqual(response.status_code, 403)

        # approval perm has no influence on visibility
        for test_url in TEST_URLS:
            self.access_all_categories({'can_approve_content': True})

            self.access_all_categories()
            response = self.client.get(test_url)
            self.assertEqual(response.status_code, 403)

        # approval perm has no influence on visibility
        for test_url in TEST_URLS:
            self.access_all_categories(base_acl={
                'can_see_unapproved_content_lists': True,
            })

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

    def test_list_shows_all_threads_for_approving_user(self):
        """list shows all threads with unapproved posts when user has perm"""
        visible_thread = testutils.post_thread(
            category=self.category_b,
            is_unapproved=True,
        )

        hidden_thread = testutils.post_thread(
            category=self.category_b,
            is_unapproved=False,
        )

        self.access_all_categories({
            'can_approve_content': True,
        }, {
            'can_see_unapproved_content_lists': True,
        })

        response = self.client.get('/unapproved/')
        self.assertEqual(response.status_code, 200)
        self.assertContainsThread(response, visible_thread)
        self.assertNotContainsThread(response, hidden_thread)

        self.access_all_categories({
            'can_approve_content': True
        }, {
            'can_see_unapproved_content_lists': True,
        })

        response = self.client.get(self.category_a.get_absolute_url() + 'unapproved/')
        self.assertEqual(response.status_code, 200)
        self.assertContainsThread(response, visible_thread)
        self.assertNotContainsThread(response, hidden_thread)

        # test api
        self.access_all_categories({
            'can_approve_content': True
        }, {
            'can_see_unapproved_content_lists': True,
        })

        response = self.client.get('%s?list=unapproved' % self.api_link)
        self.assertInApiResults(response, visible_thread)

    def test_list_shows_owned_threads_for_unapproving_user(self):
        """list shows owned threads with unapproved posts for user without perm"""
        visible_thread = testutils.post_thread(
            poster=self.user,
            category=self.category_b,
            is_unapproved=True,
        )

        hidden_thread = testutils.post_thread(
            category=self.category_b,
            is_unapproved=True,
        )

        self.access_all_categories(base_acl={
            'can_see_unapproved_content_lists': True,
        })
        response = self.client.get('/unapproved/')
        self.assertEqual(response.status_code, 200)
        self.assertContainsThread(response, visible_thread)
        self.assertNotContainsThread(response, hidden_thread)

        self.access_all_categories(base_acl={
            'can_see_unapproved_content_lists': True,
        })
        response = self.client.get(self.category_a.get_absolute_url() + 'unapproved/')
        self.assertEqual(response.status_code, 200)
        self.assertContainsThread(response, visible_thread)
        self.assertNotContainsThread(response, hidden_thread)

        # test api
        self.access_all_categories(base_acl={
            'can_see_unapproved_content_lists': True,
        })
        response = self.client.get('%s?list=unapproved' % self.api_link)
        self.assertInApiResults(response, visible_thread)


class OwnerOnlyThreadsVisibilityTests(AuthenticatedUserTestCase):
    def setUp(self):
        super(OwnerOnlyThreadsVisibilityTests, self).setUp()

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

    def override_acl(self, user):
        category_acl = user.acl_cache['categories'][self.category.pk].copy()
        category_acl.update({'can_see_all_threads': 0})
        user.acl_cache['categories'][self.category.pk] = category_acl

        override_acl(user, user.acl_cache)

    def test_owned_threads_visibility(self):
        """only user-posted threads are visible in category"""
        self.override_acl(self.user)

        visible_thread = testutils.post_thread(
            poster=self.user,
            category=self.category,
            is_unapproved=True,
        )

        hidden_thread = testutils.post_thread(
            category=self.category,
            is_unapproved=True,
        )

        response = self.client.get(self.category.get_absolute_url())

        self.assertEqual(response.status_code, 200)
        self.assertContains(response, visible_thread.get_absolute_url())
        self.assertNotContains(response, hidden_thread.get_absolute_url())

    def test_owned_threads_visibility_anonymous(self):
        """anons can't see any threads in limited visibility category"""
        self.logout_user()

        self.override_acl(AnonymousUser())

        user_thread = testutils.post_thread(
            poster=self.user,
            category=self.category,
            is_unapproved=True,
        )

        guest_thread = testutils.post_thread(
            category=self.category,
            is_unapproved=True,
        )

        response = self.client.get(self.category.get_absolute_url())

        self.assertEqual(response.status_code, 200)
        self.assertNotContains(response, user_thread.get_absolute_url())
        self.assertNotContains(response, guest_thread.get_absolute_url())