← Back to branch summary

~ubuntu-branches/ubuntu/trusty/mozjs24/trusty-proposed

« back to all changes in this revision

Viewing changes to js/src/testing/mozbase/mozsystemmonitor/mozsystemmonitor/test/test_resource_monitor.py

  • Committer: Package Import Robot
  • Author(s): Tim Lunn
  • Date: 2014-02-11 21:55:34 UTC
  • Revision ID: package-import@ubuntu.com-20140211215534-m1zyq5aj59md3y07
Tags: upstream-24.2.0
Import upstream version 24.2.0

Show diffs side-by-side

added added

removed removed

Lines of Context:
js/src/testing/mozbase/mozsystemmonitor/mozsystemmonitor/test/test_resource_monitor.py
 
1
# This Source Code Form is subject to the terms of the Mozilla Public
 
2
# License, v. 2.0. If a copy of the MPL was not distributed with this
 
3
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
4
 
 
5
import multiprocessing
 
6
import tempfile
 
7
import time
 
8
import unittest
 
9
 
 
10
try:
 
11
    import psutil
 
12
except ImportError:
 
13
    psutil = None
 
14
 
 
15
from mozsystemmonitor.resourcemonitor import (
 
16
    SystemResourceMonitor,
 
17
    SystemResourceUsage,
 
18
)
 
19
 
 
20
 
 
21
@unittest.skipIf(psutil is None, 'Resource monitor requires psutil.')
 
22
class TestResourceMonitor(unittest.TestCase):
 
23
    def test_basic(self):
 
24
        monitor = SystemResourceMonitor(poll_interval=0.5)
 
25
 
 
26
        monitor.start()
 
27
        time.sleep(3)
 
28
 
 
29
        monitor.stop()
 
30
 
 
31
        data = list(monitor.range_usage())
 
32
        self.assertGreater(len(data), 3)
 
33
 
 
34
        self.assertIsInstance(data[0], SystemResourceUsage)
 
35
 
 
36
    def test_empty(self):
 
37
        monitor = SystemResourceMonitor(poll_interval=2.0)
 
38
        monitor.start()
 
39
        monitor.stop()
 
40
 
 
41
        data = list(monitor.range_usage())
 
42
        self.assertEqual(len(data), 0)
 
43
 
 
44
    def test_phases(self):
 
45
        monitor = SystemResourceMonitor(poll_interval=0.25)
 
46
 
 
47
        monitor.start()
 
48
        time.sleep(1)
 
49
 
 
50
        with monitor.phase('phase1'):
 
51
            time.sleep(1)
 
52
 
 
53
            with monitor.phase('phase2'):
 
54
                time.sleep(1)
 
55
 
 
56
        monitor.stop()
 
57
 
 
58
        self.assertEqual(len(monitor.phases), 2)
 
59
        self.assertEqual(['phase2', 'phase1'], monitor.phases.keys())
 
60
 
 
61
        all = list(monitor.range_usage())
 
62
        data1 = list(monitor.phase_usage('phase1'))
 
63
        data2 = list(monitor.phase_usage('phase2'))
 
64
 
 
65
        self.assertGreater(len(all), len(data1))
 
66
        self.assertGreater(len(data1), len(data2))
 
67
 
 
68
        # This could fail if time.time() takes more than 0.1s. It really
 
69
        # shouldn't.
 
70
        self.assertAlmostEqual(data1[-1].end, data2[-1].end, delta=0.25)
 
71
 
 
72
    def test_no_data(self):
 
73
        monitor = SystemResourceMonitor()
 
74
 
 
75
        data = list(monitor.range_usage())
 
76
        self.assertEqual(len(data), 0)
 
77
 
 
78
    def test_events(self):
 
79
        monitor = SystemResourceMonitor(poll_interval=0.25)
 
80
 
 
81
        monitor.start()
 
82
        time.sleep(0.5)
 
83
 
 
84
        t0 = time.time()
 
85
        monitor.record_event('t0')
 
86
        time.sleep(0.5)
 
87
 
 
88
        t1 = time.time()
 
89
        monitor.record_event('t1')
 
90
        time.sleep(0.5)
 
91
        monitor.stop()
 
92
 
 
93
        events = monitor.events
 
94
        self.assertEqual(len(events), 2)
 
95
 
 
96
        event = events[0]
 
97
 
 
98
        self.assertEqual(event[1], 't0')
 
99
        self.assertAlmostEqual(event[0], t0, delta=0.25)
 
100
 
 
101
        data = list(monitor.between_events_usage('t0', 't1'))
 
102
        self.assertGreater(len(data), 0)
 
103
 
 
104
    def test_aggregate_cpu(self):
 
105
        monitor = SystemResourceMonitor(poll_interval=0.25)
 
106
 
 
107
        monitor.start()
 
108
        time.sleep(1)
 
109
        monitor.stop()
 
110
 
 
111
        values = monitor.aggregate_cpu_percent()
 
112
        self.assertIsInstance(values, list)
 
113
        self.assertEqual(len(values), multiprocessing.cpu_count())
 
114
        for v in values:
 
115
            self.assertIsInstance(v, float)
 
116
 
 
117
        value = monitor.aggregate_cpu_percent(per_cpu=False)
 
118
        self.assertIsInstance(value, float)
 
119
 
 
120
        values = monitor.aggregate_cpu_times()
 
121
        self.assertIsInstance(values, list)
 
122
        self.assertGreater(len(values), 0)
 
123
        self.assertTrue(hasattr(values[0], 'user'))
 
124
 
 
125
        t = type(values[0])
 
126
 
 
127
        value = monitor.aggregate_cpu_times(per_cpu=False)
 
128
        self.assertIsInstance(value, t)
 
129
 
 
130
    def test_aggregate_io(self):
 
131
        monitor = SystemResourceMonitor(poll_interval=0.25)
 
132
 
 
133
        # There's really no easy way to ensure I/O occurs. For all we know
 
134
        # reads and writes will all be serviced by the page cache.
 
135
        monitor.start()
 
136
        time.sleep(1.0)
 
137
        monitor.stop()
 
138
 
 
139
        values = monitor.aggregate_io()
 
140
        self.assertTrue(hasattr(values, 'read_count'))
 
141
 
 
142
    def test_memory(self):
 
143
        monitor = SystemResourceMonitor(poll_interval=0.25)
 
144
 
 
145
        monitor.start()
 
146
        time.sleep(1.0)
 
147
        monitor.stop()
 
148
 
 
149
        v = monitor.min_memory_available()
 
150
        self.assertIsInstance(v, long)
 
151
 
 
152
        v = monitor.max_memory_percent()
 
153
        self.assertIsInstance(v, float)
 
154