[Zodb-checkins] CVS: Zope/lib/python/ZODB/tests - testActivityMonitor.py:1.2

[Shane Hathaway]

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

Added Files:
	testActivityMonitor.py 
Log Message:
Merged shane-activity-monitoring-branch.


=== Zope/lib/python/ZODB/tests/testActivityMonitor.py 1.1 => 1.2 ===
+#
+# 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.
+#
+##############################################################################
+"""Tests of the default activity monitor.
+
+See ZODB/ActivityMonitor.py
+
+$Id$
+"""
+
+import unittest
+import time
+
+from ZODB.ActivityMonitor import ActivityMonitor
+
+
+class FakeConnection:
+
+    loads = 0
+    stores = 0
+
+    def _transferred(self, loads, stores):
+        self.loads = self.loads + loads
+        self.stores = self.stores + stores
+
+    def getTransferCounts(self, clear=0):
+        res = self.loads, self.stores
+        if clear:
+            self.loads = self.stores = 0
+        return res
+
+
+class Tests(unittest.TestCase):
+
+    def testAddLogEntries(self):
+        am = ActivityMonitor(history_length=3600)
+        self.assertEqual(len(am.log), 0)
+        c = FakeConnection()
+        c._transferred(1, 2)
+        am.closedConnection(c)
+        c._transferred(3, 7)
+        am.closedConnection(c)
+        self.assertEqual(len(am.log), 2)
+
+    def testTrim(self):
+        am = ActivityMonitor(history_length=0.1)
+        c = FakeConnection()
+        c._transferred(1, 2)
+        am.closedConnection(c)
+        time.sleep(0.2)
+        c._transferred(3, 7)
+        am.closedConnection(c)
+        self.assert_(len(am.log) <= 1)
+
+    def testSetHistoryLength(self):
+        am = ActivityMonitor(history_length=3600)
+        c = FakeConnection()
+        c._transferred(1, 2)
+        am.closedConnection(c)
+        time.sleep(0.2)
+        c._transferred(3, 7)
+        am.closedConnection(c)
+        self.assertEqual(len(am.log), 2)
+        am.setHistoryLength(0.1)
+        self.assertEqual(am.getHistoryLength(), 0.1)
+        self.assert_(len(am.log) <= 1)
+        
+    def testActivityAnalysis(self):
+        am = ActivityMonitor(history_length=3600)
+        c = FakeConnection()
+        c._transferred(1, 2)
+        am.closedConnection(c)
+        c._transferred(3, 7)
+        am.closedConnection(c)
+        res = am.getActivityAnalysis(start=0, end=0, divisions=10)
+        lastend = 0
+        for n in range(9):
+            div = res[n]
+            self.assertEqual(div['stores'], 0)
+            self.assertEqual(div['loads'], 0)
+            self.assert_(div['start'] > 0)
+            self.assert_(div['start'] >= lastend)
+            self.assert_(div['start'] < div['end'])
+            lastend = div['end']
+        div = res[9]
+        self.assertEqual(div['stores'], 9)
+        self.assertEqual(div['loads'], 4)
+        self.assert_(div['start'] > 0)
+        self.assert_(div['start'] >= lastend)
+        self.assert_(div['start'] < div['end'])
+        
+
+def test_suite():
+    return unittest.makeSuite(Tests)
+
+if __name__=='__main__':
+    unittest.main(defaultTest='test_suite')