← Back to branch summary

~free.ekanayaka/landscape-client/jaunty-1.5.2.1-0ubuntu0.9.04.0

« back to all changes in this revision

Viewing changes to landscape/monitor/tests/test_networkactivity.py

  • Committer: Free Ekanayaka
  • Date: 2010-06-16 10:01:20 UTC
  • Revision ID: free.ekanayaka@canonical.com-20100616100120-k185twz19yf1qqmh
https://launchpad.net/bugs/272344
https://launchpad.net/bugs/538253
https://launchpad.net/bugs/574810
https://launchpad.net/bugs/594594
* New upstream version (LP: #594594):
  - A new includes information about active network devices and their
    IP address in sysinfo output (LP: #272344).
  - A new plugin collects information about network traffic (#LP :284662).
  - Report information about which packages requested a reboot (LP: #538253).
  - Fix breakage on Lucid AMIs having no ramdisk (LP: #574810).
  - Migrate the inter-process communication system from DBus to Twisted AMP.

Show diffs side-by-side

added added

removed removed

Lines of Context:
landscape/monitor/tests/test_networkactivity.py
 
1
import socket
 
2
from landscape.monitor.networkactivity import NetworkActivity
 
3
from landscape.tests.helpers import LandscapeTest, MonitorHelper
 
4
 
 
5
 
 
6
class NetworkActivityTest(LandscapeTest):
 
7
 
 
8
    helpers = [MonitorHelper]
 
9
 
 
10
    stats_template = """\
 
11
Inter-|   Receive                           |  Transmit
 
12
 face |bytes    packets compressed multicast|bytes    packets errs drop fifo
 
13
    lo:%(lo_in)d   0       0         0       %(lo_out)d 3321049    0    0    0
 
14
    eth0: %(eth0_in)d   12539      0     62  %(eth0_out)d   12579    0    0   0
 
15
    %(extra)s
 
16
"""
 
17
 
 
18
    def setUp(self):
 
19
        super(NetworkActivityTest, self).setUp()
 
20
        self.activity_file = open(self.makeFile(), "w+")
 
21
        self.write_activity()
 
22
        self.plugin = NetworkActivity(
 
23
            network_activity_file = self.activity_file.name,
 
24
            create_time = self.reactor.time)
 
25
        self.monitor.add(self.plugin)
 
26
 
 
27
    def tearDown(self):
 
28
        self.activity_file.close()
 
29
        super(NetworkActivityTest, self).tearDown()
 
30
 
 
31
    def write_activity(self, lo_in=0, lo_out=0, eth0_in=0, eth0_out=0,
 
32
                        extra="", **kw):
 
33
        kw.update(dict(
 
34
            lo_in = lo_in,
 
35
            lo_out = lo_out,
 
36
            eth0_in = eth0_in,
 
37
            eth0_out = eth0_out,
 
38
            extra=extra))
 
39
        self.activity_file.truncate()
 
40
        self.activity_file.write(self.stats_template % kw)
 
41
        self.activity_file.flush()
 
42
 
 
43
    def test_read_proc_net_dev(self):
 
44
        """
 
45
        When the network activity plugin runs it reads data from
 
46
        /proc/net/dev which it parses and accumulates to read values.
 
47
        This test ensures that /proc/net/dev is always parseable and
 
48
        that messages are in the expected format and contain data with
 
49
        expected datatypes.
 
50
        """
 
51
        plugin = NetworkActivity(create_time = self.reactor.time)
 
52
        self.monitor.add(plugin)
 
53
        plugin.run()
 
54
        self.reactor.advance(self.monitor.step_size)
 
55
        # hmmm. try to connect anywhere to advance the net stats
 
56
        try:
 
57
            socket.socket().connect(("localhost", 9999))
 
58
        except socket.error:
 
59
            pass
 
60
        plugin.run()
 
61
        message = plugin.create_message()
 
62
        self.assertTrue(message)
 
63
 
 
64
    def test_message_contents(self):
 
65
        """
 
66
        The network plugin sends messages with the traffic delta along with the
 
67
        step per network interface. Only interfaces which have deltas are
 
68
        present in the message.
 
69
        """
 
70
        self.write_activity(lo_in=2000, lo_out=1900)
 
71
        self.plugin.run()
 
72
        self.reactor.advance(self.monitor.step_size)
 
73
        self.write_activity(lo_in=2010, lo_out=1999)
 
74
        self.plugin.run()
 
75
        message = self.plugin.create_message()
 
76
        self.assertTrue(message)
 
77
        self.assertTrue("type" in message)
 
78
        self.assertEquals(message["type"], "network-activity")
 
79
        self.assertEquals(message["activities"]["lo"],
 
80
                          [(300, 10, 99)])
 
81
        self.assertNotIn("eth0", message["activities"])
 
82
 
 
83
    def test_no_message_without_traffic_delta(self):
 
84
        """
 
85
        If no traffic delta is detected between runs, no message will be
 
86
        generated by the plugin.
 
87
        """
 
88
        self.plugin.run()
 
89
        self.reactor.advance(self.monitor.step_size)
 
90
        message = self.plugin.create_message()
 
91
        self.assertFalse(message)
 
92
        self.plugin.run()
 
93
        message = self.plugin.create_message()
 
94
        self.assertFalse(message)
 
95
 
 
96
    def test_no_message_without_traffic_delta_across_steps(self):
 
97
        """
 
98
        A traffic delta needs to cross step boundaries before a message
 
99
        is generated.
 
100
        """
 
101
        self.plugin.run()
 
102
        self.write_activity(lo_out=1000, eth0_out=1000)
 
103
        self.reactor.advance(self.monitor.step_size)
 
104
        message = self.plugin.create_message()
 
105
        self.assertFalse(message)
 
106
 
 
107
    def test_interface_temporarily_disappears(self):
 
108
        """
 
109
        When an interface is removed (ie usb hotplug) and then activated again
 
110
        its delta will be retained.
 
111
        """
 
112
        self.write_activity(extra="wlan0: 2222 0 0 0 2222 0 0 0 0")
 
113
        self.plugin.run()
 
114
        self.reactor.advance(self.monitor.step_size)
 
115
        self.write_activity()
 
116
        self.plugin.run()
 
117
        message = self.plugin.create_message()
 
118
        self.assertFalse(message)
 
119
        self.write_activity(extra="wlan0: 3333 0 0 0 3333 0 0 0 0")
 
120
        self.reactor.advance(self.monitor.step_size)
 
121
        self.plugin.run()
 
122
        message = self.plugin.create_message()
 
123
        self.assertTrue(message)
 
124
 
 
125
    def test_messaging_flushes(self):
 
126
        """
 
127
        Duplicate message should never be created.  If no data is available, no
 
128
        message is created.
 
129
        """
 
130
        self.plugin.run()
 
131
        self.reactor.advance(self.monitor.step_size)
 
132
        self.write_activity(eth0_out=1111)
 
133
        self.plugin.run()
 
134
        message = self.plugin.create_message()
 
135
        self.assertTrue(message)
 
136
        message = self.plugin.create_message()
 
137
        self.assertFalse(message)
 
138
 
 
139
    def test_exchange_no_message(self):
 
140
        """
 
141
        No message is sent to the exchange if there isn't a traffic delta.
 
142
        """
 
143
        self.reactor.advance(self.monitor.step_size)
 
144
        self.mstore.set_accepted_types([self.plugin.message_type])
 
145
        self.plugin.exchange()
 
146
        self.assertFalse(self.mstore.count_pending_messages())
 
147
 
 
148
    def test_exchange_messages(self):
 
149
        """
 
150
        The network plugin queues message when an exchange happens. Each
 
151
        message should be aligned to a step boundary; messages collected
 
152
        between exchange periods should be delivered in a single message.
 
153
        """
 
154
        self.reactor.advance(self.monitor.step_size)
 
155
        self.write_activity(lo_out=1000, eth0_out=1000)
 
156
        self.plugin.run()
 
157
        self.mstore.set_accepted_types([self.plugin.message_type])
 
158
        self.plugin.exchange()
 
159
        step_size = self.monitor.step_size
 
160
        self.assertMessages(self.mstore.get_pending_messages(),
 
161
                        [{"type": "network-activity",
 
162
                          "activities": {
 
163
                              "lo": [(step_size, 0, 1000)],
 
164
                              "eth0": [(step_size, 0, 1000)]}}])
 
165
 
 
166
    def test_config(self):
 
167
        """The network activity plugin is enabled by default."""
 
168
        self.assertIn("NetworkActivity", self.config.plugin_factories)