[Zope-Checkins] CVS: Zope/lib/python/Products/Transience/tests - fauxtime.py:1.1 testTimeoutRelated.py:1.5 testTransientObject.py:1.3 testTransientObjectContainer.py:1.5
Chris McDonough
chrism@zope.com
Wed, 21 Nov 2001 17:46:37 -0500
Update of /cvs-repository/Zope/lib/python/Products/Transience/tests
In directory cvs.zope.org:/tmp/cvs-serv11583/tests
Modified Files:
testTimeoutRelated.py testTransientObject.py
testTransientObjectContainer.py
Added Files:
fauxtime.py
Log Message:
Moved TransientObjects into their own module.
Removed wrap_with argument from new and new_or_existing methods
of Transient Data Containers.
Removed delete method of Transient Data Containers.
Added out-of-memory protection to Transient Data Containers. A
new __init__ value ('limit') is used to specify the max number
of objects that can be contained within a transient data container.
A new envvar ZSESSION_OBJECT_LIMIT can be used to control the
limit of the default session_data TDC. Also updated help and
API docs with this change.
Added a new exception, MaxTransientObjectsExceeded, which is raised
when the OOM protection kicks in.
Various implementation changes including the use of a BTrees Length
object to store Transient Data Container length info as well
as improvements to how buckets are expired.
Addition of tests for OOM protection fatures.
=== Added File Zope/lib/python/Products/Transience/tests/fauxtime.py ===
import time as origtime
epoch = origtime.time()
def time():
""" False timer -- returns time 10 x faster than normal time """
return (origtime.time() - epoch) * 10.0
def sleep(duration):
""" False sleep -- sleep for 1/10 the time specifed """
origtime.sleep(duration / 10.0)
=== Zope/lib/python/Products/Transience/tests/testTimeoutRelated.py 1.4 => 1.5 ===
from Products.Transience.Transience import TransientObjectContainer
import Products.Transience.Transience
+import Products.Transience.TransientObject
from Products.PythonScripts.PythonScript import PythonScript
from ZODB.POSException import InvalidObjectReference
from DateTime import DateTime
from unittest import TestCase, TestSuite, TextTestRunner, makeSuite
from ZODB.DemoStorage import DemoStorage
from OFS.Application import Application
-import time, threading, whrandom
+import threading, whrandom
+import fauxtime
+import time as oldtime
+
WRITEGRANULARITY = 30
-epoch = time.time()
stuff = {}
def _getApp():
@@ -54,6 +57,7 @@
class TestBase(TestCase):
def setUp(self):
Products.Transience.Transience.time = fauxtime
+ Products.Transience.TransientObject.time = fauxtime
self.app = makerequest.makerequest(_getApp())
timeout = self.timeout = 1
sm=TransientObjectContainer(
@@ -66,12 +70,14 @@
get_transaction().abort()
_delApp()
del self.app
-
+ Products.Transience.Transience.time = oldtime
+ Products.Transience.TransientObject.time = oldtime
+
class TestLastAccessed(TestBase):
def testLastAccessed(self):
sdo = self.app.sm.new_or_existing('TempObject')
la1 = sdo.getLastAccessed()
- fauxsleep(WRITEGRANULARITY + 1)
+ fauxtime.sleep(WRITEGRANULARITY + 1)
sdo = self.app.sm['TempObject']
assert sdo.getLastAccessed() > la1, (sdo.getLastAccessed(), la1)
@@ -79,7 +85,7 @@
def testAddNotification(self):
self.app.sm.setAddNotificationTarget(addNotificationTarget)
sdo = self.app.sm.new_or_existing('TempObject')
- now = fauxtime()
+ now = fauxtime.time()
k = sdo.get('starttime')
assert type(k) == type(now)
assert k <= now
@@ -88,35 +94,26 @@
self.app.sm.setDelNotificationTarget(delNotificationTarget)
sdo = self.app.sm.new_or_existing('TempObject')
timeout = self.timeout * 60
- fauxsleep(timeout + (timeout * .33))
+ fauxtime.sleep(timeout + (timeout * .33))
try: sdo1 = self.app.sm['TempObject']
except KeyError: pass
- now = fauxtime()
+ now = fauxtime.time()
k = sdo.get('endtime')
assert type(k) == type(now)
assert k <= now
def addNotificationTarget(item, context):
- item['starttime'] = fauxtime()
+ item['starttime'] = fauxtime.time()
def delNotificationTarget(item, context):
- item['endtime'] = fauxtime()
-
-def fauxtime():
- """ False timer -- returns time 10 x faster than normal time """
- return (time.time() - epoch) * 10.0
-
-def fauxsleep(duration):
- """ False sleep -- sleep for 1/10 the time specifed """
- time.sleep(duration / 10.0)
+ item['endtime'] = fauxtime.time()
def test_suite():
last_accessed = makeSuite(TestLastAccessed, 'test')
start_end = makeSuite(TestNotifications, 'test')
- runner = TextTestRunner()
suite = TestSuite((start_end, last_accessed))
return suite
if __name__ == '__main__':
- runner = TextTestRunner(sys.stdout)
+ runner = TextTestRunner(verbosity=9)
runner.run(test_suite())
=== Zope/lib/python/Products/Transience/tests/testTransientObject.py 1.2 => 1.3 ===
#
##############################################################################
-import sys, os, time, whrandom, unittest
+import sys, os, whrandom, unittest
if __name__ == "__main__":
sys.path.insert(0, '../../..')
- #os.chdir('../../..')
import ZODB
-from Products.Transience.Transience import \
- TransientObjectContainer, TransientObject
+from Products.Transience.Transience import TransientObjectContainer
+from Products.Transience.TransientObject import TransientObject
+import Products.Transience.TransientObject
import Products.Transience.Transience
from unittest import TestCase, TestSuite, TextTestRunner, makeSuite
-
-epoch = time.time()
+import time as oldtime
+import fauxtime
class TestTransientObject(TestCase):
def setUp(self):
+ Products.Transience.Transience.time = fauxtime
+ Products.Transience.TransientObject.time = fauxtime
self.errmargin = .20
self.timeout = 60
- Products.Transience.Transience.time = fauxtime
self.t = TransientObjectContainer('sdc', timeout_mins=self.timeout/60)
def tearDown(self):
+ Products.Transience.Transience.time = oldtime
+ Products.Transience.TransientObject.time = oldtime
self.t = None
del self.t
def test_id(self):
t = self.t.new('xyzzy')
assert t.getId() != 'xyzzy'
+ assert t.getContainerKey() == 'xyzzy'
def test_validate(self):
t = self.t.new('xyzzy')
@@ -119,22 +123,22 @@
def test_getLastAccessed(self):
t = self.t.new('xyzzy')
- ft = fauxtime()
+ ft = fauxtime.time()
assert t.getLastAccessed() <= ft
def test_getCreated(self):
t = self.t.new('xyzzy')
- ft = fauxtime()
+ ft = fauxtime.time()
assert t.getCreated() <= ft
def test_setLastAccessed(self):
t = self.t.new('xyzzy')
- ft = fauxtime()
+ ft = fauxtime.time()
assert t.getLastAccessed() <= ft
- fauxsleep(self.timeout) # go to sleep past the granuarity
- ft2 = fauxtime()
+ fauxtime.sleep(self.timeout) # go to sleep past the granuarity
+ ft2 = fauxtime.time()
t.setLastAccessed()
- ft3 = fauxtime()
+ ft3 = fauxtime.time()
assert t.getLastAccessed() <= ft3
assert t.getLastAccessed() >= ft2
@@ -175,19 +179,11 @@
alltests = TestSuite((testsuite,))
return alltests
-def fauxtime():
- """ False timer -- returns time 10 x faster than normal time """
- return (time.time() - epoch) * 10.0
-
-def fauxsleep(duration):
- """ False sleep -- sleep for 1/10 the time specifed """
- time.sleep(duration / 10.0)
-
data = {
'a': 'a',
1: 1,
'Mary': 'no little lamb for you today!',
- 'epoch': epoch,
+ 'epoch': 999999999,
'fauxtime': fauxtime
}
=== Zope/lib/python/Products/Transience/tests/testTransientObjectContainer.py 1.4 => 1.5 ===
if __name__ == "__main__":
sys.path.insert(0, '../../..')
- #os.chdir('../../..')
import ZODB
-from Products.Transience.Transience import \
- TransientObjectContainer, TransientObject
+from Products.Transience.Transience import TransientObjectContainer,\
+ MaxTransientObjectsExceeded
+from Products.Transience.TransientObject import TransientObject
import Products.Transience.Transience
+import Products.Transience.TransientObject
from ExtensionClass import Base
from unittest import TestCase, TestSuite, TextTestRunner, makeSuite
-
-epoch = time.time()
-stash = {}
+import time as oldtime
+import fauxtime
class TestTransientObjectContainer(TestCase):
def setUp(self):
+ Products.Transience.Transience.time = fauxtime
+ Products.Transience.TransientObject.time = fauxtime
self.errmargin = .20
self.timeout = 60
- Products.Transience.Transience.time = fauxtime
self.t = TransientObjectContainer('sdc', timeout_mins=self.timeout/60)
def tearDown(self):
self.t = None
- del self.t
-
+ Products.Transience.Transience.time = oldtime
+ Products.Transience.TransientObject.time = oldtime
+
def testGetItemFails(self):
self.assertRaises(KeyError, self._getitemfail)
@@ -357,7 +359,7 @@
for x in range(10, 110):
self.t[x] = x
# these items will time out while we sleep
- fauxsleep(self.timeout * (self.errmargin+1))
+ fauxtime.sleep(self.timeout * (self.errmargin+1))
for x in range(110, 210):
self.t[x] = x
assert len(self.t.keys()) == 100, len(self.t.keys())
@@ -374,7 +376,7 @@
# 1 minute
for x in range(10, 110):
self.t[x] = x
- fauxsleep(self.timeout * (self.errmargin+1))
+ fauxtime.sleep(self.timeout * (self.errmargin+1))
assert len(self.t.keys()) == 0, len(self.t.keys())
# 2 minutes
@@ -382,9 +384,9 @@
self.t._reset()
for x in range(10, 110):
self.t[x] = x
- fauxsleep(self.timeout * (self.errmargin+1))
+ fauxtime.sleep(self.timeout * (self.errmargin+1))
assert len(self.t.keys()) == 100, len(self.t.keys())
- fauxsleep(self.timeout * (self.errmargin+1))
+ fauxtime.sleep(self.timeout * (self.errmargin+1))
assert len(self.t.keys()) == 0, len(self.t.keys())
# 3 minutes
@@ -392,22 +394,22 @@
self.t._reset()
for x in range(10, 110):
self.t[x] = x
- fauxsleep(self.timeout * (self.errmargin+1))
+ fauxtime.sleep(self.timeout * (self.errmargin+1))
assert len(self.t.keys()) == 100, len(self.t.keys())
- fauxsleep(self.timeout * (self.errmargin+1))
+ fauxtime.sleep(self.timeout * (self.errmargin+1))
assert len(self.t.keys()) == 100, len(self.t.keys())
- fauxsleep(self.timeout * (self.errmargin+1))
+ fauxtime.sleep(self.timeout * (self.errmargin+1))
assert len(self.t.keys()) == 0, len(self.t.keys())
def testGetItemDelaysTimeout(self):
for x in range(10, 110):
self.t[x] = x
# current bucket will become old after we sleep for a while.
- fauxsleep(self.timeout/2)
+ fauxtime.sleep(self.timeout/2)
# these items will be added to the new current bucket by getitem
for x in range(10, 110):
self.t[x]
- fauxsleep(self.timeout/2)
+ fauxtime.sleep(self.timeout/2)
assert len(self.t.keys()) == 100, len(self.t.keys())
for x in range(10, 110):
assert self.t[x] == x
@@ -416,11 +418,11 @@
for x in range(10, 110):
self.t[x] = x
# current bucket will become old after we sleep for a while.
- fauxsleep(self.timeout/2)
+ fauxtime.sleep(self.timeout/2)
# these items will be added to the new current bucket by getitem
for x in range(10, 110):
self.t[x] = x + 1
- fauxsleep(self.timeout/2)
+ fauxtime.sleep(self.timeout/2)
assert len(self.t.keys()) == 100, len(self.t.keys())
for x in range(10, 110):
assert self.t[x] == x + 1
@@ -429,11 +431,11 @@
for x in range(10, 110):
self.t[x] = x
# current bucket will become old after we sleep for a while.
- fauxsleep(self.timeout/2)
+ fauxtime.sleep(self.timeout/2)
# these items will be added to the new current bucket by getitem
for x in range(10, 110):
self.t.get(x)
- fauxsleep(self.timeout/2)
+ fauxtime.sleep(self.timeout/2)
assert len(self.t.keys()) == 100, len(self.t.keys())
for x in range(10, 110):
assert self.t[x] == x
@@ -479,9 +481,18 @@
def test_getId(self):
assert self.t.getId() == 'sdc'
- def test_getContainerKey(self):
- t = self.t.new('foobieblech')
- assert t.getContainerKey() == 'foobieblech'
+ def testSubobjectLimitWorks(self):
+ self.t = TransientObjectContainer('a', timeout_mins=self.timeout/60,
+ limit=10)
+ self.assertRaises(MaxTransientObjectsExceeded, self._maxOut)
+
+ def testUnlimitedSubobjectLimitWorks(self):
+ self._maxOut()
+
+ def _maxOut(self):
+ for x in range(11):
+ self.t.new(str(x))
+
def lsubtract(l1, l2):
l1=list(l1)
@@ -491,18 +502,9 @@
return l
def test_suite():
- #print "TransientObjectContainer tests take just about forever (10+ mins)"
testsuite = makeSuite(TestTransientObjectContainer, 'test')
alltests = TestSuite((testsuite,))
return alltests
-
-def fauxtime():
- """ False timer -- returns time 10 x faster than normal time """
- return (time.time() - epoch) * 10.0
-
-def fauxsleep(duration):
- """ False sleep -- sleep for 1/10 the time specifed """
- time.sleep(duration / 10.0)
if __name__ == '__main__':
runner = TextTestRunner(verbosity=9)