"""Test the plone.app.discussion catalog indexes """ from Products.CMFCore.utils import getToolByName from datetime import datetime from plone.app.discussion.interfaces import IConversation from plone.app.discussion.testing import PLONE_APP_DISCUSSION_INTEGRATION_TESTING # noqa from plone.app.testing import setRoles from plone.app.testing import TEST_USER_ID from zope.annotation.interfaces import IAnnotations from zope.component import createObject import transaction import unittest2 as unittest class CatalogSetupTest(unittest.TestCase): layer = PLONE_APP_DISCUSSION_INTEGRATION_TESTING def setUp(self): self.portal = self.layer['portal'] def test_catalog_installed(self): self.assertTrue( 'total_comments' in self.portal.portal_catalog.indexes() ) self.assertTrue( 'commentators' in self.portal.portal_catalog.indexes() ) self.assertTrue( 'total_comments' in self.portal.portal_catalog.schema() ) self.assertTrue( 'in_response_to' in self.portal.portal_catalog.schema() ) def test_collection_criteria_installed(self): if 'portal_atct' not in self.portal: return try: self.portal.portal_atct.getIndex('commentators') self.portal.portal_atct.getIndex('total_comments') self.portal.portal_atct.getMetadata('total_comments') except AttributeError: self.fail() class ConversationCatalogTest(unittest.TestCase): layer = PLONE_APP_DISCUSSION_INTEGRATION_TESTING def setUp(self): self.portal = self.layer['portal'] setRoles(self.portal, TEST_USER_ID, ['Manager']) self.catalog = getToolByName(self.portal, 'portal_catalog') conversation = IConversation(self.portal.doc1) comment1 = createObject('plone.Comment') comment1.title = 'Comment 1' comment1.text = 'Comment text' comment1.creator = 'jim' comment1.author_username = 'Jim' comment1.creation_date = datetime(2006, 9, 17, 14, 18, 12) comment1.modification_date = datetime(2006, 9, 17, 14, 18, 12) new_comment1_id = conversation.addComment(comment1) self.comment_id = new_comment1_id brains = self.catalog.searchResults(dict( path={ 'query': '/'.join(self.portal.doc1.getPhysicalPath()) }, portal_type="Document" )) self.conversation = conversation self.brains = brains self.doc1_brain = brains[0] self.comment1 = comment1 self.new_comment1_id = new_comment1_id def test_total_comments(self): self.assertTrue('total_comments' in self.doc1_brain) self.assertEqual(self.doc1_brain.total_comments, 1) comment2 = createObject('plone.Comment') comment2.title = 'Comment 2' comment2.text = 'Comment text' new_comment2_id = self.conversation.addComment(comment2) comment2 = self.portal.doc1.restrictedTraverse( '++conversation++default/%s' % new_comment2_id) comment2.reindexObject() brains = self.catalog.searchResults(dict( path={ 'query': '/'.join(self.portal.doc1.getPhysicalPath()) }, portal_type="Document" )) doc1_brain = brains[0] self.assertEqual(doc1_brain.total_comments, 2) def test_last_comment_date(self): self.assertTrue('last_comment_date' in self.doc1_brain) self.assertEqual( self.doc1_brain.last_comment_date, datetime(2006, 9, 17, 14, 18, 12) ) # Add another comment and check if last comment date is updated. comment2 = createObject('plone.Comment') comment2.title = 'Comment 2' comment2.text = 'Comment text' comment2.creation_date = datetime(2009, 9, 17, 14, 18, 12) comment2.modification_date = datetime(2009, 9, 17, 14, 18, 12) new_comment2_id = self.conversation.addComment(comment2) comment2 = self.portal.doc1.restrictedTraverse( '++conversation++default/%s' % new_comment2_id) comment2.reindexObject() brains = self.catalog.searchResults(dict( path={ 'query': '/'.join(self.portal.doc1.getPhysicalPath()) }, portal_type="Document" )) doc1_brain = brains[0] self.assertEqual( doc1_brain.last_comment_date, datetime(2009, 9, 17, 14, 18, 12) ) # Remove the comment again del self.conversation[new_comment2_id] brains = self.catalog.searchResults(dict( path={ 'query': '/'.join(self.portal.doc1.getPhysicalPath()) }, portal_type="Document" )) doc1_brain = brains[0] self.assertEqual( doc1_brain.last_comment_date, datetime(2006, 9, 17, 14, 18, 12) ) # remove all comments del self.conversation[self.new_comment1_id] brains = self.catalog.searchResults(dict( path={ 'query': '/'.join(self.portal.doc1.getPhysicalPath()) }, portal_type="Document" )) doc1_brain = brains[0] self.assertEqual(doc1_brain.last_comment_date, None) def test_commentators(self): self.assertTrue('commentators' in self.doc1_brain) self.assertEqual(self.doc1_brain.commentators, ('Jim',)) # add another comment with another author comment2 = createObject('plone.Comment') comment2.title = 'Comment 2' comment2.text = 'Comment text' comment2.creator = 'emma' comment2.author_username = 'Emma' new_comment2_id = self.conversation.addComment(comment2) comment2 = self.portal.doc1.restrictedTraverse( '++conversation++default/%s' % new_comment2_id) comment2.reindexObject() brains = self.catalog.searchResults(dict( path={ 'query': '/'.join(self.portal.doc1.getPhysicalPath()) }, portal_type="Document" )) doc1_brain = brains[0] self.assertEqual(doc1_brain.commentators, ('Jim', 'Emma')) # remove one comments del self.conversation[new_comment2_id] brains = self.catalog.searchResults(dict( path={ 'query': '/'.join(self.portal.doc1.getPhysicalPath()) }, portal_type="Document" )) doc1_brain = brains[0] self.assertEqual(doc1_brain.commentators, ('Jim',)) # remove all comments del self.conversation[self.new_comment1_id] brains = self.catalog.searchResults(dict( path={ 'query': '/'.join(self.portal.doc1.getPhysicalPath()) }, portal_type="Document" )) doc1_brain = brains[0] self.assertEqual(doc1_brain.commentators, ()) def test_conversation_indexes_not_in_comments(self): brains = self.catalog.searchResults(dict( path={ 'query': '/'.join(self.portal.doc1.getPhysicalPath()) }, portal_type="Discussion Item" )) comment1_brain = brains[0] self.assertEqual(comment1_brain.commentators, None) self.assertEqual(comment1_brain.last_comment_date, None) self.assertEqual(comment1_brain.total_comments, None) def test_dont_index_private_commentators(self): self.comment1.manage_permission("View", roles=tuple()) self.portal.doc1.reindexObject() brains = self.catalog.searchResults(dict( path={ 'query': '/'.join(self.portal.doc1.getPhysicalPath()) }, portal_type="Document" )) doc1_brain = brains[0] self.assertEqual(doc1_brain.commentators, ()) class CommentCatalogTest(unittest.TestCase): layer = PLONE_APP_DISCUSSION_INTEGRATION_TESTING def setUp(self): self.portal = self.layer['portal'] setRoles(self.portal, TEST_USER_ID, ['Manager']) self.catalog = getToolByName(self.portal, 'portal_catalog') conversation = IConversation(self.portal.doc1) self.conversation = conversation comment1 = createObject('plone.Comment') comment1.text = 'Comment text' comment1.creator = 'jim' comment1.author_name = 'Jim' new_comment1_id = conversation.addComment(comment1) self.comment_id = new_comment1_id # Comment brain self.comment = self.portal.doc1.restrictedTraverse( '++conversation++default/%s' % new_comment1_id) brains = self.catalog.searchResults(dict( path={ 'query': '/'.join(self.comment.getPhysicalPath()) } )) self.comment_brain = brains[0] def test_title(self): self.assertEqual(self.comment_brain.Title, 'Jim on Document 1') def test_no_name_title(self): comment = createObject('plone.Comment') comment.text = 'Comment text' cid = self.conversation.addComment(comment) # Comment brain comment = self.portal.doc1.restrictedTraverse( '++conversation++default/%s' % cid) brains = self.catalog.searchResults(dict( path={ 'query': '/'.join(comment.getPhysicalPath()) } )) comment_brain = brains[0] self.assertEqual(comment_brain.Title, "Anonymous on Document 1") def test_type(self): self.assertEqual(self.comment_brain.portal_type, 'Discussion Item') self.assertEqual(self.comment_brain.meta_type, 'Discussion Item') self.assertEqual(self.comment_brain.Type, 'Comment') def test_review_state(self): self.assertEqual(self.comment_brain.review_state, 'published') def test_creator(self): self.assertEqual(self.comment_brain.Creator, 'jim') def test_in_response_to(self): """Make sure in_response_to returns the title or id of the content object the comment was added to. """ self.assertEqual(self.comment_brain.in_response_to, 'Document 1') def test_add_comment(self): self.assertTrue(self.comment_brain) def test_delete_comment(self): # Make sure a comment is removed from the catalog as well when it is # deleted. del self.conversation[self.comment_id] brains = self.catalog.searchResults(dict( path={ 'query': '/'.join(self.comment.getPhysicalPath()) } )) self.assertEqual(len(brains), 0) def test_remove_comments_when_content_object_is_removed(self): """Make sure all comments are removed from the catalog, if the content object is removed. """ brains = self.catalog.searchResults({'portal_type': 'Discussion Item'}) self.assertEqual(len(brains), 1) self.portal.manage_delObjects(["doc1"]) brains = self.catalog.searchResults({'portal_type': 'Discussion Item'}) self.assertEqual(len(brains), 0) def test_move_comments_when_content_object_is_moved(self): # Create two folders and a content object with a comment self.portal.invokeFactory( id='folder1', title='Folder 1', type_name='Folder' ) self.portal.invokeFactory( id='folder2', title='Folder 2', type_name='Folder' ) self.portal.folder1.invokeFactory( id='moveme', title='Move Me', type_name='Document' ) conversation = IConversation(self.portal.folder1.moveme) comment = createObject('plone.Comment') comment_id = conversation.addComment(comment) # We need to commit here so that _p_jar isn't None and move will work transaction.savepoint(optimistic=True) # Move moveme from folder1 to folder2 cp = self.portal.folder1.manage_cutObjects(ids=('moveme',)) self.portal.folder2.manage_pasteObjects(cp) # Make sure no old comment brains are brains = self.catalog.searchResults(dict( portal_type="Discussion Item", path={'query': '/'.join(self.portal.folder1.getPhysicalPath())} )) self.assertEqual(len(brains), 0) brains = self.catalog.searchResults(dict( portal_type="Discussion Item", path={ 'query': '/'.join(self.portal.folder2.getPhysicalPath()) } )) self.assertEqual(len(brains), 1) self.assertEqual( brains[0].getPath(), '/plone/folder2/moveme/++conversation++default/' + str(comment_id) ) def test_move_upper_level_folder(self): # create a folder with a nested structure self.portal.invokeFactory(id='sourcefolder', title='Source Folder', type_name='Folder') self.portal.sourcefolder.invokeFactory(id='moveme', title='Move Me', type_name='Folder') self.portal.sourcefolder.moveme.invokeFactory(id='mydocument', title='My Document', type_name='Folder') self.portal.invokeFactory(id='targetfolder', title='Target Folder', type_name='Folder') # create comment on my-document conversation = IConversation( self.portal.sourcefolder.moveme.mydocument ) comment = createObject('plone.Comment') comment_id = conversation.addComment(comment) # We need to commit here so that _p_jar isn't None and move will work transaction.savepoint(optimistic=True) # Move moveme from folder1 to folder2 cp = self.portal.sourcefolder.manage_cutObjects(ids=('moveme',)) self.portal.targetfolder.manage_pasteObjects(cp) # Make sure no old comment brains are left brains = self.catalog.searchResults(dict( portal_type="Discussion Item", path={'query': '/plone/sourcefolder/moveme'} )) self.assertEqual(len(brains), 0) # make sure comments are correctly index on the target brains = self.catalog.searchResults(dict( portal_type="Discussion Item", path={'query': '/plone/targetfolder/moveme'} )) self.assertEqual(len(brains), 1) self.assertEqual( brains[0].getPath(), '/plone/targetfolder/moveme/mydocument/++conversation++default/' + str(comment_id) ) def test_update_comments_when_content_object_is_renamed(self): # We need to commit here so that _p_jar isn't None and move will work transaction.savepoint(optimistic=True) self.portal.manage_renameObject("doc1", "doc2") brains = self.catalog.searchResults( portal_type='Discussion Item') self.assertEqual(len(brains), 1) self.assertEqual( brains[0].getPath(), '/plone/doc2/++conversation++default/' + str(self.comment_id) ) def test_clear_and_rebuild_catalog(self): brains = self.catalog.searchResults({'portal_type': 'Discussion Item'}) self.assertTrue(brains) # Clear and rebuild catalog self.catalog.clearFindAndRebuild() # Check if comment is still there brains = self.catalog.searchResults({'portal_type': 'Discussion Item'}) self.assertTrue(brains) comment_brain = brains[0] self.assertEqual(comment_brain.Title, u'Jim on Document 1') self.assertEqual( comment_brain.getPath(), '/plone/doc1/++conversation++default/' + str(self.comment_id) ) def test_clear_and_rebuild_catalog_for_nested_comments(self): # Create a nested comment structure: # # Conversation # +- Comment 1 # +- Comment 1_1 # | +- Comment 1_1_1 # +- Comment 1_2 # +- Comment 2 # +- Comment 2_1 comment1_1 = createObject('plone.Comment') comment1_1.title = 'Re: Comment 1' comment1_1.text = 'Comment text' comment1_1_1 = createObject('plone.Comment') comment1_1_1.title = 'Re: Re: Comment 1' comment1_1_1.text = 'Comment text' comment1_2 = createObject('plone.Comment') comment1_2.title = 'Re: Comment 1 (2)' comment1_2.text = 'Comment text' comment2 = createObject('plone.Comment') comment2.title = 'Comment 2' comment2.text = 'Comment text' comment2_1 = createObject('plone.Comment') comment2_1.title = 'Re: Comment 2' comment2_1.text = 'Comment text' # Create the nested comment structure new_id_1 = self.conversation.addComment(self.comment) new_id_2 = self.conversation.addComment(comment2) comment1_1.in_reply_to = self.comment_id new_id_1_1 = self.conversation.addComment(comment1_1) comment1_1_1.in_reply_to = new_id_1_1 self.conversation.addComment(comment1_1_1) comment1_2.in_reply_to = new_id_1 self.conversation.addComment(comment1_2) comment2_1.in_reply_to = new_id_2 self.conversation.addComment(comment2_1) # Clear and rebuild catalog self.catalog.clearFindAndRebuild() # Check if comments are still there brains = self.catalog.searchResults({'portal_type': 'Discussion Item'}) self.assertTrue(brains) self.assertEqual(len(brains), 6) class NoConversationCatalogTest(unittest.TestCase): layer = PLONE_APP_DISCUSSION_INTEGRATION_TESTING def setUp(self): self.portal = self.layer['portal'] setRoles(self.portal, TEST_USER_ID, ['Manager']) self.catalog = getToolByName(self.portal, 'portal_catalog') conversation = IConversation(self.portal.doc1) brains = self.catalog.searchResults(dict( path={ 'query': '/'.join(self.portal.doc1.getPhysicalPath()) }, portal_type="Document" )) self.conversation = conversation self.brains = brains self.doc1_brain = brains[0] def test_total_comments(self): self.assertTrue('total_comments' in self.doc1_brain) self.assertEqual(self.doc1_brain.total_comments, 0) # Make sure no conversation has been created self.assertTrue( 'plone.app.discussion:conversation' not in IAnnotations(self.portal.doc1) )