← Back to branch summary

~ubuntu-branches/ubuntu/natty/landscape-client/natty

« back to all changes in this revision

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

  • Committer: Bazaar Package Importer
  • Author(s): Free Ekanayaka
  • Date: 2010-04-21 19:58:10 UTC
  • mfrom: (1.1.16 upstream)
  • Revision ID: james.westby@ubuntu.com-20100421195810-s30uv3s6i27lue38
Tags: 1.5.2-0ubuntu0.10.10.0
https://launchpad.net/bugs/594594
https://launchpad.net/bugs/272344
https://launchpad.net/bugs/538253
https://launchpad.net/bugs/574810
* 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_plugin.py
 
1
from twisted.internet.defer import succeed
 
2
 
 
3
from landscape.monitor.plugin import MonitorPlugin, DataWatcher
 
4
from landscape.schema import Message, Int
 
5
from landscape.tests.mocker import ANY
 
6
from landscape.tests.helpers import (
 
7
    LandscapeTest, MonitorHelper, LogKeeperHelper)
 
8
 
 
9
 
 
10
class MonitorPluginTest(LandscapeTest):
 
11
 
 
12
    helpers = [MonitorHelper]
 
13
 
 
14
    def test_without_persist_name(self):
 
15
        """
 
16
        By default a L{MonitorPlugin} doesn't have a C{_persist} attribute.
 
17
        """
 
18
        plugin = MonitorPlugin()
 
19
        plugin.register(self.monitor)
 
20
        self.assertIs(plugin.persist, None)
 
21
 
 
22
    def test_with_persist_name(self):
 
23
        """
 
24
        When plugins providea C{persist_name} attribute, they get a persist
 
25
        object set at C{_persist} which is rooted at the name specified.
 
26
        """
 
27
        plugin = MonitorPlugin()
 
28
        plugin.persist_name = "wubble"
 
29
        plugin.register(self.monitor)
 
30
        plugin.persist.set("hi", "there")
 
31
        self.assertEquals(self.monitor.persist.get("wubble"), {"hi": "there"})
 
32
 
 
33
    def test_with_no_run_interval(self):
 
34
        """
 
35
        If the C{run_interval} attribute of L{MonitorPlugin} is C{None}, its
 
36
        C{run} method won't get called by the reactor.
 
37
        """
 
38
        plugin = MonitorPlugin()
 
39
        plugin.run = lambda: 1 / 0
 
40
        plugin.run_interval = None
 
41
        plugin.register(self.monitor)
 
42
        self.reactor.advance(MonitorPlugin.run_interval)
 
43
 
 
44
    def test_call_on_accepted(self):
 
45
        """
 
46
        L{MonitorPlugin}-based plugins can provide a callable to call
 
47
        when a message type becomes accepted.
 
48
        """
 
49
        plugin = MonitorPlugin()
 
50
        plugin.register(self.monitor)
 
51
        callback = self.mocker.mock()
 
52
        callback("foo", kwarg="bar")
 
53
        self.mocker.replay()
 
54
        plugin.call_on_accepted("type", callback, "foo", kwarg="bar")
 
55
        self.reactor.fire(("message-type-acceptance-changed", "type"), True)
 
56
 
 
57
    def test_call_on_accepted_when_unaccepted(self):
 
58
        """
 
59
        Notifications are only dispatched to plugins when types become
 
60
        accepted, not when they become unaccepted.
 
61
        """
 
62
        plugin = MonitorPlugin()
 
63
        plugin.register(self.monitor)
 
64
        callback = lambda: 1 / 0
 
65
        plugin.call_on_accepted("type", callback)
 
66
        self.reactor.fire(("message-type-acceptance-changed", "type"), False)
 
67
 
 
68
 
 
69
class StubDataWatchingPlugin(DataWatcher):
 
70
 
 
71
    persist_name = "ooga"
 
72
    message_type = "wubble"
 
73
    message_key = "wubblestuff"
 
74
 
 
75
    def __init__(self, data=None):
 
76
        self.data = data
 
77
 
 
78
    def get_data(self):
 
79
        return self.data
 
80
 
 
81
 
 
82
class DataWatcherTest(LandscapeTest):
 
83
 
 
84
    helpers = [MonitorHelper, LogKeeperHelper]
 
85
 
 
86
    def setUp(self):
 
87
        LandscapeTest.setUp(self)
 
88
        self.plugin = StubDataWatchingPlugin(1)
 
89
        self.plugin.register(self.monitor)
 
90
        self.mstore.add_schema(Message("wubble", {"wubblestuff": Int()}))
 
91
 
 
92
    def test_get_message(self):
 
93
        self.assertEquals(self.plugin.get_message(),
 
94
                          {"type": "wubble", "wubblestuff": 1})
 
95
 
 
96
    def test_get_message_unchanging(self):
 
97
        self.assertEquals(self.plugin.get_message(),
 
98
                          {"type": "wubble", "wubblestuff": 1})
 
99
        self.assertEquals(self.plugin.get_message(), None)
 
100
 
 
101
    def test_basic_exchange(self):
 
102
        # Is this really want we want to do?
 
103
        self.mstore.set_accepted_types(["wubble"])
 
104
        self.plugin.exchange()
 
105
        messages = self.mstore.get_pending_messages()
 
106
        self.assertEquals(messages[0]["type"], "wubble")
 
107
        self.assertEquals(messages[0]["wubblestuff"], 1)
 
108
        self.assertIn("Queueing a message with updated data watcher info for "
 
109
                      "landscape.monitor.tests.test_plugin.StubDataWatching"
 
110
                      "Plugin.", self.logfile.getvalue())
 
111
 
 
112
    def test_unchanging_value(self):
 
113
        # Is this really want we want to do?
 
114
        self.mstore.set_accepted_types(["wubble"])
 
115
        self.plugin.exchange()
 
116
        self.plugin.exchange()
 
117
        messages = self.mstore.get_pending_messages()
 
118
        self.assertEquals(len(messages), 1)
 
119
 
 
120
    def test_urgent_exchange(self):
 
121
        """
 
122
        When exchange is called with an urgent argument set to True
 
123
        make sure it sends the message urgently.
 
124
        """
 
125
        remote_broker_mock = self.mocker.replace(self.remote)
 
126
        remote_broker_mock.send_message(ANY, urgent=True)
 
127
        self.mocker.result(succeed(None))
 
128
        self.mocker.replay()
 
129
 
 
130
        self.mstore.set_accepted_types(["wubble"])
 
131
        self.plugin.exchange(True)
 
132
 
 
133
    def test_no_message_if_not_accepted(self):
 
134
        """
 
135
        Don't add any messages at all if the broker isn't currently
 
136
        accepting their type.
 
137
        """
 
138
        self.mstore.set_accepted_types([])
 
139
        self.reactor.advance(self.monitor.step_size * 2)
 
140
        self.monitor.exchange()
 
141
 
 
142
        self.mstore.set_accepted_types(["wubble"])
 
143
        self.assertMessages(list(self.mstore.get_pending_messages()), [])