test/test-session

   1 #!/usr/bin/python
   2
   3 import sys
   4 import gobject
   5 import string
   6
   7 import dbus
   8 import dbus.service
   9 import dbus.mainloop.glib
  10
  11 import glib
  12
  13 import traceback
  14
  15 def extract_list(list):
  16         val = "["
  17         for i in list:
  18                 val += " " + str(i)
  19         val += " ]"
  20         return val
  21
  22 def extract_values(values):
  23         val = "{"
  24         for key in values.keys():
  25                 val += " " + key + "="
  26                 if key in ["PrefixLength"]:
  27                         val += "%s" % (int(values[key]))
  28                 else:
  29                         if key in ["Servers", "Excludes"]:
  30                                 val += extract_list(values[key])
  31                         else:
  32                                 val += str(values[key])
  33         val += " }"
  34         return val
  35
  36 class Notification(dbus.service.Object):
  37         def __init__(self, bus, app, notify_path):
  38                 dbus.service.Object.__init__(self)
  39                 self.app = app
  40
  41         @dbus.service.method("net.connman.Notification",
  42                                 in_signature='', out_signature='')
  43         def Release(self):
  44                 print("Release")
  45                 self.app.release()
  46
  47         @dbus.service.method("net.connman.Notification",
  48                                 in_signature='a{sv}', out_signature='')
  49         def Update(self, settings):
  50                 print "Update called"
  51
  52                 try:
  53                         for key in settings.keys():
  54                                 if key in ["IPv4", "IPv6"]:
  55                                         val = extract_values(settings[key])
  56                                 elif key in  ["AllowedBearers"]:
  57                                         val = extract_list(settings[key])
  58                                 else:
  59                                         val = settings[key]
  60                                 print "    %s = %s" % (key, val)
  61                 except:
  62                         print "Exception:"
  63                         traceback.print_exc()
  64
  65 class SessionApplication(dbus.service.Object):
  66         def __init__(self, bus, object_path, mainloop):
  67                 dbus.service.Object.__init__(self, bus, object_path)
  68
  69                 self.manager = None
  70                 self.notify_path = object_path + "/notify"
  71                 self.notify = None
  72                 self.session_path = None
  73                 self.mainloop = mainloop
  74                 self.session = None
  75                 self.settings = { }
  76
  77                 try:
  78                         bus = dbus.SystemBus()
  79                         bus.watch_name_owner('net.connman', self.connman_name_owner_changed)
  80                 except dbus.DBusException:
  81                         traceback.print_exc()
  82                         exit(1)
  83
  84         def connman_name_owner_changed(self, proxy):
  85                 try:
  86                         if proxy:
  87                                 print "connman appeared on D-Bus ", str(proxy)
  88
  89                                 bus = dbus.SystemBus()
  90                                 self.manager = dbus.Interface(bus.get_object("net.connman", "/"),
  91                                                               "net.connman.Manager")
  92                         else:
  93                                 print "connman disappeared on D-Bus"
  94                                 self.manager = None
  95                                 self.session = None
  96                                 if self.notify:
  97                                         self.notify.remove_from_connection()
  98                                         self.notify = None
  99
 100                 except dbus.DBusException:
 101                         traceback.print_exc()
 102                         exit(1)
 103
 104         def release(self):
 105                 if self.session:
 106                         self.manager.DestroySession(self.session_path)
 107                         self.session = None
 108                 if self.notify:
 109                         self.notify.remove_from_connection()
 110                         self.notify = None
 111
 112         def type_convert(self, key, value):
 113                 if key in [ "AllowedBearers", "RoamingPolicy" ]:
 114                         return value
 115                 elif key in [ "Priority", "AvoidHandover",
 116                               "StayConnected", "EmergencyCall" ]:
 117                         flag = str(value[0]).strip().lower()
 118                         val = flag not in ['false', 'f', 'n', '0']
 119                         return dbus.Boolean(val)
 120                 elif key in [ "PeriodicConnect", "IdleTimeout" ]:
 121                         val = value[0]
 122                         return dbus.UInt32(val)
 123
 124                 return value
 125
 126         @dbus.service.method("com.example.TestSession",
 127                                 in_signature='', out_signature='')
 128         def CreateSession(self):
 129                 print "Create session"
 130
 131                 if self.session:
 132                         print "already running a session -> drop reqest"
 133                         return
 134
 135                 try:
 136                         bus = dbus.SystemBus()
 137
 138                         self.notify = Notification(bus, self, self.notify_path)
 139                         self.notify.add_to_connection(bus, self.notify_path)
 140
 141                         self.session_path = self.manager.CreateSession(self.settings, self.notify_path)
 142
 143                         print "notify path %s" % (self.notify_path)
 144                         print "session path %s" % (self.session_path)
 145
 146                         self.session = dbus.Interface(bus.get_object("net.connman", self.session_path),
 147                                                       "net.connman.Session")
 148
 149                 except dbus.DBusException, e:
 150                         if e.get_dbus_name() in ['net.connman.Error.Failed']:
 151                                 print e.get_dbus_message()
 152                                 return
 153                         traceback.print_exc()
 154                         exit(1)
 155
 156         @dbus.service.method("com.example.TestSession",
 157                                 in_signature='', out_signature='')
 158         def DestroySession(self):
 159                 print "Destroy session"
 160
 161                 if self.session == None:
 162                         print "no session running -> drop request"
 163                         return
 164
 165                 try:
 166                         self.release()
 167                 except dbus.DBusException:
 168                         traceback.print_exc()
 169                         exit(1)
 170
 171         @dbus.service.method("com.example.TestSession",
 172                                 in_signature='', out_signature='')
 173         def Connect(self):
 174                 print "Connect session"
 175
 176                 if self.session == None:
 177                         print "no session running -> drop request"
 178                         return
 179
 180                 try:
 181                         self.session.Connect()
 182                 except dbus.DBusException, e:
 183                         if e.get_dbus_name() in ['net.connman.Error.Failed']:
 184                                 print e.get_dbus_message()
 185                                 return
 186                         traceback.print_exc()
 187                         exit(1)
 188
 189         @dbus.service.method("com.example.TestSession",
 190                                 in_signature='', out_signature='')
 191         def Disconnect(self):
 192                 print "Disconnect session"
 193
 194                 if self.session == None:
 195                         print "no session running -> drop request"
 196                         return
 197
 198                 try:
 199                         self.session.Disconnect()
 200                 except dbus.DBusException, e:
 201                         if e.get_dbus_name() in ['net.connman.Error.Failed']:
 202                                 print e.get_dbus_message()
 203                                 return
 204                         traceback.print_exc()
 205                         exit(1)
 206
 207         @dbus.service.method("com.example.TestSession",
 208                                 in_signature='sv', out_signature='')
 209         def Change(self, key, value):
 210                 print "Update session settings"
 211
 212                 if self.session == None:
 213                         print "no session running -> drop request"
 214                         return
 215
 216                 try:
 217                         val = self.type_convert(key, value)
 218                         self.session.Change(key, val)
 219                 except dbus.DBusException, e:
 220                         if e.get_dbus_name() in ['net.connman.Error.Failed']:
 221                                 print e.get_dbus_message()
 222                                 return
 223                         traceback.print_exc()
 224                         exit(1)
 225
 226         @dbus.service.method("com.example.TestSession",
 227                                 in_signature='sv', out_signature='')
 228         def Configure(self, key, value):
 229                 print "Configure session settings"
 230                 val = self.type_convert(key, value)
 231                 self.settings[key] = val
 232
 233 def main():
 234         if len(sys.argv) < 2:
 235                 print "Usage: %s <command>" % (sys.argv[0])
 236                 print ""
 237                 print "  enable"
 238                 print "  disable"
 239                 print "  create <app_path>"
 240                 print "  destroy <app_path>"
 241                 print "  connect <app_path>"
 242                 print "  disconnect <app_path>"
 243                 print "  change <app_path> <key> <value>"
 244                 print "  configure <app_path> <key> <value>"
 245                 print ""
 246                 print "  run <app_path>"
 247                 sys.exit(1)
 248
 249         dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
 250
 251         if sys.argv[1] == "enable":
 252                 bus = dbus.SystemBus()
 253                 manager = dbus.Interface(bus.get_object("net.connman", "/"),
 254                                          "net.connman.Manager")
 255                 manager.SetProperty("SessionMode", True)
 256                 return
 257
 258         elif sys.argv[1] == "disable":
 259                 bus = dbus.SystemBus()
 260                 manager = dbus.Interface(bus.get_object("net.connman", "/"),
 261                                          "net.connman.Manager")
 262                 manager.SetProperty("SessionMode", False)
 263                 return
 264
 265         if (len(sys.argv) < 3):
 266                 print "Need test application path"
 267                 sys.exit(1)
 268
 269         app_path = sys.argv[2]
 270         bus = dbus.SessionBus()
 271
 272         app_name = "com.example.SessionApplication.%s" % (string.strip(app_path, "/"))
 273
 274         if sys.argv[1] == "run":
 275                 name = dbus.service.BusName(app_name, bus)
 276                 mainloop = gobject.MainLoop()
 277
 278                 app = SessionApplication(bus, app_path, mainloop)
 279
 280                 mainloop.run()
 281                 return
 282
 283         app = dbus.Interface(bus.get_object(app_name, app_path),
 284                              "com.example.TestSession")
 285
 286         if sys.argv[1] == "create":
 287                 app.CreateSession()
 288
 289         elif sys.argv[1] == "destroy":
 290                 app.DestroySession()
 291
 292         elif sys.argv[1] == "connect":
 293                 app.Connect()
 294
 295         elif sys.argv[1] == "disconnect":
 296                 app.Disconnect()
 297
 298         elif sys.argv[1] == "change":
 299                 if len(sys.argv) < 5:
 300                         print "Arguments missing"
 301                         sys.exit(1)
 302
 303                 app.Change(sys.argv[3], sys.argv[4:])
 304
 305         elif sys.argv[1] == "configure":
 306                 if len(sys.argv) < 5:
 307                         print "Arguments missing"
 308                         sys.exit(1)
 309
 310                 app.Configure(sys.argv[3], sys.argv[4:])
 311
 312         else:
 313                 print "Unknown command '%s'" % sys.argv[1]
 314                 sys.exit(1)
 315
 316 if __name__ == '__main__':
 317         main()