test/test-session

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