def test_sphinx_clickthrough_post(self):
        """Test Sphinx clickthrough write API."""
        u = user(save=True)
        add_permission(u, Metric, 'add_metric')

        click_kind, search_kind = self._make_sphinx_metric_kinds()

        # POST the new object:
        url = reverse('api_dispatch_list',
                      kwargs={'resource_name': 'sphinx-clickthrough-rate',
                              'api_name': 'v1'})
        auth = 'Basic ' + b64encode('%s:%s' % (u.username, 'testpass'))
        response = self.client.post(url,
                                    json.dumps({'start': '2000-01-02',
                                                'searches': 1e8,
                                                'clicks': 5e7}),
                                    content_type='application/json',
                                    HTTP_AUTHORIZATION=auth)
        eq_(response.status_code, 201)

        # Do a GET, and see if the round trip worked:
        response = self.client.get(url + '?format=json')
        self.assertContains(  # Beware of dict order changes someday.
            response,
            '"objects": [{"clicks": 50000000, "resource_uri": "", '
                         '"searches": 100000000, "start": "2000-01-02"}]')

    def setUp(self, switch_is_active):
        switch_is_active.return_value = True

        super(KarmaAPITests, self).setUp()

        try:
            self.mgr = KarmaManager()
            redis_client('karma').flushdb()
        except RedisError:
            raise SkipTest

        self.user1 = user(save=True)
        self.user2 = user(save=True)
        self.user3 = user(save=True)

        TestAction1(user=self.user1).save()
        TestAction2(user=self.user2).save()
        TestAction2(user=self.user2).save()
        TestAction1(user=self.user3).save()
        TestAction1(user=self.user3).save()
        TestAction1(user=self.user3).save()
        self.mgr.update_top()

        self.client.login(username=self.user1.username, password='testpass')
        add_permission(self.user1, models.Title, 'view_dashboard')

    def test_vote(self):
        """Test vote API call."""
        u = user(save=True)
        add_permission(u, Profile, 'view_kpi_dashboard')

        r = revision(save=True)
        helpful_vote(revision=r, save=True)
        helpful_vote(revision=r, save=True)
        helpful_vote(revision=r, helpful=True, save=True)

        a = answer(save=True)
        answer_vote(answer=a, save=True)
        answer_vote(answer=a, helpful=True, save=True)
        answer_vote(answer=a, helpful=True, save=True)

        url = reverse('api_dispatch_list',
                      kwargs={'resource_name': 'kpi_vote',
                              'api_name': 'v1'})
        self.client.login(username=u.username, password='testpass')
        response = self.client.get(url + '?format=json')
        eq_(200, response.status_code)
        r = json.loads(response.content)
        eq_(r['objects'][0]['kb_helpful'], 1)
        eq_(r['objects'][0]['kb_votes'], 3)
        eq_(r['objects'][0]['ans_helpful'], 2)
        eq_(r['objects'][0]['ans_votes'], 3)

    def test_needs_change(self):
        """Test setting and unsetting the needs change flag"""
        # Create a new document and edit it, setting needs_change.
        comment = 'Please update for Firefix.next'
        doc = revision(save=True).document
        data = new_document_data()
        data.update({'needs_change': True,
                     'needs_change_comment': comment,
                     'form': 'doc'})

        # Verify that needs_change can't be set if the user doesn't have
        # the permission.
        self.client.post(reverse('wiki.edit_document', args=[doc.slug]), data)
        doc = Document.uncached.get(pk=doc.pk)
        assert not doc.needs_change
        assert not doc.needs_change_comment

        # Give the user permission, now it should work.
        add_permission(self.u, Document, 'edit_needs_change')
        self.client.post(reverse('wiki.edit_document', args=[doc.slug]), data)
        doc = Document.uncached.get(pk=doc.pk)
        assert doc.needs_change
        eq_(comment, doc.needs_change_comment)

        # Clear out needs_change.
        data.update({'needs_change': False,
                     'needs_change_comment': comment})
        self.client.post(reverse('wiki.edit_document', args=[doc.slug]), data)
        doc = Document.uncached.get(pk=doc.pk)
        assert not doc.needs_change
        eq_('', doc.needs_change_comment)

 def setUp(self):
     super(AddRemoveLeaderTests, self).setUp()
     self.locale = locale(locale='es', save=True)
     self.user = user(save=True)
     add_permission(self.user, Locale, 'change_locale')
     self.leader = user(save=True)
     self.client.login(username=self.user.username, password='testpass')

 def setUp(self):
     super(AddRemoveLeaderTests, self).setUp()
     self.user = user(save=True)
     add_permission(self.user, GroupProfile, 'change_groupprofile')
     self.leader = user(save=True)
     self.group_profile = group_profile(group=group(save=True), save=True)
     self.client.login(username=self.user.username, password='testpass')

    def test_percent(self, switch_is_active):
        """Test user API with all defaults."""
        switch_is_active.return_value = True
        u = user()
        u.save()
        add_permission(u, Profile, 'view_kpi_dashboard')
        question = Question(title='Test Question',
                            content='Lorem Ipsum Dolor',
                            creator_id=u.id)
        question.save()
        answer = Answer(question=question, creator_id=u.id,
                        content="Test Answer")
        answer.save()

        question.solution = answer
        question.save()

        url = reverse('api_dispatch_list',
                      kwargs={'resource_name': 'kpi_solution',
                              'api_name': 'v1'})
        self.client.login(username=u.username, password='testpass')
        response = self.client.get(url + '?format=json')
        eq_(200, response.status_code)
        r = json.loads(response.content)
        eq_(r['objects'][0]['with_solutions'], 1)
        eq_(r['objects'][0]['without_solutions'], 0)

 def setUp(self):
     super(AddRemoveContributorTests, self).setUp()
     self.user = user(save=True)
     self.contributor = user(save=True)
     add_permission(self.user, Document, 'change_document')
     self.client.login(username=self.user.username, password='testpass')
     self.revision = revision(save=True)
     self.document = self.revision.document

    def test_delete_image(self):
        """Deleting an uploaded image works."""
        im = image()
        u = user(save=True)
        add_permission(u, Image, 'delete_image')
        self.client.login(username=u.username, password='testpass')
        r = post(self.client, 'gallery.delete_media', args=['image', im.id])

        eq_(200, r.status_code)
        eq_(0, Image.objects.count())

 def test_solution_with_perm(self):
     """Test marking a solution with 'change_solution' permission."""
     u = user(save=True)
     add_permission(u, Question, 'change_solution')
     self.client.login(username=u.username, password='testpass')
     answer = self.question.answers.all()[0]
     post(self.client, 'questions.solution',
          args=[self.question.id, answer.id])
     q = Question.uncached.get(pk=self.question.id)
     eq_(q.solution, answer)

    def test_edit_image_with_permissions(self):
        """Editing image sets the updated_by field."""
        img = image()
        u = user(save=True)
        add_permission(u, Image, 'change_image')
        self.client.login(username=u.username, password='testpass')
        r = post(self.client, 'gallery.edit_media', {'description': 'arrr'},
                 args=['image', img.id])

        eq_(200, r.status_code)
        eq_(u.username, Image.objects.get().updated_by.username)

    def test_schedule_rebuild_kb_on_delete(self, schedule_rebuild_kb):
        """KB rebuild scheduled on delete"""
        im = image()
        u = user(save=True)
        add_permission(u, Image, 'delete_image')
        self.client.login(username=u.username, password='testpass')
        r = post(self.client, 'gallery.delete_media', args=['image', im.id])

        eq_(200, r.status_code)
        eq_(0, Image.objects.count())
        assert schedule_rebuild_kb.called

    def setUp(self):
        super(FlaggedQueueTestCase, self).setUp()
        q = question(creator=user(save=True), save=True)
        self.answer = answer(question=q,
                             creator=user(save=True),
                             save=True)

        self.flagger = user(save=True)
        u = user(save=True)
        add_permission(u, FlaggedObject, 'can_moderate')

        self.client.login(username=u.username, password='testpass')

 def test_flag_kbforum_post(self):
     p = Post.objects.all()[0]
     f = FlaggedObject(content_object=p, reason='spam', creator_id=118577)
     f.save()
     # Make sure flagit queue page works
     u = user(save=True)
     add_permission(u, FlaggedObject, 'can_moderate')
     self.client.login(username=u.username, password='testpass')
     response = get(self.client, 'flagit.queue')
     eq_(200, response.status_code)
     doc = pq(response.content)
     eq_(1, len(doc('#flagged-queue li')))

    def test_permission_required(self):
        url = reverse('flagit.queue', force_locale=True)
        resp = self.client.get(url)
        eq_(302, resp.status_code)

        self.client.login(username=self.user.username, password='testpass')
        resp = self.client.get(url)
        eq_(403, resp.status_code)


        add_permission(self.user, FlaggedObject, 'can_moderate')
        resp = self.client.get(url)
        eq_(200, resp.status_code)

 def test_solve_unsolve_with_perm(self):
     """Test marking solve/unsolve with 'change_solution' permission."""
     u = user(save=True)
     add_permission(u, Question, "change_solution")
     self.client.login(username=u.username, password="testpass")
     answer = self.question.answers.all()[0]
     # Solve and verify
     post(self.client, "questions.solve", args=[self.question.id, answer.id])
     q = Question.uncached.get(pk=self.question.id)
     eq_(q.solution, answer)
     # Unsolve and verify
     post(self.client, "questions.unsolve", args=[self.question.id, answer.id])
     q = Question.uncached.get(pk=self.question.id)
     eq_(q.solution, None)

 def test_flag_kbforum_post(self):
     u = user(save=True)
     t = thread(save=True)
     p = t.new_post(creator=u, content='foo')
     f = FlaggedObject(content_object=p, reason='spam', creator_id=u.id)
     f.save()
     # Make sure flagit queue page works
     u2 = user(save=True)
     add_permission(u2, FlaggedObject, 'can_moderate')
     self.client.login(username=u2.username, password='testpass')
     response = get(self.client, 'flagit.queue')
     eq_(200, response.status_code)
     doc = pq(response.content)
     eq_(1, len(doc('#flagged-queue li')))

    def test_flagged_and_deleted_profile(self):
        # Flag a profile and delete it
        p = User.objects.get(id=47963).get_profile()
        f = FlaggedObject(content_object=p, reason='spam', creator_id=118577)
        f.save()
        p.delete()

        # Verify flagit queue
        u = user(save=True)
        add_permission(u, FlaggedObject, 'can_moderate')
        self.client.login(username=u.username, password='testpass')
        response = get(self.client, 'flagit.queue')
        eq_(200, response.status_code)
        doc = pq(response.content)
        eq_(1, len(doc('#flagged-queue form.update')))

    def test_edit_post_moderator(self):
        """Editing post as a moderator works."""
        u = user(save=True)
        add_permission(u, Post, 'change_post')
        self.client.login(username=u.username, password='testpass')

        p = post_(save=True)
        t = p.thread
        d = t.document

        r = post(self.client, 'wiki.discuss.edit_post',
                 {'content': 'More new content'}, args=[d.slug, t.id, p.id])
        eq_(200, r.status_code)

        edited_p = Post.uncached.get(pk=p.pk)
        eq_('More new content', edited_p.content)

 def setUp(self):
     super(KBBelongsTestCase, self).setUp()
     u = user(save=True)
     self.doc = document(title='spam', save=True)
     self.doc_2 = document(title='eggs', save=True)
     self.thread = thread(creator=u, document=self.doc, is_locked=False,
                          save=True)
     self.thread_2 = thread(creator=u, document=self.doc_2, is_locked=False,
                            save=True)
     permissions = ('sticky_thread', 'lock_thread', 'delete_thread',
                    'delete_post')
     for permission in permissions:
         add_permission(u, self.thread, permission)
     self.post = self.thread.new_post(creator=self.thread.creator,
                                      content='foo')
     self.client.login(username=u.username, password='testpass')