plone.app.discussion/plone/app/discussion/tests/test_moderation_view.py

# -*- coding: utf-8 -*-
import unittest

from DateTime import DateTime

from zope.component import createObject

from Products.CMFCore.utils import getToolByName
from Products.PloneTestCase.ptc import PloneTestCase

from plone.app.discussion.browser.moderation import View
from plone.app.discussion.browser.moderation import BulkActionsView
from plone.app.discussion.interfaces import IConversation
from plone.app.discussion.tests.layer import DiscussionLayer


class ModerationViewTest(PloneTestCase):

    layer = DiscussionLayer

    def afterSetUp(self):
        self.loginAsPortalOwner()
        typetool = self.portal.portal_types
        typetool.constructContent('Document', self.portal, 'doc1')
        
        self.portal_discussion = getToolByName(self.portal,
                                               'portal_discussion',
                                               None)
        self.membership_tool = getToolByName(self.folder,
                                             'portal_membership')
        self.memberdata = self.portal.portal_memberdata
        request = self.app.REQUEST
        context = getattr(self.portal, 'doc1')
        self.view = View(context, request)
        self.view.__of__(context)
        self.portal.portal_workflow.setChainForPortalTypes(
            ('Discussion Item',), 'comment_review_workflow')
        self.wf_tool = self.portal.portal_workflow

        # Add a conversation with three comments

        conversation = IConversation(self.portal.doc1)

        comment1 = createObject('plone.Comment')
        comment1.title = 'Comment 1'
        comment1.text = 'Comment text'
        comment1.Creator = 'Jim'
        new_id_1 = conversation.addComment(comment1)
        self.comment1 = self.portal.doc1.restrictedTraverse(\
                            '++conversation++default/%s' % new_id_1)

        comment2 = createObject('plone.Comment')
        comment2.title = 'Comment 2'
        comment2.text = 'Comment text'
        comment2.Creator = 'Joe'
        new_id_2 = conversation.addComment(comment2)
        self.comment2 = self.portal.doc1.restrictedTraverse(\
                            '++conversation++default/%s' % new_id_2)

        comment3 = createObject('plone.Comment')
        comment3.title = 'Comment 3'
        comment3.text = 'Comment text'
        comment3.Creator = 'Emma'
        new_id_3 = conversation.addComment(comment3)
        self.comment3 = self.portal.doc1.restrictedTraverse(\
                            '++conversation++default/%s' % new_id_3)

    def test_moderation_enabled(self):
        """Make sure that moderation_enabled returns true if the comment 
           workflow implements a 'pending' state.
        """
        # The one_state_workflow does not have a 'pending' state
        self.wf_tool.setChainForPortalTypes(('Discussion Item',),
                                            ('one_state_workflow,'))
        self.assertEquals(self.view.moderation_enabled(), False)
        # The comment_review_workflow does have a 'pending' state
        self.wf_tool.setChainForPortalTypes(('Discussion Item',),
                                            ('comment_review_workflow,'))
        self.assertEquals(self.view.moderation_enabled(), True)

    def test_old_comments_not_shown_in_moderation_view(self):
        # Create an old comment and make sure it is not shown
        # in the moderation view.
        
        # Create old comment
        discussion = getToolByName(self.portal, 'portal_discussion', None)
        discussion.overrideDiscussionFor(self.portal.doc1, 1)        
        talkback = discussion.getDiscussionFor(self.portal.doc1)
        self.portal.doc1.talkback.createReply('My Title', 'My Text', Creator='Jim')
        reply = talkback.getReplies()[0]
        reply.setReplyTo(self.portal.doc1)
        reply.creation_date = DateTime(2003, 3, 11, 9, 28, 6)
        reply.modification_date = DateTime(2009, 7, 12, 19, 38, 7)
        self.assertEquals(reply.Title(), 'My Title')
        self.assertEquals(reply.EditableBody(), 'My Text')
        self.failUnless('Jim' in reply.listCreators())
        self.assertEquals(talkback.replyCount(self.portal.doc1), 1)
        self.assertEquals(reply.inReplyTo(), self.portal.doc1)
        
        # Make sure only the two new comments are shown
        self.view()
        self.assertEquals(len(self.view.comments), 3)


