Line | |
---|
1 | """ |
---|
2 | Ported to Python 3. |
---|
3 | """ |
---|
4 | |
---|
5 | from twisted.trial import unittest |
---|
6 | from twisted.application import service |
---|
7 | from allmydata.stats import CPUUsageMonitor |
---|
8 | from allmydata.util import pollmixin |
---|
9 | import allmydata.test.common_util as testutil |
---|
10 | |
---|
11 | class FasterMonitor(CPUUsageMonitor): |
---|
12 | POLL_INTERVAL = 0.01 |
---|
13 | |
---|
14 | |
---|
15 | class CPUUsage(unittest.TestCase, pollmixin.PollMixin, testutil.StallMixin): |
---|
16 | def setUp(self): |
---|
17 | self.s = service.MultiService() |
---|
18 | self.s.startService() |
---|
19 | |
---|
20 | def tearDown(self): |
---|
21 | return self.s.stopService() |
---|
22 | |
---|
23 | def test_monitor(self): |
---|
24 | m = FasterMonitor() |
---|
25 | s = m.get_stats() # before it has been started |
---|
26 | self.failIf("cpu_monitor.1min_avg" in s) |
---|
27 | m.setServiceParent(self.s) |
---|
28 | def _poller(): |
---|
29 | return bool(len(m.samples) == m.HISTORY_LENGTH+1) |
---|
30 | d = self.poll(_poller) |
---|
31 | # pause a couple more intervals, to make sure that the history-trimming |
---|
32 | # code is exercised |
---|
33 | d.addCallback(self.stall, FasterMonitor.POLL_INTERVAL * 2) |
---|
34 | def _check(res): |
---|
35 | s = m.get_stats() |
---|
36 | self.failUnless("cpu_monitor.1min_avg" in s) |
---|
37 | self.failUnless("cpu_monitor.5min_avg" in s) |
---|
38 | self.failUnless("cpu_monitor.15min_avg" in s) |
---|
39 | self.failUnless("cpu_monitor.total" in s) |
---|
40 | d.addCallback(_check) |
---|
41 | return d |
---|
42 | |
---|
Note: See
TracBrowser
for help on using the repository browser.