← Back to branch summary

~free.ekanayaka/landscape-client/lucid-1.5.4-0ubuntu0.10.04.0

« back to all changes in this revision

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

  • Committer: Bazaar Package Importer
  • Author(s): Free Ekanayaka
  • Date: 2010-06-28 18:07:18 UTC
  • mfrom: (1.2.5 upstream)
  • Revision ID: james.westby@ubuntu.com-20100628180718-vytyqgbtkiirv5sb
Tags: 1.5.2.1-0ubuntu0.10.04.0
https://launchpad.net/bugs/597000
Filter duplicate network interfaces in get_active_interfaces (LP: #597000)

Show diffs side-by-side

added added

removed removed

Lines of Context:
landscape/manager/tests/test_plugin.py
 
1
from landscape.tests.helpers import LandscapeTest
 
2
from landscape.tests.helpers import ManagerHelper
 
3
from landscape.manager.plugin import ManagerPlugin, SUCCEEDED, FAILED
 
4
 
 
5
 
 
6
class BrokerPluginTest(LandscapeTest):
 
7
 
 
8
    helpers = [ManagerHelper]
 
9
 
 
10
    def test_call_with_operation_result_success(self):
 
11
        """
 
12
        A helper method exists which calls a function and sends an
 
13
        operation-result message based on the success of that method.
 
14
        """
 
15
        plugin = ManagerPlugin()
 
16
        plugin.register(self.manager)
 
17
        broker_service = self.broker_service
 
18
        broker_service.message_store.set_accepted_types(["operation-result"])
 
19
        message = {"operation-id": 12312}
 
20
        operation = lambda: None
 
21
 
 
22
        def assert_messages(ignored):
 
23
            messages = broker_service.message_store.get_pending_messages()
 
24
            self.assertMessages(messages,
 
25
                                [{"type": "operation-result",
 
26
                                  "status": SUCCEEDED,
 
27
                                  "operation-id": 12312}])
 
28
 
 
29
        result = plugin.call_with_operation_result(message, operation)
 
30
        return result.addCallback(assert_messages)
 
31
 
 
32
    def test_call_with_operation_result_error(self):
 
33
        """
 
34
        The helper for operation-results sends an appropriate message when an
 
35
        exception is raised from the given method.
 
36
        """
 
37
        self.log_helper.ignore_errors(RuntimeError)
 
38
        plugin = ManagerPlugin()
 
39
        plugin.register(self.manager)
 
40
        broker_service = self.broker_service
 
41
        broker_service.message_store.set_accepted_types(["operation-result"])
 
42
        message = {"operation-id": 12312}
 
43
 
 
44
        def operation():
 
45
            raise RuntimeError("What the crap!")
 
46
 
 
47
        def assert_messages(ignored):
 
48
            messages = broker_service.message_store.get_pending_messages()
 
49
            self.assertMessages(messages,
 
50
                                [{"type": "operation-result", "status": FAILED,
 
51
                                  "result-text": "RuntimeError: What the "
 
52
                                  "crap!", "operation-id": 12312}])
 
53
            logdata = self.logfile.getvalue()
 
54
            self.assertTrue("RuntimeError: What the crap!" in logdata, logdata)
 
55
 
 
56
        result = plugin.call_with_operation_result(message, operation)
 
57
        return result.addCallback(assert_messages)
 
58
 
 
59
    def test_call_with_operation_result_exchanges_urgently(self):
 
60
        """
 
61
        Operation results are reported to the server as quickly as possible.
 
62
        """
 
63
        plugin = ManagerPlugin()
 
64
        plugin.register(self.manager)
 
65
        broker_service = self.broker_service
 
66
        broker_service.message_store.set_accepted_types(["operation-result"])
 
67
        message = {"operation-id": 123}
 
68
        operation = lambda: None
 
69
 
 
70
        def assert_urgency(ignored):
 
71
            self.assertTrue(broker_service.exchanger.is_urgent())
 
72
 
 
73
        result = plugin.call_with_operation_result(message, operation)
 
74
        return result.addCallback(assert_urgency)