mirror of
https://github.com/undera/pylgbst.git
synced 2020-11-18 19:37:26 -08:00
32eecac1a6
* It's HUB ID * Rename file * Working with official doc * Some progress * AttachedIO msg * device action impl * some better device alert impl * restructuring * Some port commands handled * Some command feedback waiting * Some more request-reply things * Some more request-reply things * Reworked msg classes * Getting back to UTs * Getting back to UTs * Facing sync lock problems * Facing sync lock problems * Testing it * Covering more with tests * handle actions * Hub class is almost covered * done coverage for Hub * done coverage for MoveHub * Button is tested * remove debug server from examples * Current and voltage tested * color sensor basic test * cover tilt sensor * motor sensor tested * constant motor * motor is tested * hold_speed impl for motor * motor commands recorded * some cleanup * some cleanup * some cleanup * debug * debug * FIX a bug * acc/dec profiles figured out * UT motor ops * UT motor ops * Get rid of weird piggyback * fix UT * Fix encoding? * fix test mb * More robust * Checked demo works * cosmetics * cosmetics * Maybe better test * fetching and decoding some device caps * describing devs * describing devs works * Applying modes we've learned * Simple and extensible dev attach * Reworking peripherals based on modes * Applying modes we've learned * implemented getting sensor data * fixed port subscribe * Added led out cmds on vision sensor * Worked on color-distance sensor * Introduce some locking for consistency * Improved it all * Travis flags * improve * improve * improve docs
62 lines
1.8 KiB
Python
62 lines
1.8 KiB
Python
import unittest
|
|
|
|
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)
|