[Zope-Checkins] CVS: Zope/lib/python/Lifetime/tests - __init__.py:1.1.2.1 testLifetime.py:1.1.2.1

[Chris McDonough]

Update of /cvs-repository/Zope/lib/python/Lifetime/tests
In directory cvs.zope.org:/tmp/cvs-serv31665/lib/python/Lifetime/tests

Added Files:
      Tag: chrism-scheduling-branch
	__init__.py testLifetime.py 
Log Message:
This branch includes low-level scheduling services for Zope.

See http://www.plope.com/Members/chrism/scheduling_service for more information.


=== Added File Zope/lib/python/Lifetime/tests/__init__.py ===
""" """


=== Added File Zope/lib/python/Lifetime/tests/testLifetime.py ===
##############################################################################
#
# Copyright (c) 2001, 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 os, sys, unittest, time, math
import Lifetime, ZServer
from Lifetime.timeslice import timeslice

class TestLifetime( unittest.TestCase ):
    """ Test the Lifetime module """
    def test_shutdown(self):
        old_code = ZServer.exit_code
        old_phase = Lifetime._shutdown_phase
        old_timeout = Lifetime._shutdown_timeout
        try:
            Lifetime.shutdown('foo')
            self.assertEqual(ZServer.exit_code, 'foo')
            self.assertEqual(Lifetime._shutdown_phase, 1)
            Lifetime.shutdown('bar', fast=1)
            self.assertEqual(Lifetime._shutdown_phase, 1)
            self.assertEqual(Lifetime._shutdown_timeout, 1.0)
        finally:
            ZServer.exit_code = old_code
            Lifetime._shutdown_phase = old_phase
            Lifetime._shutdown_timeout = old_timeout

    def test_maintain_load_indicator(self):
        old_slice = Lifetime._last_slice
        old_list = Lifetime._load_list[:]
        try:
            Lifetime._last_slice = 0
            Lifetime.maintain_load_indicator(2, 30)
            self.assertEqual(Lifetime._last_slice, 2)
            self.assertEqual(Lifetime._load_list, [2])
            Lifetime.maintain_load_indicator(4, 30)
            self.assertEqual(Lifetime._last_slice, 4)
            self.assertEqual(Lifetime._load_list, [2, 4])
            Lifetime.maintain_load_indicator(34, 30)
            self.assertEqual(Lifetime._last_slice, 34)
            self.assertEqual(Lifetime._load_list, [2, 4])
        finally:
            Lifetime._last_slice = old_slice
            Lifetime._load_list = old_list

    def test_load_avg(self):
        # XXX will these tests fly on all platforms or is fp math going to
        # make the assertions inaccurate?
        old_list = Lifetime._load_list[:]
        old_delay = Lifetime._load_avg_delay
        try:
            LL = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
            Lifetime._load_list = LL
            Lifetime._load_avg_delay = 1
            self.assertEqual(Lifetime.load_avg(2, 2), 1)
            self.assertEqual(Lifetime.load_avg(10, 10), 1)
            self.assertEqual(Lifetime.load_avg(10, 20), 0)
            self.assertEqual(Lifetime.load_avg(20, 20), 0.5)
            LL = [2, 4, 6, 8, 10, 12, 14, 16, 18, 20]
            Lifetime._load_list = LL
            Lifetime._load_avg_delay = 2
            self.assertEqual(Lifetime.load_avg(2, 2), 1)
            self.assertEqual(Lifetime.load_avg(10, 20), 1)
            self.assertEqual(Lifetime.load_avg(10, 30), 0)
            self.assertEqual(Lifetime.load_avg(20, 30), 0.5)
            LL = [2, 16, 30, 110, 160, 178, 180, 182, 184, 186]
            Lifetime._load_list = LL
            Lifetime._load_avg_delay = 2
            self.assertEqual(Lifetime.load_avg(2, 2), 1)
            self.assertEqual(Lifetime.load_avg(15, 30), .25)
            self.assertEqual(Lifetime.load_avg(160, 160), .0625)
            self.assertEqual(Lifetime.load_avg(10, 186), 1)
        finally:
            Lifetime._load_list = old_list
            Lifetime._load_avg_delay = old_delay
        

class TestCTimeslice(unittest.TestCase):
    def test_with_when(self):
        t = timeslice(2, 200)
        t2 = self._python_impl(2, 200)
        self.assertEqual(t, t2)
        t = timeslice(4, 1957)
        t2 = self._python_impl(4, 1957)
        self.assertEqual(t, t2)
        tt = int(time.time())
        t = timeslice(167, tt)
        t2 = self._python_impl(167, tt)
        self.assertEqual(t, t2)

    def test_without_when(self):
        tt = int(time.time())
        t = timeslice(2)
        t2 = self._python_impl(2, tt) # possible race condition, grr.
        self.assertEqual(t, t2)
        tt = int(time.time())
        t = timeslice(100)
        t2 = self._python_impl(100, tt) # possible race condition, grr.
        self.assertEqual(t, t2)

    def _python_impl(self, period, now=None):
        if now is None:
            now = time.time()
        low = int(math.floor(now)) - period + 1
        high = int(math.ceil(now)) + 1
        for x in range(low, high):
            if x % period == 0:
                return x

def test_suite():
    suite = unittest.TestSuite()
    suite.addTest( unittest.makeSuite( TestLifetime ) )
    suite.addTest( unittest.makeSuite( TestCTimeslice ) )
    return suite

def main():
    unittest.main(defaultTest='test_suite')

if __name__ == '__main__':
    main()