/usr/share/pyshared/JobService/root.py is in jobservice 0.8.0-0ubuntu4.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 | # This file is part of jobservice.
# Copyright 2010 Jacob Peddicord <jpeddicord@ubuntu.com>
#
# jobservice is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# jobservice is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with jobservice. If not, see <http://www.gnu.org/licenses/>.
import sys
import logging
from dbus.service import Object as DBusObject, method as DBusMethod
from JobService import DBUS_PATH, DBUS_IFACE
from JobService.backends import ServiceProxy
from JobService.job import SingleJobService
from JobService.policy import Policy
from JobService.util import dbus_safe_name
log = logging.getLogger('jobservice')
class RootJobService(DBusObject):
def __init__(self, conn=None, object_path=None, bus_name=None, idle=None, enforce=True):
"""
Fire up this service as well as all of the paths for individual jobs.
"""
DBusObject.__init__(self, conn, object_path, bus_name)
self.idle = idle
self.policy = Policy(enforce)
self.proxy = ServiceProxy()
self.jobs = []
allsvcs = self.proxy.get_all_services()
for job in allsvcs:
self.jobs.append(SingleJobService(conn,
'/'.join((DBUS_PATH, dbus_safe_name(job))),
name=job, root=self
))
log.info('Ready')
@DBusMethod(DBUS_IFACE, in_signature='', out_signature='a(so)',
sender_keyword='sender', connection_keyword='conn')
def GetAllJobs(self, sender=None, conn=None):
"""
Returns all jobs known by the backend(s), along with their states.
Returns array of struct (
string name
object path
)
"""
self.idle.ping()
log.debug("GetAllJobs called")
svclist = []
for job in self.jobs:
svclist.append((job.name, job.path))
return svclist
|