← 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/manager/tests/test_manager.py

  • Committer: Bazaar Package Importer
  • Author(s): Rick Clark
  • Date: 2008-09-08 16:35:57 UTC
  • mfrom: (1.1.1 upstream)
  • Revision ID: james.westby@ubuntu.com-20080908163557-l3ixzj5dxz37wnw2
Tags: 1.0.18-0ubuntu1
New upstream release 

Show diffs side-by-side

added added

removed removed

Lines of Context:
landscape/manager/tests/test_manager.py
 
1
from twisted.internet.defer import Deferred
 
2
 
 
3
from landscape.manager.deployment import ManagerService, ManagerConfiguration
 
4
from landscape.manager.manager import (
 
5
    ManagerPlugin, ManagerDBusObject, SUCCEEDED, FAILED)
 
6
 
 
7
from landscape.lib.dbus_util import get_object
 
8
 
 
9
from landscape.tests.helpers import (
 
10
    LandscapeTest, LandscapeIsolatedTest, ManagerHelper)
 
11
 
 
12
 
 
13
class PluginOperationResultTest(LandscapeTest):
 
14
 
 
15
    helpers = [ManagerHelper]
 
16
 
 
17
    def test_call_with_operation_result_success(self):
 
18
        """
 
19
        A helper method exists which calls a function and sends an
 
20
        operation-result message based on the success of that method.
 
21
        """
 
22
        plugin = ManagerPlugin()
 
23
        plugin.register(self.manager)
 
24
        service = self.broker_service
 
25
        service.message_store.set_accepted_types(["operation-result"])
 
26
        message = {"operation-id": 12312}
 
27
        def operation():
 
28
            pass
 
29
        plugin.call_with_operation_result(message, operation)
 
30
        messages = self.broker_service.message_store.get_pending_messages()
 
31
        self.assertMessages(messages,
 
32
                            [{"type": "operation-result", "status": SUCCEEDED,
 
33
                              "operation-id": 12312}])
 
34
 
 
35
    def test_call_with_operation_result_error(self):
 
36
        """
 
37
        The helper for operation-results sends an appropriate message when an
 
38
        exception is raised from the given method.
 
39
        """
 
40
        self.log_helper.ignore_errors(RuntimeError)
 
41
        plugin = ManagerPlugin()
 
42
        plugin.register(self.manager)
 
43
        service = self.broker_service
 
44
        service.message_store.set_accepted_types(["operation-result"])
 
45
        message = {"operation-id": 12312}
 
46
        def operation():
 
47
            raise RuntimeError("What the crap!")
 
48
        plugin.call_with_operation_result(message, operation)
 
49
        messages = self.broker_service.message_store.get_pending_messages()
 
50
        self.assertMessages(messages,
 
51
                            [{"type": "operation-result", "status": FAILED,
 
52
                              "result-text": "RuntimeError: What the crap!",
 
53
                              "operation-id": 12312}])
 
54
 
 
55
        logdata = self.logfile.getvalue()
 
56
        self.assertTrue("RuntimeError: What the crap!" in logdata, logdata)
 
57
 
 
58
    def test_call_with_operation_result_exchanges_urgently(self):
 
59
        """
 
60
        Operation results are reported to the server as quickly as possible.
 
61
        """
 
62
        plugin = ManagerPlugin()
 
63
        plugin.register(self.manager)
 
64
        service = self.broker_service
 
65
        service.message_store.set_accepted_types(["operation-result"])
 
66
        message = {"operation-id": 123}
 
67
        def operation():
 
68
            pass
 
69
        plugin.call_with_operation_result(message, operation)
 
70
        self.assertTrue(service.exchanger.is_urgent())
 
71
 
 
72
 
 
73
class ManagerDBusObjectTest(LandscapeIsolatedTest):
 
74
 
 
75
    helpers = [ManagerHelper]
 
76
 
 
77
    def setUp(self):
 
78
        super(ManagerDBusObjectTest, self).setUp()
 
79
        configuration = ManagerConfiguration()
 
80
        configuration.load(["-d", self.makeFile(), "--bus", "session",
 
81
                            "--manager-plugins", "ProcessKiller"])
 
82
        self.manager_service = ManagerService(configuration)
 
83
        self.broker_service.startService()
 
84
        self.manager_service.startService()
 
85
        self.dbus_object = get_object(self.broker_service.bus,
 
86
                                      ManagerDBusObject.bus_name,
 
87
                                      ManagerDBusObject.object_path)
 
88
 
 
89
    def tearDown(self):
 
90
        super(ManagerDBusObjectTest, self).tearDown()
 
91
        self.broker_service.stopService()
 
92
 
 
93
    def test_ping(self):
 
94
        result = self.dbus_object.ping()
 
95
        def got_result(result):
 
96
            self.assertEquals(result, True)
 
97
        return result.addCallback(got_result)
 
98
 
 
99
    def test_exit(self):
 
100
        result = Deferred()
 
101
        reactor = self.mocker.replace("twisted.internet.reactor")
 
102
        self.expect(reactor.stop()).call(lambda: result.callback(None))
 
103
        self.mocker.replay()
 
104
        self.dbus_object.exit()
 
105
        return result