[Zope-CVS] CVS: Products/Ape/lib/apelib/tests - testscanner.py:1.1.2.1 testall.py:1.4.2.1 testimpl.py:1.1.6.1 testio.py:1.2.4.1
Shane Hathaway
shane@zope.com
Wed, 23 Jul 2003 00:13:25 -0400
Update of /cvs-repository/Products/Ape/lib/apelib/tests
In directory cvs.zope.org:/tmp/cvs-serv21220/lib/apelib/tests
Modified Files:
Tag: ape-scan-branch
testall.py testimpl.py testio.py
Added Files:
Tag: ape-scan-branch
testscanner.py
Log Message:
Rough implementation of cache freshness scanning.
This will hopefully enable smoother filesystem storage.
=== Added File Products/Ape/lib/apelib/tests/testscanner.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.
#
##############################################################################
"""Cache scanner tests
$Id: testscanner.py,v 1.1.2.1 2003/07/23 04:12:49 shane Exp $
"""
import unittest
from time import time
from apelib.zodb3.scanner import ScanControl
class FakeRepository:
def freshen(self, d):
res = {}
for source in d.keys():
repo, location = source
if repo is not self:
raise AssertionError, "repo must be self"
if str(location) != location:
raise AssertionError, "location is expected to be a string"
# Always report a change
res[source] = time()
return res
class ScanControlTests(unittest.TestCase):
def setUp(self):
ctl = self.ctl = ScanControl()
self.conn1 = ctl.newConnection()
self.conn2 = ctl.newConnection()
def testSetNewOIDs(self):
self.conn1.setOIDs([5, 8])
oids = list(self.ctl.oids.keys())
self.assertEqual(oids, [5, 8])
self.assertEqual(list(self.ctl.conn_oids.keys()), [self.conn1.conn_id])
def testSetMultipleConnectionOIDs(self):
self.conn1.setOIDs([5, 8])
self.conn2.setOIDs([8, 9])
oids = list(self.ctl.oids.keys())
self.assertEqual(oids, [5,8,9])
conns = list(self.ctl.conn_oids.keys())
self.assertEqual(conns, [self.conn1.conn_id, self.conn2.conn_id])
def testRemoveOIDs(self):
self.conn1.setOIDs([5, 8])
self.conn2.setOIDs([8, 9])
self.conn1.setOIDs([8])
oids = list(self.ctl.oids.keys())
self.assertEqual(oids, [8,9])
conns = list(self.ctl.conn_oids.keys())
self.assertEqual(conns, [self.conn1.conn_id, self.conn2.conn_id])
self.conn1.setOIDs([])
oids = list(self.ctl.oids.keys())
self.assertEqual(oids, [8,9])
self.assertEqual(list(self.ctl.conn_oids.keys()), [self.conn2.conn_id])
class ScannerTests(unittest.TestCase):
def setUp(self):
ctl = self.ctl = ScanControl()
self.conn1 = ctl.newConnection()
self.conn2 = ctl.newConnection()
self.scanner = ctl.scanner
self.repo = FakeRepository()
def testAddSource(self):
source = (self.repo, '5')
self.scanner.setSources(5, [source])
self.assertEqual(len(self.scanner.future), 1)
self.assertEqual(self.scanner.future[5][0], [source])
def testChangeSources(self):
self.conn1.setOIDs([5])
old_sources = [(self.repo, '5')]
self.scanner.setSources(5, old_sources)
self.assertEqual(len(self.scanner.future), 0)
self.assertEqual(len(self.scanner.current), 1)
self.assertEqual(self.scanner.current[5], {old_sources[0]: None})
new_sources = [(self.repo, '6'), (self.repo, '7')]
self.scanner.setSources(5, new_sources)
self.assertEqual(len(self.scanner.future), 0)
self.assertEqual(len(self.scanner.current), 1)
self.assertEqual(self.scanner.current[5],
{new_sources[0]: None, new_sources[1]: None})
def testRemoveOID(self):
self.conn1.setOIDs([5])
self.assertEqual(len(self.scanner.current), 1)
self.conn1.setOIDs([])
self.assertEqual(len(self.scanner.current), 0)
def testNewScan(self):
self.conn1.setOIDs([5])
new_sources = [(self.repo, '6'), (self.repo, '7')]
self.scanner.setSources(5, new_sources)
to_invalidate = self.scanner.scan()
self.assertEqual(len(to_invalidate), 0)
def testUpdateScan(self):
self.conn1.setOIDs([5])
new_sources = [(self.repo, '6'), (self.repo, '7')]
self.scanner.setSources(5, new_sources)
to_invalidate = self.scanner.scan()
to_invalidate = self.scanner.scan()
self.assertEqual(len(to_invalidate), 1)
def testPruneFuture(self):
# Simulate some data.
self.scanner.future[5] = ([], time()) # Should not be pruned
self.scanner.future[900] = ([], time() - 100000) # Should be pruned
self.scanner.pruneFuture()
self.assertEqual(len(self.scanner.future), 1)
self.assert_(self.scanner.future.has_key(5))
if __name__ == '__main__':
unittest.main()
=== Products/Ape/lib/apelib/tests/testall.py 1.4 => 1.4.2.1 ===
--- Products/Ape/lib/apelib/tests/testall.py:1.4 Wed Jul 9 11:40:08 2003
+++ Products/Ape/lib/apelib/tests/testall.py Wed Jul 23 00:12:49 2003
@@ -38,6 +38,7 @@
from testparams import ParamsTests
from testsqlimpl import ApelibSQLImplTests
from apelib.config.tests.test_minitables import MiniTableTests
+from testscanner import ScanControlTests, ScannerTests
import testzope2sql
sql_suite = testzope2sql.test_suite()
@@ -54,6 +55,8 @@
Zope2FSUnderscoreTests,
ParamsTests,
ApelibSQLImplTests,
+ ScanControlTests,
+ ScannerTests,
):
suite.addTest(unittest.makeSuite(klass, 'test'))
suite.addTest(sql_suite)
=== Products/Ape/lib/apelib/tests/testimpl.py 1.1 => 1.1.6.1 ===
--- Products/Ape/lib/apelib/tests/testimpl.py:1.1 Wed Apr 9 23:09:57 2003
+++ Products/Ape/lib/apelib/tests/testimpl.py Wed Jul 23 00:12:49 2003
@@ -21,6 +21,7 @@
from types import ListType, TupleType
from Interface import Interface
+from Interface.IInterface import IInterface
from Interface.Verify import verifyClass
@@ -35,9 +36,11 @@
raise
def _testAllInModule(self, m):
+ name = m.__name__
for attr, value in m.__dict__.items():
if (hasattr(value, '__implements__') and
- not Interface.isImplementedBy(value)):
+ not IInterface.isImplementedBy(value)
+ and getattr(value, '__module__', None) == name):
self._testObjectImpl(value)
def _testAllInPackage(self, p):
=== Products/Ape/lib/apelib/tests/testio.py 1.2 => 1.2.4.1 ===
--- Products/Ape/lib/apelib/tests/testio.py:1.2 Mon May 26 22:19:13 2003
+++ Products/Ape/lib/apelib/tests/testio.py Wed Jul 23 00:12:49 2003
@@ -85,7 +85,7 @@
event, classified_state = obsys.serialize(keychain, ob)
gwsys.store(keychain, classified_state)
- cs, hash_value = gwsys.load(keychain)
+ event, cs, hash_value = gwsys.load(keychain)
ob2 = obsys.newObject(cs)
obsys.deserialize(keychain, ob2, cs)
self.assertEqual(ob.strdata, ob2.strdata)