1#!/usr/bin/python3
2
3#%%
4
5import dbus
6import dbus.service
7import dbus.mainloop.glib
8import time
9import unittest
10
11HCI_DEVICES_DIR = "/sys/class/bluetooth"
12
13# File to store the Bluetooth daemon to use (bluez or floss)
14BLUETOOTH_DAEMON_CURRENT = "/var/lib/misc/bluetooth-daemon.current"
15
16# File to store the config for BluetoothManager
17BTMANAGERD_CONF = "/var/lib/bluetooth/btmanagerd.json"
18
19# D-BUS bus name
20BUS_NAME = "org.chromium.bluetooth.Manager"
21
22# D-Bus Bluetooth Manager object path
23MANAGER_OBJECT_PATH = "/org/chromium/bluetooth/Manager"
24
25# D-Bus Bluetooth Manager interface name
26MANAGER_INTERFACE_NAME = "org.chromium.bluetooth.Manager"
27
28# D-Bus Bluetooth Manager client (this test) object path
29CLIENT_OBJECT_PATH = "/test_client"
30
31
32def use_floss():
33  with open(BLUETOOTH_DAEMON_CURRENT, "w") as f:
34    f.write("floss")
35
36def use_bluez():
37  with open(BLUETOOTH_DAEMON_CURRENT, "w") as f:
38    f.write("floss")
39
40class BtmanagerdTest(unittest.TestCase):
41
42    def setUp(self):
43      self.bus = dbus.SystemBus()
44      self.manager_object = self.bus.get_object(BUS_NAME, MANAGER_OBJECT_PATH)
45      self.client_object = self.bus.get_object(BUS_NAME, CLIENT_OBJECT_PATH)
46      self.manager_object.Stop(0, dbus_interface=MANAGER_INTERFACE_NAME)
47      time.sleep(2.5)
48      self.assertEqual(self._get_state(), 0)
49
50    def _start_hci(self, hci=0):
51      self.manager_object.Start(hci, dbus_interface=MANAGER_INTERFACE_NAME)
52      time.sleep(2.5)
53
54    def _stop_hci(self, hci=0):
55      self.manager_object.Stop(hci, dbus_interface=MANAGER_INTERFACE_NAME)
56      time.sleep(2.5)
57
58    def _get_state(self):
59      return self.manager_object.GetState(dbus_interface=MANAGER_INTERFACE_NAME)
60
61    def _register_hci_device_change_observer(self):
62      return self.manager_object.RegisterStateChangeObserver(self.client_object_path, dbus_interface=MANAGER_INTERFACE_NAME)
63
64    def _unregister_hci_device_change_observer(self):
65      return self.manager_object.UnregisterHciDeviceChangeObserver(self.client_object_path, dbus_interface=MANAGER_INTERFACE_NAME)
66
67    def _get_floss_enabled(self):
68      return self.manager_object.GetFlossEnabled(dbus_interface=MANAGER_INTERFACE_NAME)
69
70    def _set_floss_enabled(self, enabled=True):
71      return self.manager_object.SetFlossEnabled(enabled, dbus_interface=MANAGER_INTERFACE_NAME)
72
73    def _list_hci_devices(self):
74      return self.manager_object.ListHciDevices(dbus_interface=MANAGER_INTERFACE_NAME)
75
76    def _register_hci_device_change_observer(self):
77      return self.manager_object.RegisterHciDeviceChangeObserver(self.client_object_path, dbus_interface=MANAGER_INTERFACE_NAME)
78
79    def _unregister_hci_device_change_observer(self):
80      return self.manager_object.UnregisterHciDeviceChangeObserver(self.client_object_path, dbus_interface=MANAGER_INTERFACE_NAME)
81
82    def test_list_hci_devices(self):
83      self.assertTrue(len(self._list_hci_devices()) > 0)
84
85    def test_floss_enabled(self):
86      self.assertTrue(self._get_floss_enabled())
87
88    def test_disable_floss(self):
89      self._set_floss_enabled(False)
90      self.assertFalse(self._get_floss_enabled())
91      with open(BLUETOOTH_DAEMON_CURRENT, "r") as f:
92        self.assertEqual(f.read(), "bluez")
93
94      self._set_floss_enabled(True)
95      self.assertTrue(self._get_floss_enabled())
96
97    def test_start(self):
98      self._start_hci()
99      self.assertEqual(self._get_state(), 2)
100
101    def test_astart_and_start(self):
102      self._start_hci()
103      self._start_hci()
104      self.assertEqual(self._get_state(), 2)
105
106    def test_stop(self):
107      self._start_hci()
108      self._stop_hci()
109      self.assertEqual(self._get_state(), 0)
110
111    def test_stop_and_stop(self):
112      self._start_hci()
113      self._stop_hci()
114      self._stop_hci()
115      self.assertEqual(self._get_state(), 0)
116
117
118# %%
119if __name__ == '__main__':
120  dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
121  use_floss()
122  unittest.main()
123