ftw.publisher.sender.persistence.Queue

All Samples(6) | Call(4) | Derive(0) | Import(2)

src/f/t/ftw.publisher.sender-2.2.0/ftw/publisher/sender/browser/publishformgen.py ftw.publisher.sender(Download)

from Products.Five import BrowserView
from Products.statusmessages.interfaces import IStatusMessage
from ftw.publisher.sender import getLogger
from ftw.publisher.sender import message_factory as _
from ftw.publisher.sender.persistence import Queue

        username = user.getUserName()
        # create Job
        queue = Queue(self.context)
        queue.createJob('push', self.context, username)
        self.logger.info('Created "%s" Job for "%s" at %s' % (

src/f/t/ftw.publisher.sender-2.2.0/ftw/publisher/sender/tests/test_views.py ftw.publisher.sender(Download)

from Products.PloneTestCase import ptc
from Products.statusmessages.interfaces import IStatusMessage
from ftw.publisher.sender.browser.views import PublishObject, DeleteObject
from ftw.publisher.sender.persistence import Queue
from ftw.publisher.sender.tests.layer import Layer

    def test_publish_object(self):
        queue = Queue(self.testdoc2)
        self.assertEquals(queue.countJobs(),0)
        PublishObject(self.testdoc2,self.testdoc2.REQUEST)()
        self.assertEquals(queue.countJobs(),1)

    def test_delete_object(self):
        queue = Queue(self.testdoc2)
        self.assertEquals(queue.countJobs(),0)
        DeleteObject(self.testdoc2,self.testdoc2.REQUEST)()
        self.assertEquals(queue.countJobs(),1)

    def test_delete_with_rollback(self):
        # This test simulates Products.CMFPlone.utils.isLinked which
        # deletes and rolls back for integrity checking.
        # When it is rolled back it should create a status message.
        queue = Queue(self.testdoc2)