/usr/lib/python2.7/dist-packages/ZODB/tests/testActivityMonitor.py is in python-zodb 1:3.10.7-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | ##############################################################################
#
# Copyright (c) 2001, 2002 Zope Foundation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (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')
|