← Back to branch summary

~ahasenack/landscape-client/landscape-client-1.5.5-0ubuntu0.9.04.0

« back to all changes in this revision

Viewing changes to landscape/lib/tests/test_monitor.py

  • Committer: Bazaar Package Importer
  • Author(s): Rick Clark
  • Date: 2008-09-08 16:35:57 UTC
  • mto: This revision was merged to the branch mainline in revision 2.
  • Revision ID: james.westby@ubuntu.com-20080908163557-fl0d2oc35hur473w
Tags: upstream-1.0.18
Import upstream version 1.0.18

Show diffs side-by-side

added added

removed removed

Lines of Context:
landscape/lib/tests/test_monitor.py
 
1
from landscape.lib.monitor import (
 
2
    Timer, Monitor, BurstMonitor, CoverageMonitor, FrequencyMonitor)
 
3
from landscape.reactor import FakeReactor
 
4
from landscape.tests.helpers import LandscapeTest
 
5
 
 
6
 
 
7
class ReactorHavingTest(LandscapeTest):
 
8
    def setUp(self):
 
9
        super(ReactorHavingTest, self).setUp()
 
10
        self.reactor = FakeReactor()
 
11
 
 
12
 
 
13
class TimerTest(ReactorHavingTest):
 
14
 
 
15
    def setUp(self):
 
16
        super(TimerTest, self).setUp()
 
17
        self.timer = Timer(create_time=self.reactor.time)
 
18
 
 
19
    def test_since_start(self):
 
20
        self.assertEquals(self.timer.since_start(), 0.0)
 
21
 
 
22
        self.reactor.advance(1)
 
23
        self.assertEquals(self.timer.since_start(), 1.0)
 
24
 
 
25
        self.reactor.advance(2)
 
26
        self.assertEquals(self.timer.since_start(), 3.0)
 
27
 
 
28
    def test_since_reset(self):
 
29
        self.reactor.advance(1)
 
30
        self.assertEquals(self.timer.since_reset(), 1.0)
 
31
 
 
32
        self.reactor.advance(1)
 
33
        self.assertEquals(self.timer.since_start(), 2.0)
 
34
 
 
35
        self.reactor.advance(2)
 
36
        self.timer.reset()
 
37
        self.assertEquals(self.timer.since_start(), 4.0)
 
38
 
 
39
 
 
40
class MonitorTest(ReactorHavingTest):
 
41
 
 
42
    def setUp(self):
 
43
        super(MonitorTest, self).setUp()
 
44
        self.monitor = Monitor("test", create_time=self.reactor.time)
 
45
 
 
46
    def test_ping(self):
 
47
        self.assertEquals(self.monitor.count, 0)
 
48
        self.assertEquals(self.monitor.total_count, 0)
 
49
 
 
50
        self.monitor.ping()
 
51
        self.assertEquals(self.monitor.count, 1)
 
52
        self.assertEquals(self.monitor.total_count, 1)
 
53
 
 
54
    def test_reset(self):
 
55
        self.assertEquals(self.monitor.count, 0)
 
56
 
 
57
        self.monitor.ping()
 
58
        self.monitor.ping()
 
59
        self.assertEquals(self.monitor.count, 2)
 
60
        self.assertEquals(self.monitor.total_count, 2)
 
61
 
 
62
        self.monitor.reset()
 
63
        self.monitor.ping()
 
64
        self.assertEquals(self.monitor.count, 1)
 
65
        self.assertEquals(self.monitor.total_count, 3)
 
66
 
 
67
    def test_log(self):
 
68
        for i in range(100):
 
69
            self.monitor.ping()
 
70
            self.reactor.advance(1)
 
71
        self.monitor.log()
 
72
        self.assertTrue("INFO: 100 test events occurred in the last 100.00s."
 
73
                        in self.logfile.getvalue())
 
74
 
 
75
 
 
76
class BurstMonitorTest(ReactorHavingTest):
 
