| # Copyright (c) 2011 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import logging, os, subprocess |
| import dbus, dbus.mainloop.glib, gobject |
| |
| from autotest_lib.client.bin import test |
| from autotest_lib.client.common_lib import error |
| from autotest_lib.client.cros.cellular import sms, mmtest |
| from autotest_lib.client.cros.mainloop import GenericTesterMainLoop |
| from autotest_lib.client.cros.mainloop import ExceptionForward |
| |
| class SmsTester(GenericTesterMainLoop): |
| def __init__(self, autodir, srcdir, mmt, *args, **kwargs): |
| self.autodir = autodir |
| self.srcdir = srcdir |
| self.mmt = mmt |
| self.remaining_requirements = ['Received', 'Completed'] |
| super(SmsTester, self).__init__(*args, timeout_s = 10, **kwargs) |
| |
| # The GenericTesterMainLoop will run this routine from the idle |
| # loop. In a successful test, the two SMS signals will be |
| # recieved by the routines registered in the main test class and |
| # will call in to the methods below. Order of reception is not |
| # important. |
| @ExceptionForward |
| def perform_one_test(self): |
| self.gsmsms = self.mmt.mm.GsmSms(self.mmt.modem_object_path) |
| self.smstest = sms.SmsTest(self.gsmsms) |
| self.smsstore = sms.SmsStore(self.mmt.fakemodem) |
| # Actual test |
| self.smstest.test_has_none() |
| self.testsms = sms.sample |
| self.smsstore.sms_receive(1, self.testsms['pdu']) |
| |
| @ExceptionForward |
| def SmsReceived(self, index, complete): |
| if index != 1: |
| raise error.TestFail("Wrong index %d != 1" % index) |
| if complete == False: |
| raise error.TestFail("Message not complete") |
| self.requirement_completed('Received') |
| |
| @ExceptionForward |
| def SmsCompleted(self, index, complete): |
| if index != 1: |
| raise error.TestFail("Wrong index %d != 1" % index) |
| if complete == False: |
| raise error.TestFail("Message not complete") |
| self.smstest.test_has_one(self.testsms['parsed']) |
| self.smsstore.sms_remove(1) |
| self.smstest.test_has_none() |
| self.requirement_completed('Completed') |
| |
| |
| class SmsMultipartTester(GenericTesterMainLoop): |
| def __init__(self, autodir, srcdir, mmt, *args, **kwargs): |
| self.autodir = autodir |
| self.srcdir = srcdir |
| self.mmt = mmt |
| self.remaining_requirements = ['Received', 'Received', 'Completed'] |
| super(SmsMultipartTester, self).__init__(*args, timeout_s = 10, |
| **kwargs) |
| |
| # The GenericTesterMainLoop will run this routine from the idle |
| # loop. In a successful test, the first SMSReceived signal will |
| # be recieved by the routine registered in the main test class and |
| # will call SmsReceived below; that routine will send the second |
| # part of the message, and the two resulting signals will call |
| # SmsReceived and SmsCompleted. |
| @ExceptionForward |
| def perform_one_test(self): |
| self.gsmsms = self.mmt.mm.GsmSms(self.mmt.modem_object_path) |
| self.smstest = sms.SmsTest(self.gsmsms) |
| self.smsstore = sms.SmsStore(self.mmt.fakemodem) |
| # Actual test |
| self.smstest.test_has_none() |
| self.testsms = sms.sample_multipart |
| self.smsstore.sms_receive(1, self.testsms['pdu'][0]) |
| self.second = False |
| |
| @ExceptionForward |
| def SmsReceived(self, index, complete): |
| logging.info("Received, index %d"%index) |
| if index != 1: |
| raise error.TestFail("Wrong index %d != 1" % index) |
| if complete != self.second: |
| raise error.TestFail("Complete is wrong, should be %s" % |
| self.second) |
| self.requirement_completed('Received') |
| if self.second == False: |
| self.smsstore.sms_receive(2, self.testsms['pdu'][1]) |
| self.second = True |
| |
| @ExceptionForward |
| def SmsCompleted(self, index, complete): |
| logging.info("Completed, index %d"%index) |
| if index != 1: |
| raise error.TestFail("Wrong index %d != 1" % index) |
| if complete == False: |
| raise error.TestFail("Message not complete") |
| self.smstest.test_has_one(self.testsms['parsed']) |
| self.smsstore.sms_remove(1) |
| self.smsstore.sms_remove(2) |
| self.smstest.test_has_none() |
| self.requirement_completed('Completed') |
| |
| |
| class network_ModemManagerSMSSignal(test.test): |
| version = 1 |
| |
| def setup(self): |
| self.job.setup_dep(['fakegudev', 'fakemodem']) |
| |
| @ExceptionForward |
| def SmsReceived(self, *args, **kwargs): |
| self.smstester.SmsReceived(*args, **kwargs) |
| |
| @ExceptionForward |
| def SmsCompleted(self, *args, **kwargs): |
| self.smstester.SmsCompleted(*args, **kwargs) |
| |
| def run_once(self, **kwargs): |
| self.job.install_pkg('fakegudev', 'dep', |
| os.path.join(self.autodir, 'deps', 'fakegudev')) |
| self.job.install_pkg('fakemodem', 'dep', |
| os.path.join(self.autodir, 'deps', 'fakemodem')) |
| subprocess.check_call(["modprobe", "tun"]) |
| subprocess.check_call(["initctl", "stop", "modemmanager"]) |
| |
| dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) |
| self.main_loop = gobject.MainLoop() |
| self.bus = dbus.SystemBus() |
| self.bus.add_signal_receiver(self.SmsReceived, |
| signal_name='SmsReceived') |
| self.bus.add_signal_receiver(self.SmsCompleted, |
| signal_name='Completed') |
| |
| try: |
| paths = [os.path.join(self.srcdir, 'fake-gsm'), |
| os.path.join(self.srcdir, 'fake-icera')] |
| with mmtest.ModemManagerTest(self.autodir, paths) as mmt: |
| self.smstester = SmsTester(self.autodir, self.srcdir, |
| mmt, self, self.main_loop) |
| self.smstester.run(**kwargs) |
| |
| with mmtest.ModemManagerTest(self.autodir, paths) as mmt: |
| self.smstester = SmsMultipartTester(self.autodir, self.srcdir, |
| mmt, self, self.main_loop) |
| self.smstester.run(**kwargs) |
| |
| finally: |
| subprocess.check_call(["initctl", "start", "modemmanager"]) |
| subprocess.check_call(["rmmod", "tun"]) |
