| #!/usr/bin/env python |
| # Copyright (C) 2010-2016 Vincent Pelletier <plr.vincent@gmail.com> |
| # |
| # This library is free software; you can redistribute it and/or |
| # modify it under the terms of the GNU Lesser General Public |
| # License as published by the Free Software Foundation; either |
| # version 2.1 of the License, or (at your option) any later version. |
| # |
| # This library is distributed in the hope that it will be useful, |
| # but WITHOUT ANY WARRANTY; without even the implied warranty of |
| # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU |
| # Lesser General Public License for more details. |
| # |
| # You should have received a copy of the GNU Lesser General Public |
| # License along with this library; if not, write to the Free Software |
| # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA |
| |
| # pylint: disable=invalid-name, missing-docstring, too-many-public-methods |
| import unittest |
| import sys |
| import select |
| import threading |
| import usb1 |
| import libusb1 |
| from ctypes import pointer |
| |
| if sys.version_info[0] == 3: |
| buff = bytes([0, 0xff]) |
| other_buff = bytes((ord(x) for x in 'foo')) |
| else: |
| buff = '\x00\xff' |
| other_buff = 'foo' |
| buff_len = 2 |
| |
| class USBContext(usb1.USBContext): |
| def open(self): |
| try: |
| return super(USBContext, self).open() |
| except usb1.USBError: |
| raise unittest.SkipTest( |
| 'usb1.USBContext() fails - no USB bus on system ?' |
| ) |
| |
| class PollDetector(object): |
| def __init__(self, *args, **kw): |
| try: |
| poll = select.poll |
| except AttributeError: |
| raise unittest.SkipTest('select.poll missing') |
| self.__poll = poll(*args, **kw) |
| self.__event = threading.Event() |
| |
| def poll(self, *args, **kw): |
| self.__event.set() |
| return self.__poll.poll(*args, **kw) |
| |
| def wait(self, *args, **kw): |
| self.__event.wait(*args, **kw) |
| |
| def __getattr__(self, name): |
| return getattr(self.__poll, name) |
| |
| class USBTransferTests(unittest.TestCase): |
| @staticmethod |
| def getTransfer(iso_packets=0): |
| # Dummy handle |
| return usb1.USBTransfer( |
| pointer(libusb1.libusb_device_handle()), |
| iso_packets, lambda x: None, lambda x: None) |
| |
| @staticmethod |
| def testGetVersion(): |
| """ |
| Just testing getVersion doesn't raise... |
| """ |
| usb1.getVersion() |
| |
| @staticmethod |
| def testHasCapability(): |
| """ |
| Just testing hasCapability doesn't raise... |
| """ |
| usb1.hasCapability(usb1.CAP_HAS_CAPABILITY) |
| |
| def testSetControl(self): |
| """ |
| Simplest test: feed some data, must not raise. |
| """ |
| transfer = self.getTransfer() |
| request_type = usb1.TYPE_STANDARD |
| request = usb1.REQUEST_GET_STATUS |
| value = 0 |
| index = 0 |
| def callback(transfer): |
| pass |
| user_data = [] |
| timeout = 1000 |
| |
| # All provided, buffer variant |
| transfer.setControl( |
| request_type, request, value, index, buff, |
| callback=callback, user_data=user_data, timeout=timeout) |
| self.assertEqual(buff, transfer.getBuffer()) |
| self.assertRaises(ValueError, transfer.setBuffer, buff) |
| # All provided, buffer length variant |
| transfer.setControl( |
| request_type, request, value, index, buff_len, |
| callback=callback, user_data=user_data, timeout=timeout) |
| # No timeout |
| transfer.setControl( |
| request_type, request, value, index, buff, |
| callback=callback, user_data=user_data) |
| # No user data |
| transfer.setControl( |
| request_type, request, value, index, buff, callback=callback) |
| # No callback |
| transfer.setControl(request_type, request, value, index, buff) |
| |
| def _testSetBulkOrInterrupt(self, setter_id): |
| transfer = self.getTransfer() |
| endpoint = 0x81 |
| def callback(transfer): |
| pass |
| user_data = [] |
| timeout = 1000 |
| setter = getattr(transfer, setter_id) |
| # All provided, buffer variant |
| setter( |
| endpoint, buff, callback=callback, user_data=user_data, |
| timeout=timeout) |
| self.assertEqual(buff, transfer.getBuffer()) |
| transfer.setBuffer(other_buff) |
| self.assertEqual(other_buff, transfer.getBuffer()) |
| transfer.setBuffer(buff_len) |
| self.assertEqual(buff_len, len(transfer.getBuffer())) |
| # All provided, buffer length variant |
| setter( |
| endpoint, buff_len, callback=callback, user_data=user_data, |
| timeout=timeout) |
| # No timeout |
| setter(endpoint, buff, callback=callback, user_data=user_data) |
| # No user data |
| setter(endpoint, buff, callback=callback) |
| # No callback |
| setter(endpoint, buff) |
| |
| def testSetBulk(self): |
| """ |
| Simplest test: feed some data, must not raise. |
| Also, test setBuffer/getBuffer. |
| """ |
| self._testSetBulkOrInterrupt('setBulk') |
| |
| def testSetInterrupt(self): |
| """ |
| Simplest test: feed some data, must not raise. |
| Also, test setBuffer/getBuffer. |
| """ |
| self._testSetBulkOrInterrupt('setInterrupt') |
| |
| def testSetGetCallback(self): |
| transfer = self.getTransfer() |
| def callback(transfer): |
| pass |
| transfer.setCallback(callback) |
| got_callback = transfer.getCallback() |
| self.assertEqual(callback, got_callback) |
| |
| def testUSBPollerThreadExit(self): |
| """ |
| USBPollerThread must exit by itself when context is destroyed. |
| """ |
| with USBContext() as context: |
| poll_detector = PollDetector() |
| try: |
| poller = usb1.USBPollerThread(context, poll_detector) |
| except OSError: |
| raise unittest.SkipTest('libusb without file descriptor events') |
| poller.start() |
| poll_detector.wait(1) |
| poller.join(1) |
| self.assertFalse(poller.is_alive()) |
| |
| def testUSBPollerThreadException(self): |
| """ |
| USBPollerThread exception handling. |
| """ |
| class FakeEventPoll(PollDetector): |
| # pylint: disable=method-hidden |
| def poll(self, *args, **kw): |
| self.poll = super(FakeEventPoll, self).poll |
| return ['dummy'] |
| # pylint: enable=method-hidden |
| with USBContext() as context: |
| def fakeHandleEventsLocked(): |
| raise usb1.USBError(0) |
| context.handleEventsLocked = fakeHandleEventsLocked |
| exception_event = threading.Event() |
| exception_list = [] |
| def exceptionHandler(exc): |
| exception_list.append(exc) |
| exception_event.set() |
| try: |
| poller = usb1.USBPollerThread( |
| context, FakeEventPoll(), exceptionHandler) |
| except OSError: |
| raise unittest.SkipTest('libusb without file descriptor events') |
| poller.start() |
| exception_event.wait(1) |
| self.assertTrue(exception_list, exception_list) |
| self.assertTrue(poller.is_alive()) |
| |
| @staticmethod |
| def testDescriptors(): |
| """ |
| Test descriptor walk. |
| Needs any usb device, which won't be opened. |
| """ |
| with USBContext() as context: |
| device_list = context.getDeviceList(skip_on_error=True) |
| found = False |
| for device in device_list: |
| device.getBusNumber() |
| device.getPortNumber() |
| device.getPortNumberList() |
| device.getDeviceAddress() |
| for settings in device.iterSettings(): |
| for endpoint in settings: |
| pass |
| for configuration in device.iterConfigurations(): |
| for interface in configuration: |
| for settings in interface: |
| for endpoint in settings: |
| found = True |
| if not found: |
| raise unittest.SkipTest('descriptor walk test did not complete') |
| |
| def testDefaultEnumScope(self): |
| """ |
| Enum instances must only affect the scope they are created in. |
| """ |
| ENUM_NAME = 'THE_ANSWER' |
| ENUM_VALUE = 42 |
| local_dict = locals() |
| global_dict = globals() |
| self.assertEqual(local_dict.get(ENUM_NAME), None) |
| self.assertEqual(global_dict.get(ENUM_NAME), None) |
| self.assertEqual(getattr(libusb1, ENUM_NAME, None), None) |
| # pylint: disable=unused-variable |
| TEST_ENUM = libusb1.Enum({ENUM_NAME: ENUM_VALUE}) |
| # pylint: enable=unused-variable |
| self.assertEqual(local_dict.get(ENUM_NAME), ENUM_VALUE) |
| self.assertEqual(global_dict.get(ENUM_NAME), None) |
| self.assertEqual(getattr(libusb1, ENUM_NAME, None), None) |
| |
| def testExplicitEnumScope(self): |
| """ |
| Enum instances must only affect the scope they are created in. |
| """ |
| ENUM_NAME = 'THE_ANSWER' |
| ENUM_VALUE = 42 |
| local_dict = locals() |
| global_dict = globals() |
| self.assertEqual(local_dict.get(ENUM_NAME), None) |
| self.assertEqual(global_dict.get(ENUM_NAME), None) |
| self.assertEqual(getattr(libusb1, ENUM_NAME, None), None) |
| # pylint: disable=unused-variable |
| TEST_ENUM = libusb1.Enum({ENUM_NAME: ENUM_VALUE}, global_dict) |
| # pylint: enable=unused-variable |
| try: |
| self.assertEqual(local_dict.get(ENUM_NAME), None) |
| self.assertEqual(global_dict.get(ENUM_NAME), ENUM_VALUE) |
| self.assertEqual(getattr(libusb1, ENUM_NAME, None), None) |
| finally: |
| del global_dict[ENUM_NAME] |
| |
| def testImplicitUSBContextOpening(self): |
| """ |
| Test pre-1.5 API backward compatibility. |
| First method call which needs a context succeeds. |
| Further calls return None. |
| """ |
| context = USBContext() # Deprecated |
| try: |
| fd_list = context.getPollFDList() |
| except NotImplementedError: |
| raise unittest.SkipTest('libusb without file descriptor events') |
| self.assertNotEqual(fd_list, None) |
| context.exit() # Deprecated |
| self.assertEqual(context.getPollFDList(), None) |
| |
| if __name__ == '__main__': |
| unittest.main() |
