test/ipc_interface.py

   1 # Copyright(C) 2007,2008 Ixonos Plc
   2 #
   3 # This program is free software; you can redistribute it and/or modify
   4 # it under the terms of the GNU General Public License as published by
   5 # the Free Software Foundation; either version 2, or (at your option)
   6 # any later version.
   7 #
   8 # This program is distributed in the hope that it will be useful,
   9 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11 # GNU General Public License for more details.
  12 #
  13 # You should have received a copy of the GNU General Public License
  14 # along with this program; if not, write to the Free Software
  15 # Foundation, Inc., 59 Temple Place, Boston, MA 02111.
  16
  17 # Written by Matti Katila, Yi Zheng, Vesa Pikki and Heikki Paajanen
  18 from __future__ import with_statement
  19 import threading, time
  20
  21 class IPCInterface:
  22     def __init__ (self):
  23         self.debug = True
  24         self.debug_signal = True
  25         self.sync_lock = threading.Lock()
  26         self.signals = []
  27         self.reply_event = threading.Event()
  28         self.call_id = None
  29         self.message_index = None
  30         self.timeout_fails = True
  31         self.reply_message = None
  32
  33     def Initialize(self):
  34         self.Reset()
  35         self.call_id = None
  36         self.message_index = None
  37     def DeInitialize(self):
  38         pass
  39
  40     def Reset(self):
  41         with self.sync_lock:
  42             self.signals = []
  43         self.error = None
  44         self.reply = None
  45         self.reply_event.clear()
  46
  47     def getResultAndReset(self):
  48         self.reply_event.wait()
  49         if self.error:
  50             result = self.error
  51         else:
  52             result = self.reply
  53         self.Reset()
  54         return result
  55
  56     def getReply(self):
  57         print "Waiting for reply"
  58         self.reply_event.wait()
  59         return self.reply
  60     def getError(self):
  61         print "Waiting for error"
  62         self.reply_event.wait()
  63         return self.error
  64
  65     def ensure_value(self,name,value,expected_value=None):
  66         if type(expected_value) != tuple and expected_value != None:
  67             expected_value =(expected_value,)
  68
  69         assert value == expected_value,"Didn't get %(expected_value)s as a %(name)s, got %(value)s instead!" % locals()
  70
  71     def ensure_value_set(self,name,value,set):
  72         if set:
  73             assert value != None, "Value '%s' not set!" % name
  74         else:
  75             assert value == None, "Value '%s' is set!" % name
  76
  77     def ensure_error(self,value=None):
  78         #Do additional check on error reply
  79         self.CheckError()
  80         if self.debug:
  81             print "Ensuring error is ",value
  82         self.ensure_value("error",self.getError(),value)
  83
  84     def ensure_reply_initiate(self):
  85         self.ensure_reply_set(True)
  86         reply = self.getReply()
  87         assert type(reply) == tuple, "Reply %s is not tuple!" % str(reply)
  88         assert len(reply) == 1, "Reply %s has wrong number of elements!" % str(reply)
  89         self.call_id = reply[0]
  90         if self.debug:
  91             print "Initiate got call id %d" % self.call_id
  92
  93     def ensure_reply(self,value=None):
  94         if self.debug:
  95             print "Ensuring reply is ",value
  96         self.ensure_value("reply",self.getReply(),value)
  97
  98     def ensure_error_set(self,value=True):
  99         #Do additional check on error reply
 100         self.CheckError()
 101         if self.debug:
 102             if value == True:
 103                 print "Ensuring error is set"
 104             else:
 105                 print "Ensuring error is not set"
 106
 107         self.ensure_value_set("error",self.getError(),value)
 108
 109
 110     def ensure_reply_set(self,value=True):
 111         if self.debug:
 112             if value == True:
 113                 print "Ensuring reply is set"
 114             else:
 115                 print "Ensuring reply is not set"
 116         self.ensure_value_set("reply",self.getReply(),value)
 117
 118     def ReplyHandler(self,*arg):
 119         if len(arg) >= 1 and type(arg[0]) == tuple:
 120             arg = arg[0]
 121
 122         self.reply = arg
 123         if self.debug:
 124             if self.reply_message:
 125                 print self.reply_message,str(arg)
 126                 self.reply_message = None
 127             else:
 128                 print "Got reply: ",str(arg)
 129         self.reply_event.set()
 130
 131     #Used to do ipc related checks (for example dbus timeout)
 132     def CheckError(self):
 133         pass
 134
 135     def ErrorHandler(self,*arg):
 136         if len(arg) >= 1 and type(arg[0]) == tuple:
 137             arg = arg[0]
 138
 139         self.error = arg
 140         if self.debug:
 141             print "Got error: ",arg
 142         if self.reply_message:
 143             self.reply_message = None
 144         self.reply_event.set()
 145
 146     def AddSignal(self,signal):
 147         with self.sync_lock:
 148             if self.debug_signal:
 149                 print "Received signal: ",signal
 150             self.signals.append(signal)
 151
 152     def CheckSignal(self, tup):
 153         print "Checking signal",tup
 154         t0 = time.time()
 155         while time.time() - t0 < 5:
 156             with self.sync_lock:
 157                 if len(self.signals) == 0:
 158                     pass
 159                 else:
 160                     if (tup in self.signals):
 161                         self.signals.remove(tup)
 162                         return True
 163                     else:
 164                         if self.debug:
 165                             print "Previous signals: " + str (self.signals) + "\n"
 166                             print "Tup is: "+ str (tup) + "\n"
 167             time.sleep(0.2)
 168         raise Exception, 'time out - did not receive signal '+str(tup)
 169
 170     def WaitSignalValues(self, *arg):
 171         print "Waiting for signal with values '%s'" % str(arg)
 172         t0 = time.time()
 173         while time.time() - t0 < 5:
 174             with self.sync_lock:
 175                 if len(self.signals) == 0:
 176                     pass
 177                 else:
 178                     def filter_func(tup):
 179                         i = 0
 180                         for a in arg:
 181                             if tup[i] != a:
 182                                 return False
 183                             i+=1
 184
 185                         return True
 186                     filtered = filter(filter_func, self.signals)
 187                     if len(filtered) > 0:
 188                         tup = filtered[0]
 189                         self.signals.remove(tup)
 190                         return tup
 191                     else:
 192                         if self.debug:
 193                             print "Previous signals: " + str (self.signals) + "\n"
 194             time.sleep(0.2)
 195         raise Exception, 'time out - did not receive signal with these arguments'+str(arg)
 196
 197     #Call Interface
 198     def Call_Emergency(self):
 199         raise Exception("Not implemented")
 200     def Call_Activate(self, id):
 201         raise Exception("Not implemented")
 202     def Call_ActivateConference(self, id):
 203         raise Exception("Not implemented")
 204     def Call_Release(self,message,id):
 205         raise Exception("Not implemented")
 206     def Call_ReleaseHeld(self,message):
 207         raise Exception("Not implemented")
 208     def Call_ReleaseAll(self,message):
 209         raise Exception("Not implemented")
 210     def Call_Initiate(self,number,type):
 211         raise Exception("Not implemented")
 212     def Call_ListCalls(self):
 213         raise Exception("Not implemented")
 214     def Call_GetStatus(self,id):
 215         raise Exception("Not implemented")
 216     def Call_SendDtmf(self):
 217         raise Exception("Not implemented")
 218     def Call_SetDtmfDuration(self,mode):
 219         raise Exception("Not implemented")
 220     def Call_GetDtmfDuration(self):
 221         raise Exception("Not implemented")
 222     def Call_SetSendIdentification(self,mode):
 223         raise Exception("Not implemented")
 224     def Call_GetSendIdentification(self):
 225         raise Exception("Not implemented")
 226
 227
 228     #Device interface
 229     def Device_GetInfo(self):
 230         raise Exception("Not implemented")
 231     def Device_SetAntennaPower(self,antenna_power):
 232         raise Exception("Not implemented")
 233     def Device_GetAntennaPower(self):
 234         raise Exception("Not implemented")
 235     def Device_GetFeatures(self):
 236         raise Exception("Not implemented")
 237
 238     #Network interface
 239     def Network_Register(self):
 240         raise Exception("Not implemented")
 241     def Network_GetStatus(self):
 242         raise Exception("Not implemented")
 243     def Network_ListProviders(self):
 244         raise Exception("Not implemented")
 245     def Network_RegisterWithProvider(self,index):
 246         raise Exception("Not implemented")
 247     def Network_GetSubscriberNumbers(self):
 248         raise Exception("Not implemented")
 249     def Network_GetCountryCode(self):
 250         raise Exception("Not implemented")
 251     def Network_GetHomeCountryCode(self):
 252         raise Exception("Not implemented")
 253
 254     #SMS interface
 255     def SMS_SendMessage(self,message,number,want_report):
 256         raise Exception("Not implemented")
 257
 258     #SIM interface
 259     def SIM_GetAuthStatus(self):
 260         raise Exception("Not implemented")
 261     def SIM_SendAuthCode(self,pin):
 262         raise Exception("Not implemented")
 263     def SIM_Unlock(self,puk,new_pin):
 264         raise Exception("Not implemented")
 265     def SIM_ChangeAuthCode(self,old_pin,new_pin):
 266         raise Exception("Not implemented")
 267     def SIM_GetSimInfo(self):
 268         raise Exception("Not implemented")
 269     def SIM_SendMessage(self,index,want_report):
 270         raise Exception("Not implemented")
 271     def SIM_ListMessages(self,status):
 272         raise Exception("Not implemented")
 273     def SIM_RetrieveMessage(self,index):
 274         raise Exception("Not implemented")
 275     def SIM_DeleteMessage(self,index):
 276         raise Exception("Not implemented")
 277     def SIM_StoreMessage(self,number,message):
 278         raise Exception("Not implemented")
 279     def SIM_GetPhonebookInfo(self):
 280         raise Exception("Not implemented")
 281     def SIM_DeleteEntry(self,index):
 282         raise Exception("Not implemented")
 283     def SIM_StoreEntry(self,index,name,number):
 284         raise Exception("Not implemented")
 285     def SIM_RetrieveEntry(self,index):
 286         raise Exception("Not implemented")
 287
 288
 289     #Phonebook interface
 290
 291     #PDP interface
 292     def PDP_ListGprsClasses(self):
 293         raise Exception("Not implemented")
 294     def PDP_SelectGprsClass(self,klass):
 295         raise Exception("Not implemented")
 296     def PDP_Activate(self,index):
 297         raise Exception("Not implemented")
 298     def PDP_Deactivate(self,index):
 299         raise Exception("Not implemented")
 300     def PDP_SelectContext(self,context):
 301         raise Exception("Not implemented")
 302     def PDP_AddContext(self,context):
 303         raise Exception("Not implemented")
 304     def PDP_DeleteContext(self,index):
 305         raise Exception("Not implemented")
 306     def PDP_ListContexts(self):
 307         raise Exception("Not implemented")