• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1"""Tests for blueberry.map.bluetooth_map."""
2from __future__ import absolute_import
3from __future__ import division
4from __future__ import print_function
5
6import queue
7import time
8
9from mobly import test_runner
10from mobly import signals
11from mobly import utils
12
13from blueberry.controllers import android_bt_target_device
14from blueberry.utils import blueberry_base_test
15
16_SMS_MSG_EVENT = 'SmsReceived'
17_MAP_MSG_EVENT = 'MapMessageReceived'
18
19_EVENT_TIMEOUT_SEC = 180
20_TEXT_LENGTH = 10
21_TEXT_COUNT = 5
22
23
24class BluetoothMapTest(blueberry_base_test.BlueberryBaseTest):
25  """Test Class for Bluetooth MAP Test."""
26
27  def __init__(self, configs):
28    super().__init__(configs)
29    self.derived_bt_device = None
30    self.pri_phone = None
31    self.pri_number = None
32    self.sec_phone = None
33
34  def setup_class(self):
35    """Standard Mobly setup class."""
36    super().setup_class()
37    for device in self.android_devices:
38      device.init_setup()
39      device.sl4a_setup()
40
41    # Primary phone which role is Message Server Equipment (MSE).
42    self.pri_phone = self.android_devices[0]
43    self.pri_phone.sl4a.smsStartTrackingIncomingSmsMessage()
44    self.pri_number = self.pri_phone.dimensions['phone_number']
45
46    # Secondary phone which is used to send SMS messages to primary phone.
47    self.sec_phone = self.android_devices[1]
48
49    # Bluetooth carkit which role is Message Client Equipment (MCE).
50    self.derived_bt_device = self.derived_bt_devices[0]
51
52    mac_address = self.derived_bt_device.get_bluetooth_mac_address()
53    self.derived_bt_device.activate_pairing_mode()
54    self.pri_phone.pair_and_connect_bluetooth(mac_address)
55    # Sleep to make the connection to be steady.
56    time.sleep(5)
57
58    if isinstance(
59        self.derived_bt_device, android_bt_target_device.AndroidBtTargetDevice):
60      # Allow sl4a to receive the intent with ACTION_MESSAGE_RECEIVED.
61      self.derived_bt_device.adb.shell(
62          'pm grant com.googlecode.android_scripting '
63          'android.permission.RECEIVE_SMS')
64      # Connect derived bt device to primary phone via MAP MCE profile.
65      self.derived_bt_device.add_sec_ad_device(self.pri_phone)
66
67  def teardown_test(self):
68    """Standard Mobly teardown test.
69
70    Disconnects the MAP connection after a test completes.
71    """
72    super().teardown_test()
73    self.derived_bt_device.map_disconnect()
74
75  def _wait_for_message_on_mce(self, text):
76    """Waits for that MCE gets an event with specific message.
77
78    Args:
79      text: String, Text of the message.
80
81    Raises:
82      TestFailure: Raised if timed out.
83    """
84    try:
85      self.derived_bt_device.ed.wait_for_event(
86          _MAP_MSG_EVENT, lambda e: e['data'] == text, _EVENT_TIMEOUT_SEC)
87      self.derived_bt_device.log.info(
88          'Successfully got the unread message: %s' % text)
89    except queue.Empty:
90      raise signals.TestFailure(
91          'Timed out after %ds waiting for "%s" event with the message: %s' %
92          (_EVENT_TIMEOUT_SEC, _MAP_MSG_EVENT, text))
93
94  def _wait_for_message_on_mse(self, text):
95    """Waits for that MSE gets an event with specific message.
96
97    This method is used to make sure that MSE has received the test message.
98
99    Args:
100      text: String, Text of the message.
101
102    Raises:
103      TestError: Raised if timed out.
104    """
105    try:
106      self.pri_phone.ed.wait_for_event(
107          _SMS_MSG_EVENT, lambda e: e['data']['Text'] == text,
108          _EVENT_TIMEOUT_SEC)
109      self.pri_phone.log.info(
110          'Successfully received the incoming message: %s' % text)
111    except queue.Empty:
112      raise signals.TestError(
113          'Timed out after %ds waiting for "%s" event with the message: %s' %
114          (_EVENT_TIMEOUT_SEC, _SMS_MSG_EVENT, text))
115
116  def _create_message_on_mse(self, text):
117    """Creates a new incoming message on MSE.
118
119    Args:
120      text: String, Text of the message.
121    """
122    self.sec_phone.sl4a.smsSendTextMessage(self.pri_number, text, False)
123    self._wait_for_message_on_mse(text)
124
125  def test_get_existing_unread_messages(self):
126    """Test for the feature of getting existing unread messages on MCE.
127
128    Tests MCE can list existing messages of MSE.
129    """
130    text_list = []
131    # Creates 5 SMS messages on MSE before establishing connection.
132    for _ in range(_TEXT_COUNT):
133      text = utils.rand_ascii_str(_TEXT_LENGTH)
134      self._create_message_on_mse(text)
135      text_list.append(text)
136    self.derived_bt_device.map_connect()
137    # Gets the unread messages of MSE and checks if they are downloaded
138    # successfully on MCE.
139    self.derived_bt_device.get_unread_messages()
140    for text in text_list:
141      self._wait_for_message_on_mce(text)
142
143  def test_receive_unread_message(self):
144    """Test for the feature of receiving unread message on MCE.
145
146    Tests MCE can get an unread message when MSE receives an incoming message.
147    """
148    self.derived_bt_device.map_connect()
149    text = utils.rand_ascii_str(_TEXT_LENGTH)
150    self._create_message_on_mse(text)
151    self._wait_for_message_on_mce(text)
152
153
154if __name__ == '__main__':
155  test_runner.main()
156