77
 
 
78
    def setUp(self):
 
79
        super(BurstMonitorTest, self).setUp()
 
80
        self.monitor = BurstMonitor(60, 1, "test",
 
81
                                    create_time=self.reactor.time)
 
82
 
 
83
    def test_warn_no_pings(self):
 
84
        self.assertFalse(self.monitor.warn())
 
85
 
 
86
    def test_warn_below_threshold(self):
 
87
        self.monitor.ping()
 
88
        self.reactor.advance(61)
 
89
        self.assertFalse(self.monitor.warn())
 
90
 
 
91
    def test_warn_on_threshold(self):
 
92
        self.monitor.ping()
 
93
        self.reactor.advance(61)
 
94
        self.assertFalse(self.monitor.warn())
 
95
 
 
96
    def test_warn_over_threshold(self):
 
97
        self.monitor.ping()
 
98
        self.reactor.advance(30)
 
99
        self.monitor.ping()
 
100
        self.assertTrue(self.monitor.warn())
 
101
 
 
102
        self.reactor.advance(31)
 
103
        self.assertFalse(self.monitor.warn())
 
104
 
 
105
    def test_warn_in_first_interval(self):
 
106
        self.monitor.ping()
 
107
        self.reactor.advance(59)
 
108
        self.assertFalse(self.monitor.warn())
 
109
 
 
110
    def test_warn_unexpected_burst(self):
 
111
        self.monitor.ping()
 
112
        self.reactor.advance(5000)
 
113
        self.assertFalse(self.monitor.warn())
 
114
 
 
115
        self.monitor.ping()
 
116
        self.assertFalse(self.monitor.warn())
 
117
 
 
118
        self.monitor.ping()
 
119
        self.assertTrue(self.monitor.warn())
 
120
 
 
121
    def test_warn_maximum_count(self):
 
122
        monitor = BurstMonitor(60, 2, "test", create_time=self.reactor.time)
 
123
        monitor.ping()
 
124
        monitor.ping()
 
125
        self.assertFalse(monitor.warn())
 
126
 
 
127
        monitor.ping()
 
128
        self.assertTrue(monitor.warn())
 
129
 
 
130
    def test_warn_maximum_count_over_time_span(self):
 
131
        monitor = BurstMonitor(60, 3, "test", create_time=self.reactor.time)
 
132
        monitor.ping()
 
133
        monitor.ping()
 
134
        self.assertFalse(monitor.warn())
 
135
 
 
136
        self.reactor.advance(30)
 
137
        monitor.ping()
 
138
        self.assertFalse(monitor.warn())
 
139
 
 
140
        self.reactor.advance(31)
 
141
        monitor.ping()
 
142
        self.assertFalse(monitor.warn())
 
143
 
 
144
        monitor.ping()
 
145
        monitor.ping()
 
146
        self.assertTrue(monitor.warn())
 
147
 
 
148
 
 
149
class CoverageMonitorTest(ReactorHavingTest):
 
150
 
 
151
    def setUp(self):
 
152
        super(CoverageMonitorTest, self).setUp()
 
153
        self.monitor = CoverageMonitor(1, 1.0, "test",
 
154
                                       create_time=self.reactor.time)
 
155
 
 
156
    def test_warn(self):
 
157
        self.monitor.ping()
 
158
        self.reactor.advance(1)
 
159
        self.assertFalse(self.monitor.warn())
 
160
 
 
161
        self.reactor.advance(1)
 
162
        self.assertTrue(self.monitor.warn())
 
163
 
 
164
        self.monitor.reset()
 
165
        self.assertFalse(self.monitor.warn())
 
166
 
 
167
    def test_percent_no_data(self):
 
168
        """
 
169
        If no time has passed and the monitor hasn't received any
 
170
        pings it should return 100%.
 
171
        """
 
172
        self.assertEquals(self.monitor.percent, 1.0)
 
173
 
 
174
    def test_percent(self):
 
