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
5import logging, os, subprocess
6import dbus, dbus.mainloop.glib, gobject
7
8from autotest_lib.client.bin import test
9from autotest_lib.client.common_lib import error
10from autotest_lib.client.cros.cellular import sms, mmtest
11from autotest_lib.client.cros.mainloop import GenericTesterMainLoop
12from autotest_lib.client.cros.mainloop import ExceptionForward
13
14class 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
57class 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
110class 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