[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)