• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# Lint as: python3
3"""
4Base class for setting up devices for CDM functionalities.
5"""
6
7from mobly import base_test
8from mobly import utils
9from mobly.controllers import android_device
10from test_utils import wait
11from time import sleep
12
13CDM_SNIPPET_PACKAGE = 'android.companion.cts.multidevice'
14
15BT_DISCOVERABLE_TIME = 15
16OPERATION_DELAY_TIME = 5
17
18def paired_devices(self):
19    return map(lambda device: device['Address'], self.cdm.btGetPairedDevices())
20
21class BaseTestClass(base_test.BaseTestClass):
22
23    def setup_class(self):
24        android_device.AndroidDevice.paired_devices = paired_devices
25
26        # Declare that two Android devices are needed.
27        self.primary, self.secondary = self.register_controller(
28            android_device, min_number=2)
29
30        def _setup_device(device):
31            device.load_snippet('cdm', CDM_SNIPPET_PACKAGE)
32            # Enable bluetooth and enable receivers
33            device.cdm.btEnable()
34            device.address = device.cdm.btGetAddress()
35
36            # Clean up existing associations
37            device.cdm.disassociateAll()
38
39        self._execute_on_devices(_setup_device)
40
41
42    def setup_test(self):
43
44        def _setup_device(device):
45            # Touch the screen to make sure the device is wake up for each test run.
46            device.adb.shell('input keyevent KEYCODE_WAKEUP')
47            device.adb.shell('input keyevent KEYCODE_MENU')
48            device.adb.shell('input keyevent KEYCODE_HOME')
49
50        self._execute_on_devices(_setup_device)
51
52
53    def teardown_test(self):
54        """Clean up tests"""
55        self.primary.cdm.disassociateAll()
56        self.secondary.cdm.disassociateAll()
57
58        def _teardown_device(device):
59            # Remove all associations
60            device.cdm.disassociateAll()
61
62        self._execute_on_devices(_teardown_device)
63        self.bt_unpair_devices()
64
65
66    def bt_pair_devices(self):
67        """Pair two devices using BT classic bond"""
68        # Wait until devices are fully un-paired. If not, skip pairing.
69        if wait(lambda: self.secondary.address not in self.primary.paired_devices()):
70            self.secondary.cdm.btBecomeDiscoverable(BT_DISCOVERABLE_TIME)
71            self.secondary.cdm.btStartAutoAcceptIncomingPairRequest()
72            self.primary.cdm.btDiscoverAndGetResults()
73            self.primary.cdm.btPairDevice(self.secondary.address)
74            wait(lambda: self.secondary.address in self.primary.paired_devices())
75
76
77    def bt_unpair_devices(self):
78        """Unpair two devices connected with BT classic bond."""
79        # Unpair only if already paired
80        if wait(lambda: self.secondary.address in self.primary.paired_devices()):
81            try:
82                self.primary.cdm.btUnpairDevice(self.secondary.address)
83                wait(lambda: self.secondary.address not in self.primary.paired_devices())
84            except:
85                pass
86
87        # Also clean up the other device just in case. This significantly reduces flakes.
88        if wait(lambda: self.primary.address in self.secondary.paired_devices()):
89            try:
90                self.secondary.cdm.btUnpairDevice(self.primary.address)
91                wait(lambda: self.primary.address not in self.secondary.paired_devices())
92            except:
93                pass
94
95
96    def _execute_on_devices(self, func, raise_on_exception=True):
97        # Executes a function on both primary and secondary devices concurrently to same time
98        utils.concurrent_exec(
99            func,
100            ((self.primary,), (self.secondary,)),
101            max_workers=2,
102            raise_on_exception=raise_on_exception)
103
104