[Zope-CVS] CVS: Products/Scheduler/tests - testScheduler.py:1.1

[Chris McDonough]

Update of /cvs-repository/Products/Scheduler/tests
In directory cvs.zope.org:/tmp/cvs-serv11728/tests

Added Files:
	testScheduler.py 
Log Message:
Changes which allow the descheduling of a task by its task id rather than by its time.  Now, when a task is scheduled, it is assigned a task id (available as the taskid attribute of the task).  The task must now be descheduled using its task id rather than its time.  A tag was made before this change named "before_taskid" in CVS in case changes to the HEAD cause breakages in dependent code.


=== Added File Products/Scheduler/tests/testScheduler.py ===
import sys, os, unittest, time
from unittest import TestCase, TestSuite, TextTestRunner, makeSuite

from OFS.Application import Application
from ZPublisher.HTTPRequest import HTTPRequest
from ZPublisher.HTTPResponse import HTTPResponse
from Products.PythonScripts.PythonScript import PythonScript
from Acquisition import aq_base

from Products.Scheduler.Scheduler import Scheduler
from Products.Scheduler.Task import Task
from Products.Scheduler import IScheduler
from Products.Scheduler import IScheduledEvent
from Products.Scheduler import IDescheduledEvent
from Products.Scheduler import ITimeEvent
from Products.Event.EventService import EventService

stuff = {} # used for reporting during tests

put_stuff = """from Products.Scheduler.tests.testScheduler import stuff
stuff.update({'stuff_put':1})"""
get_stuff = """from Products.Scheduler.tests.testScheduler import stuff
return stuff"""

class TestScheduler(TestCase):
    def setUp(self):
        self.app = Application()
        app = self.app
        app.id = 'App'
        app._setObject('portal_events', EventService())
        app._setObject('portal_scheduler', Scheduler('portal_scheduler'))
        app._setObject('put_stuff', PythonScript('put_stuff'))
        app.put_stuff.ZPythonScript_edit('', put_stuff)
        app._setObject('get_stuff', PythonScript('get_stuff'))
        app.get_stuff.ZPythonScript_edit('', get_stuff)
        resp = HTTPResponse()
        os.environ['SERVER_NAME']='server'
        os.environ['SERVER_PORT']='80'
        self.req = HTTPRequest(sys.stdin, os.environ, resp)
        app.REQUEST = self.req

    def tearDown(self):
        self.app.REQUEST = None
        del self.app
        stuff.clear()

    def testSchedulerIsSubscribedToEventService(self):
        app = self.app
        self.failUnless(app.portal_events.listSubscriptions(
            app.portal_scheduler, ITimeEvent))
        self.failUnless(app.portal_events.listSubscriptions(
            app.portal_scheduler, IScheduledEvent))
        self.failUnless(
            (aq_base(app.portal_events), ITimeEvent, None) in
            app.portal_scheduler.subscribed
            )
        self.failUnless(
            (aq_base(app.portal_events), IScheduledEvent, None) in
            app.portal_scheduler.subscribed
            )
    def testSchedulerIsUnsubscribedFromEventService(self):
        app = self.app
        ps = app.portal_scheduler
        app._delObject('portal_scheduler')
        self.failIf(app.portal_events.listSubscriptions(ps, ITimeEvent))
        self.failIf(app.portal_events.listSubscriptions(ps, IScheduledEvent))

    def testTaskIsScheduledDirectly(self):
        app = self.app
        now = time.time()
        t = Task('test', now, '/put_stuff')
        app.portal_scheduler.schedule(now, t)
        task = app.portal_scheduler.getPendingTasks()[0]
        self.assertEqual(task[0], int(now))
        self.assertEqual(aq_base(task[1]), t)

    def testTaskIsDescheduledDirectly(self):
        app = self.app
        now = time.time()
        task = Task('test', now, '/put_stuff')
        app.portal_scheduler.schedule(now, task)
        t, scheduled_task = app.portal_scheduler.getPendingTasks()[0]
        self.assertEqual(t, int(now))
        self.assertEqual(aq_base(scheduled_task), task)
        app.portal_scheduler.deschedule(task.taskid)
        self.assert_(not app.portal_scheduler.getPendingTasks())
        

from AccessControl import ModuleSecurityInfo
ModuleSecurityInfo('Products').declarePublic('Scheduler')
ModuleSecurityInfo('Products.Scheduler').declarePublic('tests')
ModuleSecurityInfo('Products.Scheduler.tests').declarePublic('testScheduler')
ModuleSecurityInfo('Products.Scheduler.tests.testScheduler').declarePublic('stuff')

ModuleSecurityInfo('Products.Groups.Group').declarePublic('stripGroupTag')

def test_suite():
    testsuite = makeSuite(TestScheduler, 'test')
    alltests = TestSuite((testsuite,))
    return alltests

if __name__ == '__main__':
    runner = TextTestRunner(verbosity=9)
    runner.run(test_suite())