[Zope3-checkins] CVS: Zope3/lib/python/Zope/App/Tasks/tests - testSchedule.py:1.1.2.1

[Ulrich Eck]

Update of /cvs-repository/Zope3/lib/python/Zope/App/Tasks/tests
In directory cvs.zope.org:/tmp/cvs-serv530/tests

Added Files:
      Tag: jack-e_scheduler_branch
	testSchedule.py 
Log Message:
Zope.App.Tasks:

Prototype Implementation of Jim's Schedule Design


=== Added File Zope3/lib/python/Zope/App/Tasks/tests/testSchedule.py ===
##############################################################################
#
# Copyright (c) 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL).  A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################

import unittest

from Interface import Interface
from Interface.Verify import verifyClass

from Zope.App.OFS.Services.ServiceManager.tests.PlacefulSetup import PlacefulSetup
from Zope.ComponentArchitecture import getServiceManager, getService
from Zope.App.Traversing import traverse
from Zope.App.OFS.Services.ServiceManager.ServiceManager import ServiceManager
from Zope.App.Security.IAuthenticationService import IAuthenticationService
from Zope.App.Security.Registries.PrincipalRegistry import principalRegistry


from Zope.App.Tasks.ITask import ITask
from Zope.App.Tasks.ISchedule import ISchedule
from Zope.App.Tasks.Task import Task
from Zope.App.Tasks.Schedule import Schedule

from datetime import datetime

test_principalId = "DummyPrincipal"

dt1 = datetime(2002,01,01)
dt2 = datetime(2002,12,31)

class TestTask(Task):

    processed = False

    def __call__(self):
        self.processed = True


class ChainingTask(TestTask):

    def __call__(self):
        self.processed = True

    def schedule(self):
        return TestTask(dt2, test_principalId)


class BuggyTask(TestTask):

    def __call__(self):
        raise Exception, "I'm a buggy task"

class BuggyTaskWithHandler(BuggyTask):

    def __call__(self):
        try:
            raise
        except:
            return TestTask(dt2, test_principalId)


class ScheduleTests(PlacefulSetup, unittest.TestCase):

    def setUp(self):
        PlacefulSetup.setUp(self)
        self.buildFolders()

        root_sm = getServiceManager(None)
        root_sm.defineService('Authentication', IAuthenticationService)
        root_sm.provideService('Authentication', principalRegistry)

        # create a service manager in rootFolder (default)
        self.createServiceManager()
        self.sm = sm = traverse(self.rootFolder, '++etc++Services')
        self.default = default = traverse(sm, 'Packages/default')

        auth_svc = getService(self.rootFolder, "Authentication")
        auth_svc.definePrincipal(test_principalId, "Test Principal")

        schedule = Schedule()
        default.setObject('test_schedule', schedule)

        self.schedule = schedule
        
    def createServiceManager(self):
        self.rootFolder.setServiceManager(ServiceManager())


    def testInterface(self):
        verifyClass(ISchedule, Schedule)
        verifyClass(ITask, Task)

    def testSchedule(self):
        task1 = TestTask(dt1, test_principalId)
        self.schedule.schedule(task1)
        self.assertEqual((task1,), self.schedule.tasks[dt1])

        task2 = TestTask(dt2, test_principalId)
        self.schedule.schedule(task2)
        self.assertEqual((task2,) ,self.schedule.tasks[dt2])

        task3 = TestTask(dt1, test_principalId)
        self.schedule.schedule(task3)
        self.assertEqual((task1,task3,), self.schedule.tasks[dt1])
        

    def testUnschedule(self):
        task1 = TestTask(dt1, test_principalId)
        self.schedule.schedule(task1)
        self.assertEqual((task1,), self.schedule.tasks[dt1])

        self.schedule.unschedule(task1)
        self.assertEqual(0, len(self.schedule.tasks))


    def testProcessSimple(self):
        task1 = TestTask(dt1, test_principalId)
        self.schedule.schedule(task1)

        task2 = TestTask(dt2, test_principalId)
        self.schedule.schedule(task2)

        task3 = TestTask(dt1, test_principalId)
        self.schedule.schedule(task3)
        
        self.schedule.process(dt1)
        self.assertEqual(task1.processed, True)
        self.assertEqual(task2.processed, False)
        self.assertEqual(task3.processed, True)

        self.assertEqual(self.schedule.tasks.get(dt1, None), None)

        self.schedule.process(dt2)
        self.assertEqual(task2.processed, True)

        self.assertEqual(self.schedule.tasks.get(dt2, None), None)


    def testProcessChained(self):
        task1 = ChainingTask(dt1, test_principalId)
        self.schedule.schedule(task1)
        
        self.schedule.process(dt1)
        self.assertEqual(task1.processed, True)

        self.assertEqual(len(self.schedule.tasks[dt2]), 1)

        task2 = self.schedule.tasks[dt2][0]
        self.assertEqual(task2.processed, False)

        self.schedule.process(dt2)
        self.assertEqual(task2.processed, True)

    def testProcessBuggy(self):
        task1 = BuggyTask(dt1, test_principalId)
        self.schedule.schedule(task1)
        
        self.schedule.process(dt1)
        self.assertEqual(task1.processed, False)

        self.assertEqual(self.schedule.tasks.get(dt1, None), None)

    def testProcessBuggyWithHandler(self):
        task1 = BuggyTaskWithHandler(dt1, test_principalId)
        self.schedule.schedule(task1)
        
        self.schedule.process(dt1)
        self.assertEqual(task1.processed, False)

        self.assertEqual(len(self.schedule.tasks[dt2]), 1)

        task2 = self.schedule.tasks[dt2][0]

        self.schedule.process(dt2)
        self.assertEqual(task2.processed, True)

        


def test_suite():
    suite = unittest.TestSuite()
    suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(
        ScheduleTests))
    return suite

if __name__ == '__main__':
    unittest.main()