Home | History | Annotate | Download | only in network_ModemManagerSMSSignal 
      1 # Copyright (c) 2011 The Chromium OS Authors. All rights reserved.
      2 # Use of this source code is governed by a BSD-style license that can be
      3 # found in the LICENSE file.
      4 
      5 import logging, os, subprocess
      6 import dbus, dbus.mainloop.glib, gobject
      7 
      8 from autotest_lib.client.bin import test
      9 from autotest_lib.client.common_lib import error
     10 from autotest_lib.client.cros.cellular import sms, mmtest
     11 from autotest_lib.client.cros.mainloop import GenericTesterMainLoop
     12 from autotest_lib.client.cros.mainloop import ExceptionForward
     13 
     14 class SmsTester(GenericTesterMainLoop):
     15     def __init__(self, autodir, srcdir, mmt, *args, **kwargs):
     16         self.autodir = autodir
     17         self.srcdir = srcdir
     18         self.mmt = mmt
     19         self.remaining_requirements = ['Received', 'Completed']
     20         super(SmsTester, self).__init__(*args, timeout_s = 10, **kwargs)
     21 
     22     # The GenericTesterMainLoop will run this routine from the idle
     23     # loop.  In a successful test, the two SMS signals will be
     24     # recieved by the routines registered in the main test class and
     25     # will call in to the methods below. Order of reception is not
     26     # important.
     27     @ExceptionForward
     28     def perform_one_test(self):
     29         self.gsmsms = self.mmt.mm.GsmSms(self.mmt.modem_object_path)
     30         self.smstest = sms.SmsTest(self.gsmsms)
     31         self.smsstore = sms.SmsStore(self.mmt.fakemodem)
     32         # Actual test
     33         self.smstest.test_has_none()
     34         self.testsms = sms.sample
     35         self.smsstore.sms_receive(1, self.testsms['pdu'])
     36 
     37     @ExceptionForward
     38     def SmsReceived(self, index, complete):
     39         if index != 1:
     40             raise error.TestFail("Wrong index %d != 1" % index)
     41         if complete == False:
     42             raise error.TestFail("Message not complete")
     43         self.requirement_completed('Received')
     44 
     45     @ExceptionForward
     46     def SmsCompleted(self, index, complete):
     47         if index != 1:
     48             raise error.TestFail("Wrong index %d != 1" % index)
     49         if complete == False:
     50             raise error.TestFail("Message not complete")
     51         self.smstest.test_has_one(self.testsms['parsed'])
     52         self.smsstore.sms_remove(1)
     53         self.smstest.test_has_none()
     54         self.requirement_completed('Completed')
     55 
     56 
     57 class SmsMultipartTester(GenericTesterMainLoop):
     58     def __init__(self, autodir, srcdir, mmt, *args, **kwargs):
     59         self.autodir = autodir
     60         self.srcdir = srcdir
     61         self.mmt = mmt
     62         self.remaining_requirements = ['Received', 'Received', 'Completed']
     63         super(SmsMultipartTester, self).__init__(*args, timeout_s = 10,
     64                                                   **kwargs)
     65 
     66     # The GenericTesterMainLoop will run this routine from the idle
     67     # loop.  In a successful test, the first SMSReceived signal will
     68     # be recieved by the routine registered in the main test class and
     69     # will call SmsReceived below; that routine will send the second
     70     # part of the message, and the two resulting signals will call
     71     # SmsReceived and SmsCompleted.
     72     @ExceptionForward
     73     def perform_one_test(self):
     74         self.gsmsms = self.mmt.mm.GsmSms(self.mmt.modem_object_path)
     75         self.smstest = sms.SmsTest(self.gsmsms)
     76         self.smsstore = sms.SmsStore(self.mmt.fakemodem)
     77         # Actual test
     78         self.smstest.test_has_none()
     79         self.testsms = sms.sample_multipart
     80         self.smsstore.sms_receive(1, self.testsms['pdu'][0])
     81         self.second = False
     82 
     83     @ExceptionForward
     84     def SmsReceived(self, index, complete):
     85         logging.info("Received, index %d"%index)
     86         if index != 1:
     87             raise error.TestFail("Wrong index %d != 1" % index)
     88         if complete != self.second:
     89             raise error.TestFail("Complete is wrong, should be %s" %
     90                                  self.second)
     91         self.requirement_completed('Received')
     92         if self.second == False:
     93             self.smsstore.sms_receive(2, self.testsms['pdu'][1])
     94             self.second = True
     95 
     96     @ExceptionForward
     97     def SmsCompleted(self, index, complete):
     98         logging.info("Completed, index %d"%index)
     99         if index != 1:
    100             raise error.TestFail("Wrong index %d != 1" % index)
    101         if complete == False:
    102             raise error.TestFail("Message not complete")
    103         self.smstest.test_has_one(self.testsms['parsed'])
    104         self.smsstore.sms_remove(1)
    105         self.smsstore.sms_remove(2)
    106         self.smstest.test_has_none()
    107         self.requirement_completed('Completed')
    108 
    109 
    110 class network_ModemManagerSMSSignal(test.test):
    111     version = 1
    112 
    113     def setup(self):
    114         self.job.setup_dep(['fakegudev', 'fakemodem'])
    115 
    116     @ExceptionForward
    117     def SmsReceived(self, *args, **kwargs):
    118         self.smstester.SmsReceived(*args, **kwargs)
    119 
    120     @ExceptionForward
    121     def SmsCompleted(self, *args, **kwargs):
    122         self.smstester.SmsCompleted(*args, **kwargs)
    123 
    124     def run_once(self, **kwargs):
    125         self.job.install_pkg('fakegudev', 'dep',
    126                              os.path.join(self.autodir, 'deps', 'fakegudev'))
    127         self.job.install_pkg('fakemodem', 'dep',
    128                              os.path.join(self.autodir, 'deps', 'fakemodem'))
    129         subprocess.check_call(["modprobe", "tun"])
    130         subprocess.check_call(["initctl", "stop", "modemmanager"])
    131 
    132         dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
    133         self.main_loop = gobject.MainLoop()
    134         self.bus = dbus.SystemBus()
    135         self.bus.add_signal_receiver(self.SmsReceived,
    136                                      signal_name='SmsReceived')
    137         self.bus.add_signal_receiver(self.SmsCompleted,
    138                                      signal_name='Completed')
    139 
    140         try:
    141             paths = [os.path.join(self.srcdir, 'fake-gsm'),
    142                      os.path.join(self.srcdir, 'fake-icera')]
    143             with mmtest.ModemManagerTest(self.autodir, paths) as mmt:
    144                 self.smstester = SmsTester(self.autodir, self.srcdir,
    145                                            mmt, self, self.main_loop)
    146                 self.smstester.run(**kwargs)
    147 
    148             with mmtest.ModemManagerTest(self.autodir, paths) as mmt:
    149                 self.smstester = SmsMultipartTester(self.autodir, self.srcdir,
    150                                                     mmt, self, self.main_loop)
    151                 self.smstester.run(**kwargs)
    152 
    153         finally:
    154             subprocess.check_call(["initctl", "start", "modemmanager"])
    155             subprocess.check_call(["rmmod", "tun"])
    156