← Back to branch summary

~ubuntu-branches/ubuntu/lucid/checkbox/lucid

« back to all changes in this revision

Viewing changes to plugins/backend_manager.py

  • Committer: Bazaar Package Importer
  • Author(s): Dave Murphy, Dave Murphy, Andy Whitcroft
  • Date: 2009-03-17 09:46:16 UTC
  • Revision ID: james.westby@ubuntu.com-20090317094616-1p1ifmdy702ilpl9
Tags: 0.7
https://launchpad.net/bugs/331481
https://launchpad.net/bugs/331055
https://launchpad.net/bugs/331056
https://launchpad.net/bugs/331057
[ Dave Murphy ]
* Fixed viewing of report files in Firefox 3 (LP: #331481)
* Added additional contextual information
 * /etc/sysctl* (LP: #331055)
 * /etc/modprobe.d (LP: #331056)
 * /etc/modules (LP: #331057)
* Fixed packaging for Jaunty
 * https://lists.ubuntu.com/archives/ubuntu-devel/2009-February/027439.html
 * Uses --install-layout=deb
 * Installs to dist-packages instead of site-packages

[ Andy Whitcroft ]
* suspend_test: update suspend_test to version V6 matching kernel version.
  The version here will become the master copy.
* suspend_test: add a --dry-run mode to simplify developement
* suspend_test: add a automation mode for checkbox integration
* suspend_test: add a new pm-suspend test
* suspend_test: record and restore timer_delay around the variable
  time test.
* suspend_test: release v7.
* suspend_test: initial version of suspend power consumption test
  from a patch by Pete Graner.
* suspend_test: power -- made the sleep time configurable
* suspend_test: detect batteries and disable ac/power tests
* suspend_test: disable dbus tests when we have no primary user
* suspend_test: handle AC transitions better
* suspend_test: enable power test as part of --full
* suspend_test: reduce the noise in the test instructions
* suspend_test: use minutes in output when that is more appropriate
* suspend_test: track actual AC transitions and report them
* suspend_test: only mention AC at all if we have a battery
* suspend_test: report useful data at the bottom for posting
* suspend_test: document the new power test in the usage
* suspend_test: power -- indicate when the result is unreliable
* suspend_test: report -- fix up spacing issues
* suspend_test: release v8

Show diffs side-by-side

added added

removed removed

Lines of Context:
plugins/backend_manager.py
 
1
#
 
2
# This file is part of Checkbox.
 
3
#
 
4
# Copyright 2008 Canonical Ltd.
 
5
#
 
6
# Checkbox is free software: you can redistribute it and/or modify
 
7
# it under the terms of the GNU General Public License as published by
 
8
# the Free Software Foundation, either version 3 of the License, or
 
9
# (at your option) any later version.
 
10
#
 
11
# Checkbox is distributed in the hope that it will be useful,
 
12
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
13
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
14
# GNU General Public License for more details.
 
15
#
 
16
# You should have received a copy of the GNU General Public License
 
17
# along with Checkbox.  If not, see <http://www.gnu.org/licenses/>.
 
18
#
 
19
 
 
20
import logging
 
21
 
 
22
import gobject
 
23
import dbus
 
24
import dbus.service
 
25
import dbus.mainloop.glib
 
26
 
 
27
from checkbox.lib.environ import append_path
 
28
 
 
29
from checkbox.properties import Int, String
 
30
 
 
31
 
 
32
class PermissionDeniedByPolicy(dbus.DBusException):
 
33
    _dbus_error_name = "com.ubuntu.checkbox.PermissionDeniedByPolicy"
 
34
 
 
35
 
 
36
class UnknownRegistryException(dbus.DBusException):
 
37
    _dbus_error_name = 'com.ubuntu.DeviceDriver.InvalidDriverDBException'
 
38
 
 
39
class UnknownTestException(dbus.DBusException):
 
40
    _dbus_error_name = 'com.ubuntu.DeviceDriver.InvalidDriverDBException'
 
41
 
 
42
 
 
43
class BackendManager(dbus.service.Object):
 
44
 
 
45
    DBUS_INTERFACE_NAME = "com.ubuntu.checkbox"
 
46
 
 
47
    DBUS_BUS_NAME = "com.ubuntu.checkbox"
 
48
 
 
49
    timeout = Int(default=600)
 
50
    path = String(default="/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin")
 
51
 
 
52
    def __init__(self):
 
53
        self.dbus_info = None
 
54
        self.polkit = None
 
55
        self.loop = False
 
56
        self.tests = {}
 
57
 
 
58
    def __repr__(self):
 
59
        return "<BackendManager>"
 
60
 
 
61
    def register(self, manager):
 
62
        import dbus.mainloop.glib
 
63
        dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
 
64
        self.bus = dbus.SystemBus()
 
65
        self.dbus_name = dbus.service.BusName(self.DBUS_BUS_NAME, self.bus)
 
66
 
 
67
        # Set environment which has been reset by dbus
 
68
        for path in self.path.split(":"):
 
69
            append_path(path)
 
70
 
 
71
        self._manager = manager
 
72
        for (rt, rh) in [
 
73
             ("test-.*", self.test_all),
 
74
             ("run", self.run)]:
 
75
            self._manager.reactor.call_on(rt, rh)
 
76
 
 
77
    @dbus.service.method(DBUS_INTERFACE_NAME,
 
78
        in_signature="s", out_signature="s", sender_keyword="sender",
 
79
        connection_keyword="conn")
 
80
    def get_registry(self, name, sender=None, conn=None):
 
81
        self._check_polkit_privilege(sender, conn, "com.ubuntu.checkbox.info")
 
82
        if name not in self._manager.registry:
 
83
            raise UnknownRegistryException, "Registry not found: %s" % name
 
84
 
 
85
        return str(self._manager.registry[name])
 
86
 
 
87
    @dbus.service.method(DBUS_INTERFACE_NAME,
 
88
        in_signature="ss", out_signature="as", sender_keyword="sender",
 
89
        connection_keyword="conn")
 
90
    def get_test_result(self, suite, name, sender=None, conn=None):
 
91
        self._check_polkit_privilege(sender, conn, "com.ubuntu.checkbox.test")
 
92
        if (suite, name) not in self.tests:
 
93
            raise UnknownTestException, \
 
94
                "Suite/test not found: %s/%s" % (suite, name)
 
95
 
 
96
        test = self.tests[(suite, name)]
 
97
        result = test.command()
 
98
        return (result.status, result.data, str(result.duration))
 
99
 
 
100
    @dbus.service.method(DBUS_INTERFACE_NAME,
 
101
        in_signature="ss", out_signature="s", sender_keyword="sender",
 
102
        connection_keyword="conn")
 
103
    def get_test_description(self, suite, name, sender=None, conn=None):
 
104
        self._check_polkit_privilege(sender, conn, "com.ubuntu.checkbox.test")
 
105
        if (suite, name) not in self.tests:
 
106
            raise UnknownTestException, \
 
107
                "Suite/test not found: %s/%s" % (suite, name)
 
108
 
 
109
        test = self.tests[(suite, name)]
 
110
        return test.description()
 
111
 
 
112
    def test_all(self, test):
 
113
        self.tests[(test.suite, test.name)] = test
 
114
 
 
115
    def run(self):
 
116
        self._manager.reactor.fire("gather")
 
117
 
 
118
        # Delay instantiating the service after register
 
119
        dbus.service.Object.__init__(self, self.bus, "/checkbox")
 
120
        main_loop = gobject.MainLoop()
 
121
        self.loop = False
 
122
        if self.timeout:
 
123
            def _t():
 
124
                main_loop.quit()
 
125
                return True
 
126
            gobject.timeout_add(self.timeout * 1000, _t)
 
127
 
 
128
        # run until we time out
 
129
        while not self.loop:
 
130
            if self.timeout:
 
131
                self.loop = True
 
132
            main_loop.run()
 
133
 
 
134
    def _check_polkit_privilege(self, sender, conn, privilege):
 
135
        """Verify that sender has a given PolicyKit privilege.
 
136
 
 
137
        sender is the sender's (private) D-BUS name, such as ":1:42"
 
138
        (sender_keyword in @dbus.service.methods). conn is
 
139
        the dbus.Connection object (connection_keyword in
 
140
        @dbus.service.methods). privilege is the PolicyKit privilege string.
 
141
 
 
142
        This method returns if the caller is privileged, and otherwise throws a
 
143
        PermissionDeniedByPolicy exception.
 
144
        """
 
145
        if sender is None and conn is None:
 
146
            # called locally, not through D-BUS
 
147
            return
 
148
 
 
149
        # get peer PID
 
150
        if self.dbus_info is None:
 
151
            self.dbus_info = dbus.Interface(conn.get_object("org.freedesktop.DBus",
 
152
                "/org/freedesktop/DBus/Bus", False), "org.freedesktop.DBus")
 
153
        pid = self.dbus_info.GetConnectionUnixProcessID(sender)
 
154
 
 
155
        # query PolicyKit
 
156
        if self.polkit is None:
 
157
            self.polkit = dbus.Interface(dbus.SystemBus().get_object(
 
158
                "org.freedesktop.PolicyKit", "/", False),
 
159
                "org.freedesktop.PolicyKit")
 
160
        try:
 
161
            res = self.polkit.IsProcessAuthorized(privilege, pid, True)
 
162
        except dbus.DBusException, e:
 
163
            if e._dbus_error_name == "org.freedesktop.DBus.Error.ServiceUnknown":
 
164
                # polkitd timed out, connect again
 
165
                self.polkit = None
 
166
                return self._check_polkit_privilege(sender, conn, privilege)
 
167
            else:
 
168
                raise
 
169
 
 
170
        if res != "yes":
 
171
            logging.debug("_check_polkit_privilege: sender %s "
 
172
                "on connection %s pid %i requested %s: %s",
 
173
                sender, conn, pid, privilege, res)
 
174
            raise PermissionDeniedByPolicy(privilege + " " + res)
 
175
 
 
176
 
 
177
factory = BackendManager