class ModerationBulkActionsViewTest(PloneTestCase):

    layer = DiscussionLayer

    def afterSetUp(self):
        self.loginAsPortalOwner()
        typetool = self.portal.portal_types
        typetool.constructContent('Document', self.portal, 'doc1')
        self.wf = getToolByName(self.portal, 
                                'portal_workflow',
                                None)
        
        self.request = self.app.REQUEST
        self.context = self.portal
        self.portal.portal_workflow.setChainForPortalTypes(
            ('Discussion Item',), 'comment_review_workflow')
        self.wf_tool = self.portal.portal_workflow

        # Add a conversation with three comments

        conversation = IConversation(self.portal.doc1)

        comment1 = createObject('plone.Comment')
        comment1.title = 'Comment 1'
        comment1.text = 'Comment text'
        comment1.Creator = 'Jim'
        new_id_1 = conversation.addComment(comment1)
        self.comment1 = self.portal.doc1.restrictedTraverse(\
                            '++conversation++default/%s' % new_id_1)

        comment2 = createObject('plone.Comment')
        comment2.title = 'Comment 2'
        comment2.text = 'Comment text'
        comment2.Creator = 'Joe'
        new_id_2 = conversation.addComment(comment2)
        self.comment2 = self.portal.doc1.restrictedTraverse(\
                            '++conversation++default/%s' % new_id_2)

        comment3 = createObject('plone.Comment')
        comment3.title = 'Comment 3'
        comment3.text = 'Comment text'
        comment3.Creator = 'Emma'
        new_id_3 = conversation.addComment(comment3)
        self.comment3 = self.portal.doc1.restrictedTraverse(\
                            '++conversation++default/%s' % new_id_3)

        self.conversation = conversation
    
    def test_default_bulkaction(self):
        # Make sure no error is raised when no bulk actions has been supplied
        self.request = self.app.REQUEST
        self.context = self.portal
        self.request.set('form.select.BulkAction', '-1')
        self.request.set('paths', ['/'.join(self.comment1.getPhysicalPath())])
        view = BulkActionsView(self.context, self.request)
        self.failIf(view())
    
    def test_retract(self):
        self.request = self.app.REQUEST
        self.context = self.portal
        self.request.set('form.select.BulkAction', 'retract')
        self.request.set('paths', ['/'.join(self.comment1.getPhysicalPath())])
        view = BulkActionsView(self.context, self.request)
        
        self.assertRaises(NotImplementedError,
                          view)

    def test_publish(self):
        self.request = self.app.REQUEST
        self.context = self.portal
        self.request.set('form.select.BulkAction', 'publish')
        self.request.set('paths', ['/'.join(self.comment1.getPhysicalPath())]) 
        view = BulkActionsView(self.context, self.request)
        view()
        
        # Count published comments
        published_comments = 0
        for r in self.conversation.getThreads():
            comment_obj = r['comment']
            workflow_status = self.wf.getInfoFor(comment_obj, 'review_state')
            if workflow_status == 'published':
                published_comments += 1
        
        # Make sure the comment has been published
        self.assertEquals(published_comments, 1)
        
    def test_mark_as_spam(self):
        self.request = self.app.REQUEST
        self.context = self.portal
        self.request.set('form.select.BulkAction', 'mark_as_spam')
        self.request.set('paths', ['/'.join(self.comment1.getPhysicalPath())])
        view = BulkActionsView(self.context, self.request)
        
        self.assertRaises(NotImplementedError,
                          view)

    def test_delete(self):
        self.request = self.app.REQUEST
        self.context = self.app

        # Initially we have three comments
        self.assertEquals(self.conversation.total_comments, 3)
        
        # Delete two comments with bulk actions
        self.request.set('form.select.BulkAction', 'delete')
        self.request.set('paths', ['/'.join(self.comment1.getPhysicalPath()),
                                   '/'.join(self.comment3.getPhysicalPath())])        
        view = BulkActionsView(self.context, self.request)
        view()
        
        # Make sure that the two comments have been deleted
        self.assertEquals(self.conversation.total_comments, 1)
        comment = self.conversation.getComments().next()
        self.failUnless(comment)
        self.assertEquals(comment, self.comment2)

def test_suite():
    return unittest.defaultTestLoader.loadTestsFromName(__name__)