mirror of
https://github.com/undera/pylgbst.git
synced 2020-11-18 19:37:26 -08:00
dc1c388fe8
* Added support for bluepy communication backend. * Added bluepy information into the readme. * Added tests, fixed dependency specs in setup.py. * Fixed dep in travis. * Removed unused import. Added ability to fail the application on dispatcher thread error. * Fixed bluepy test to be more appropriate. * Properly handle hub mac if set.
62 lines
1.8 KiB
Python
62 lines
1.8 KiB
Python
import unittest
|
|
import time
|
|
|
|
import pylgbst.comms.cbluepy as bp_backend
|
|
|
|
|
|
class PeripheralMock(object):
|
|
def __init__(self, addr, addrType, ifaceNumber):
|
|
pass
|
|
|
|
def waitForNotifications(self, timeout):
|
|
pass
|
|
|
|
def writeCharacteristic(self, handle, data):
|
|
pass
|
|
|
|
def withDelegate(self, delegate):
|
|
pass
|
|
|
|
def disconnect(self):
|
|
pass
|
|
|
|
bp_backend.PROPAGATE_DISPATCHER_EXCEPTION = True
|
|
bp_backend.btle.Peripheral = lambda *args, **kwargs: PeripheralMock(*args, **kwargs)
|
|
|
|
|
|
class BluepyTestCase(unittest.TestCase):
|
|
def test_get_iface_number(self):
|
|
self.assertEqual(bp_backend._get_iface_number('hci0'), 0)
|
|
self.assertEqual(bp_backend._get_iface_number('hci10'), 10)
|
|
try:
|
|
bp_backend._get_iface_number('ads12')
|
|
self.fail('Missing exception for incorrect value')
|
|
except ValueError:
|
|
pass
|
|
|
|
def test_delegate(self):
|
|
def _handler(handle, data):
|
|
_handler.called = True
|
|
delegate = bp_backend.BluepyDelegate(_handler)
|
|
delegate.handleNotification(123, 'qwe')
|
|
self.assertEqual(_handler.called, True)
|
|
|
|
def test_threaded_peripheral(self):
|
|
tp = bp_backend.BluepyThreadedPeripheral('address', 'addrType', 'hci0')
|
|
self.assertEqual(tp._addr, 'address')
|
|
self.assertEqual(tp._addrType, 'addrType')
|
|
self.assertEqual(tp._iface_number, 0)
|
|
self.assertNotEqual(tp._dispatcher_thread, None)
|
|
|
|
# Schedule some methods to async queue and give them some time to resolve
|
|
tp.set_notify_handler(lambda: '')
|
|
tp.write(123, 'qwe')
|
|
|
|
tp._dispatcher_thread.join(1)
|
|
self.assertEqual(tp._dispatcher_thread.is_alive(), True)
|
|
tp.disconnect()
|
|
|
|
tp._dispatcher_thread.join(2)
|
|
self.assertEqual(tp._dispatcher_thread.is_alive(), False)
|
|
|