/usr/lib/python2.7/dist-packages/framework/subsystems/testing/testing.py is in fso-frameworkd 0.10.1-3.
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 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 | #!/usr/bin/env python
"""
Dummy Subsystem for Testing Purposes
(C) 2008-2009 Michael 'Mickey' Lauer <mlauer@vanille-media.de>
(C) 2008 Openmoko, Inc.
GPLv2 or later
Package: testing
Module: testing
"""
MODULE_NAME = "testing"
__version__ = "0.0.0"
from framework import resource
import dbus
import dbus.service
import gobject
import logging
logger = logging.getLogger( MODULE_NAME )
import time
DBUS_INTERFACE = "org.freesmartphone.Testing"
DBUS_OBJECT_PATH = "/org/freesmartphone/Testing"
#============================================================================#
class Resource( resource.Resource ):
#============================================================================#
def __init__( self, bus ):
self.path = DBUS_OBJECT_PATH
self.bus = bus
self.virgin = True
dbus.service.Object.__init__( self, bus, self.path )
resource.Resource.__init__( self, bus, "TEST" )
logger.info("%s %s at %s initialized.", self.__class__.__name__, __version__, self.path )
# default behaviour: everything works
self.catmap = { "enabling":"ok",
"disabling":"ok",
"suspending":"ok",
"resuming":"ok" }
#
# framework.Resource
#
def _enable( self, on_ok, on_error ):
logger.info( "enabling" )
time.sleep( 5.0 )
self._doit( "enabling", on_ok, on_error )
def _disable( self, on_ok, on_error ):
logger.info( "disabling" )
if self.virgin == True:
self.virgin = False
else:
time.sleep( 5.0 )
self._doit( "disabling", on_ok, on_error )
def _suspend( self, on_ok, on_error ):
logger.info( "suspending" )
time.sleep( 5.0 )
self._doit( "suspending", on_ok, on_error )
def _resume( self, on_ok, on_error ):
logger.info("resuming")
time.sleep( 5.0 )
self._doit( "resuming", on_ok, on_error )
def _doit( self, category, on_ok, on_error ):
action = self.catmap[ category ]
if action == "ok":
on_ok()
elif action == "error":
on_error( "unspecified" )
elif action == "veto":
on_error( resource.SuspendVeto( "not allowed to suspend this resource" ) )
else:
foobar
#
# dbus interface
#
@dbus.service.method( DBUS_INTERFACE, "", "",
async_callbacks=( "dbus_ok", "dbus_error" ) )
@resource.checkedmethod
def SetResourceBehaviour( self, category, behaviour, dbus_ok, dbus_error ):
try:
value = self.catmap[category]
except KeyError:
dbus_error( "unknown category, valid categories are: %s" % self.catmap.keys() )
else:
if behaviour not in "ok error veto".split():
dbus_error( "unknown behaviour. valid behaviours are: ok error veto" )
self.catmap[category] = str( behaviour )
dbus_ok()
@dbus.service.method( DBUS_INTERFACE, "", "aa{sv}",
async_callbacks=( "dbus_ok", "dbus_error" ) )
@resource.checkedmethod
def ReturnTest( self, dbus_ok, dbus_error ):
d = {"foo":"bar"}
dbus_ok( [d,d] )
@dbus.service.method( DBUS_INTERFACE, "", "",
async_callbacks=( "dbus_ok", "dbus_error" ) )
@resource.checkedmethod
def SignalTest( self, dbus_ok, dbus_error ):
self.Test( dict(yo="kurt") )
dbus_ok()
@dbus.service.signal( DBUS_INTERFACE, "a{sv}" )
def Test( self, asv ):
logger.info( "emitting signal" )
#============================================================================#
def factory(prefix, controller):
#============================================================================#
"""This is the magic function that will be called by the framework module manager"""
return [ Resource( controller.bus ) ]
#============================================================================#
if __name__ == "__main__":
#============================================================================#
pass
|