175
        self.reactor.advance(1)
 
176
        self.assertEquals(self.monitor.percent, 0.0)
 
177
 
 
178
        self.monitor.ping()
 
179
        self.reactor.advance(1)
 
180
        self.assertEquals(self.monitor.percent, 0.5)
 
181
 
 
182
    def test_percent_reset(self):
 
183
        self.reactor.advance(1)
 
184
        self.assertEquals(self.monitor.percent, 0.0)
 
185
 
 
186
        self.monitor.reset()
 
187
        self.monitor.ping()
 
188
        self.reactor.advance(1)
 
189
        self.assertEquals(self.monitor.percent, 1.0)
 
190
 
 
191
    def test_expected_count(self):
 
192
        self.reactor.advance(1)
 
193
        self.assertEquals(self.monitor.expected_count, 1.0)
 
194
 
 
195
        self.reactor.advance(1)
 
196
        self.assertEquals(self.monitor.expected_count, 2.0)
 
197
 
 
198
    def test_expected_count_reset(self):
 
199
        self.reactor.advance(1)
 
200
        self.assertEquals(self.monitor.expected_count, 1.0)
 
201
 
 
202
        self.monitor.reset()
 
203
        self.reactor.advance(1)
 
204
        self.assertEquals(self.monitor.expected_count, 1.0)
 
205
 
 
206
    def test_log(self):
 
207
        for i in range(100):
 
208
            self.monitor.ping()
 
209
            self.reactor.advance(1)
 
210
        self.monitor.log()
 
211
        self.assertTrue("INFO: 100 of 100 expected test events (100.00%) "
 
212
                        "occurred in the last 100.00s."
 
213
                        in self.logfile.getvalue())
 
214
 
 
215
    def test_log_warning(self):
 
216
        for i in range(100):
 
217
            self.reactor.advance(1)
 
218
        self.monitor.log()
 
219
        self.assertTrue("WARNING: 0 of 100 expected test events (0.00%) "
 
220
                        "occurred in the last 100.00s."
 
221
                        in self.logfile.getvalue())
 
222
 
 
223
 
 
224
class FrequencyMonitorTest(ReactorHavingTest):
 
225
 
 
226
    def setUp(self):
 
227
        super(FrequencyMonitorTest, self).setUp()
 
228
        self.monitor = FrequencyMonitor(100, 1, "test",
 
229
                                        create_time=self.reactor.time)
 
230
 
 
231
    def test_expected_count(self):
 
232
        self.assertEquals(self.monitor.expected_count, 0)
 
233
 
 
234
        self.reactor.advance(99)
 
235
        self.assertEquals(self.monitor.expected_count, 0)
 
236
 
 
237
        self.reactor.advance(1)
 
238
        self.assertEquals(self.monitor.expected_count, 1)
 
239
 
 
240
    def test_ping(self):
 
241
        self.assertFalse(self.monitor.warn())
 
242
 
 
243
        self.reactor.advance(80)
 
244
        self.monitor.ping()
 
245
        self.assertFalse(self.monitor.warn())
 
246
 
 
247
        self.reactor.advance(80)
 
248
        self.assertFalse(self.monitor.warn())
 
249
 
 
250
    def test_warn(self):
 
251
        self.assertFalse(self.monitor.warn())
 
252
 
 
253
        self.reactor.advance(101)
 
254
        self.assertTrue(self.monitor.warn())
 
255
 
 
256
    def test_log(self):
 
257
        self.monitor.ping()
 
258
        self.reactor.advance(100)
 
259
        self.monitor.log()
 
260
        self.assertTrue("minimum expected test events"
 
261
                        not in self.logfile.getvalue())
 
262
        self.reactor.advance(1)
 
263
        self.monitor.log()
 
264
        self.assertTrue("WARNING: Only 0 of 1 minimum expected test events "
 
265
                        "occurred in the last 100.00s."
 
266
                        in self.logfile.getvalue())