[usb_gadget p02] Basic USB device implementation.

The gadget.Gadget class implements the basic USB control requests
required by all devices. It handles common operations like changing
the configuration and interface alternatives and fetching descriptors.
The device interfaces with a "chip" which is the USB peripheral
controller that connects the emulated device to a USB host.

BUG=396682
R=rockot@chromium.org,rpaquay@chromium.org,kalman@chromium.org

Review URL: https://codereview.chromium.org/407353002

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@285098 0039d316-1c4b-4281-b951-d872f2087c98

2 files changed, 717 insertions, 0 deletions

diff --git a/tools/usb_gadget/gadget.py b/tools/usb_gadget/gadget.py
new file mode 100644
index 0000000..fb48a2c
--- /dev/null
+++ b/tools/usb_gadget/gadget.py
@@ -0,0 +1,432 @@
+# Copyright 2014 The Chromium Authors. All rights reserved.
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+"""Generic USB gadget functionality.
+"""
+
+import struct
+
+import usb_constants
+
+
+class Gadget(object):
+  """Basic functionality for a USB device.
+
+  Implements standard control requests assuming that a subclass will handle
+  class- or vendor-specific requests.
+  """
+
+  def __init__(self, device_desc, fs_config_desc, hs_config_desc):
+    """Create a USB gadget device.
+
+    Args:
+      device_desc: USB device descriptor.
+      fs_config_desc: Low/full-speed device descriptor.
+      hs_config_desc: High-speed device descriptor.
+    """
+    self._speed = usb_constants.Speed.UNKNOWN
+    self._chip = None
+    self._device_desc = device_desc
+    self._fs_config_desc = fs_config_desc
+    self._hs_config_desc = hs_config_desc
+    # dict mapping language codes to a dict mapping indexes to strings
+    self._strings = {}
+    # dict mapping interface numbers to a set of endpoint addresses
+    self._active_endpoints = {}
+
+  def GetDeviceDescriptor(self):
+    return self._device_desc
+
+  def GetFullSpeedConfigurationDescriptor(self):
+    return self._fs_config_desc
+
+  def GetHighSpeedConfigurationDescriptor(self):
+    return self._hs_config_desc
+
+  def GetConfigurationDescriptor(self):
+    if self._speed == usb_constants.Speed.FULL:
+      return self._fs_config_desc
+    elif self._speed == usb_constants.Speed.HIGH:
+      return self._hs_config_desc
+    else:
+      raise RuntimeError('Device is not connected.')
+
+  def GetSpeed(self):
+    return self._speed
+
+  def AddStringDescriptor(self, index, value, lang=0x0409):
+    """Add a string descriptor to this device.
+
+    Args:
+      index: String descriptor index (matches 'i' fields in descriptors).
+      value: The string.
+      lang: Language code (default: English).
+
+    Raises:
+      ValueError: The index or language code is invalid.
+    """
+    if index < 1 or index > 255:
+      raise ValueError('String descriptor index out of range.')
+    if lang < 0 or lang > 0xffff:
+      raise ValueError('String descriptor language code out of range.')
+
+    lang_strings = self._strings.setdefault(lang, {})
+    lang_strings[index] = value
+
+  def Connected(self, chip, speed):
+    """The device has been connected to a USB host.
+
+    Args:
+      chip: USB controller.
+      speed: Connection speed.
+    """
+    self._speed = speed
+    self._chip = chip
+
+  def Disconnected(self):
+    """The device has been disconnected from the USB host."""
+    self._speed = usb_constants.Speed.UNKNOWN
+    self._chip = None
+    self._active_endpoints.clear()
+
+  def IsConnected(self):
+    return self._chip is not None
+
+  def ControlRead(self, request_type, request, value, index, length):
+    """Handle a read on the control pipe (endpoint zero).
+
+    Args:
+      request_type: bmRequestType field of the setup packet.
+      request: bRequest field of the setup packet.
+      value: wValue field of the setup packet.
+      index: wIndex field of the setup packet.
+      length: Maximum amount of data the host expects the device to return.
+
+    Returns:
+      A buffer to return to the USB host with len <= length on success or
+      None to stall the pipe.
+    """
+    assert request_type & usb_constants.Dir.IN
+    typ = request_type & usb_constants.Type.MASK
+    recipient = request_type & usb_constants.Recipient.MASK
+    if typ == usb_constants.Type.STANDARD:
+      return self.StandardControlRead(
+          recipient, request, value, index, length)
+    elif typ == usb_constants.Type.CLASS:
+      return self.ClassControlRead(
+          recipient, request, value, index, length)
+    elif typ == usb_constants.Type.VENDOR:
+      return self.VendorControlRead(
+          recipient, request, value, index, length)
+
+  def ControlWrite(self, request_type, request, value, index, data):
+    """Handle a write to the control pipe (endpoint zero).
+
+    Args:
+      request_type: bmRequestType field of the setup packet.
+      request: bRequest field of the setup packet.
+      value: wValue field of the setup packet.
+      index: wIndex field of the setup packet.
+      data: Data stage of the request.
+
+    Returns:
+      True on success, None to stall the pipe.
+    """
+    assert not request_type & usb_constants.Dir.IN
+    typ = request_type & usb_constants.Type.MASK
+    recipient = request_type & usb_constants.Recipient.MASK
+    if typ == usb_constants.Type.STANDARD:
+      return self.StandardControlWrite(
+          recipient, request, value, index, data)
+    elif typ == usb_constants.Type.CLASS:
+      return self.ClassControlWrite(
+          recipient, request, value, index, data)
+    elif typ == usb_constants.Type.VENDOR:
+      return self.VendorControlWrite(
+          recipient, request, value, index, data)
+
+  def SendPacket(self, endpoint, data):
+    """Send a data packet on the given endpoint.
+
+    Args:
+      endpoint: Endpoint address.
+      data: Data buffer.
+
+    Raises:
+      ValueError: If the endpoint address is not valid.
+      RuntimeError: If the device is not connected.
+    """
+    if self._chip is None:
+      raise RuntimeError('Device is not connected.')
+    if not endpoint & usb_constants.Dir.IN:
+      raise ValueError('Cannot write to non-input endpoint.')
+    self._chip.SendPacket(endpoint, data)
+
+  def ReceivePacket(self, endpoint, data):
+    """Handle an incoming data packet on one of the device's active endpoints.
+
+    This method should be overridden by a subclass implementing endpoint-based
+    data transfers.
+
+    Args:
+      endpoint: Endpoint address.
+      data: Data buffer.
+    """
+    pass
+
+  def HaltEndpoint(self, endpoint):
+    """Signals a STALL condition to the host on the given endpoint.
+
+    Args:
+      endpoint: Endpoint address.
+    """
+    self._chip.HaltEndpoint(endpoint)
+
+  def StandardControlRead(self, recipient, request, value, index, length):
+    """Handle standard control transfers.
+
+    Args:
+      recipient: Request recipient (device, interface, endpoint, etc.)
+      request: bRequest field of the setup packet.
+      value: wValue field of the setup packet.
+      index: wIndex field of the setup packet.
+      length: Maximum amount of data the host expects the device to return.
+
+    Returns:
+      A buffer to return to the USB host with len <= length on success or
+      None to stall the pipe.
+    """
+    if request == usb_constants.Request.GET_DESCRIPTOR:
+      desc_type = value >> 8
+      desc_index = value & 0xff
+      desc_lang = index
+
+      print 'GetDescriptor(recipient={}, type={}, index={}, lang={})'.format(
+          recipient, desc_type, desc_index, desc_lang)
+
+      return self.GetDescriptor(recipient, desc_type, desc_index, desc_lang,
+                                length)
+
+  def GetDescriptor(self, recipient, typ, index, lang, length):
+    """Handle a standard GET_DESCRIPTOR request.
+
+    See Universal Serial Bus Specification Revision 2.0 section 9.4.3.
+
+    Args:
+      recipient: Request recipient (device, interface, endpoint, etc.)
+      typ: Descriptor type.
+      index: Descriptor index.
+      lang: Descriptor language code.
+      length: Maximum amount of data the host expects the device to return.
+
+    Returns:
+      The value of the descriptor or None to stall the pipe.
+    """
+    if recipient == usb_constants.Recipient.DEVICE:
+      if typ == usb_constants.DescriptorType.STRING:
+        return self.GetStringDescriptor(index, lang, length)
+
+  def ClassControlRead(self, recipient, request, value, index, length):
+    """Handle class-specific control transfers.
+
+    This function should be overridden by a subclass implementing a particular
+    device class.
+
+    Args:
+      recipient: Request recipient (device, interface, endpoint, etc.)
+      request: bRequest field of the setup packet.
+      value: wValue field of the setup packet.
+      index: wIndex field of the setup packet.
+      length: Maximum amount of data the host expects the device to return.
+
+    Returns:
+      A buffer to return to the USB host with len <= length on success or
+      None to stall the pipe.
+    """
+    _ = recipient, request, value, index, length
+    return None
+
+  def VendorControlRead(self, recipient, request, value, index, length):
+    """Handle vendor-specific control transfers.
+
+    This function should be overridden by a subclass if implementing a device
+    that responds to vendor-specific requests.
+
+    Args:
+      recipient: Request recipient (device, interface, endpoint, etc.)
+      request: bRequest field of the setup packet.
+      value: wValue field of the setup packet.
+      index: wIndex field of the setup packet.
+      length: Maximum amount of data the host expects the device to return.
+
+    Returns:
+      A buffer to return to the USB host with len <= length on success or
+      None to stall the pipe.
+    """
+    _ = recipient, request, value, index, length
+    return None
+
+  def StandardControlWrite(self, recipient, request, value, index, data):
+    """Handle standard control transfers.
+
+    Args:
+      recipient: Request recipient (device, interface, endpoint, etc.)
+      request: bRequest field of the setup packet.
+      value: wValue field of the setup packet.
+      index: wIndex field of the setup packet.
+      data: Data stage of the request.
+
+    Returns:
+      True on success, None to stall the pipe.
+    """
+    _ = data
+
+    if request == usb_constants.Request.SET_CONFIGURATION:
+      if recipient == usb_constants.Recipient.DEVICE:
+        return self.SetConfiguration(value)
+    elif request == usb_constants.Request.SET_INTERFACE:
+      if recipient == usb_constants.Recipient.INTERFACE:
+        return self.SetInterface(index, value)
+
+  def ClassControlWrite(self, recipient, request, value, index, data):
+    """Handle class-specific control transfers.
+
+    This function should be overridden by a subclass implementing a particular
+    device class.
+
+    Args:
+      recipient: Request recipient (device, interface, endpoint, etc.)
+      request: bRequest field of the setup packet.
+      value: wValue field of the setup packet.
+      index: wIndex field of the setup packet.
+      data: Data stage of the request.
+
+    Returns:
+      True on success, None to stall the pipe.
+    """
+    _ = recipient, request, value, index, data
+    return None
+
+  def VendorControlWrite(self, recipient, request, value, index, data):
+    """Handle vendor-specific control transfers.
+
+    This function should be overridden by a subclass if implementing a device
+    that responds to vendor-specific requests.
+
+    Args:
+      recipient: Request recipient (device, interface, endpoint, etc.)
+      request: bRequest field of the setup packet.
+      value: wValue field of the setup packet.
+      index: wIndex field of the setup packet.
+      data: Data stage of the request.
+
+    Returns:
+      True on success, None to stall the pipe.
+    """
+    _ = recipient, request, value, index, data
+    return None
+
+  def GetStringDescriptor(self, index, lang, length):
+    """Handle a GET_DESCRIPTOR(String) request from the host.
+
+    Descriptor index 0 returns the set of languages supported by the device.
+    All other indices return the string descriptors registered with those
+    indices.
+
+    See Universal Serial Bus Specification Revision 2.0 section 9.6.7.
+
+    Args:
+      index: Descriptor index.
+      lang: Descriptor language code.
+      length: Maximum amount of data the host expects the device to return.
+
+    Returns:
+      The string descriptor or None to stall the pipe if the descriptor is not
+      found.
+    """
+    if index == 0:
+      length = 2 + len(self._strings) * 2
+      header = struct.pack('<BB', length, usb_constants.DescriptorType.STRING)
+      lang_codes = [struct.pack('<H', lang)
+                    for lang in self._strings.iterkeys()]
+      buf = header + ''.join(lang_codes)
+      assert len(buf) == length
+      return buf[:length]
+    elif lang not in self._strings:
+      return None
+    elif index not in self._strings[lang]:
+      return None
+    else:
+      string = self._strings[lang][index].encode('UTF-16LE')
+      header = struct.pack(
+          '<BB', 2 + len(string), usb_constants.DescriptorType.STRING)
+      buf = header + string
+      return buf[:length]
+
+  def SetConfiguration(self, index):
+    """Handle a SET_CONFIGURATION request from the host.
+
+    See Universal Serial Bus Specification Revision 2.0 section 9.4.7.
+
+    Args:
+      index: Configuration index selected.
+
+    Returns:
+      True on success, None on error to stall the pipe.
+    """
+    print 'SetConfiguration({})'.format(index)
+
+    for endpoint_addrs in self._active_endpoints.values():
+      for endpoint_addr in endpoint_addrs:
+        self._chip.StopEndpoint(endpoint_addr)
+      endpoint_addrs.clear()
+
+    if index == 0:
+      # SET_CONFIGRATION(0) puts the device into the Address state which
+      # Windows does before suspending the port.
+      return True
+    elif index != 1:
+      return None
+
+    config_desc = self.GetConfigurationDescriptor()
+    for interface_desc in config_desc.GetInterfaces():
+      if interface_desc.bAlternateSetting != 0:
+        continue
+      endpoint_addrs = self._active_endpoints.setdefault(
+          interface_desc.bInterfaceNumber, set())
+      for endpoint_desc in interface_desc.GetEndpoints():
+        self._chip.StartEndpoint(endpoint_desc)
+        endpoint_addrs.add(endpoint_desc.bEndpointAddress)
+    return True
+
+  def SetInterface(self, interface, alt_setting):
+    """Handle a SET_INTERFACE request from the host.
+
+    See Universal Serial Bus Specification Revision 2.0 section 9.4.10.
+
+    Args:
+      interface: Interface number to configure.
+      alt_setting: Alternate setting to select.
+
+    Returns:
+      True on success, None on error to stall the pipe.
+    """
+    print 'SetInterface({}, {})'.format(interface, alt_setting)
+
+    config_desc = self.GetConfigurationDescriptor()
+    interface_desc = None
+    for interface_option in config_desc.GetInterfaces():
+      if (interface_option.bInterfaceNumber == interface and
+          interface_option.bAlternateSetting == alt_setting):
+        interface_desc = interface_option
+    if interface_desc is None:
+      return None
+
+    endpoint_addrs = self._active_endpoints.setdefault(interface, set())
+    for endpoint_addr in endpoint_addrs:
+      self._chip.StopEndpoint(endpoint_addr)
+    for endpoint_desc in interface_desc.GetEndpoints():
+      self._chip.StartEndpoint(endpoint_desc)
+      endpoint_addrs.add(endpoint_desc.bEndpointAddress)
+    return True
diff --git a/tools/usb_gadget/gadget_test.py b/tools/usb_gadget/gadget_test.py
new file mode 100755
index 0000000..066914c
--- /dev/null
+++ b/tools/usb_gadget/gadget_test.py
@@ -0,0 +1,285 @@
+#!/usr/bin/python
+# Copyright 2014 The Chromium Authors. All rights reserved.
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+import unittest
+
+import mock
+
+import gadget
+import usb_constants
+import usb_descriptors
+
+
+device_desc = usb_descriptors.DeviceDescriptor(
+    idVendor=0x18D1,   # Google Inc.
+    idProduct=0xFF00,
+    bcdUSB=0x0200,
+    iManufacturer=1,
+    iProduct=2,
+    iSerialNumber=3,
+    bNumConfigurations=1,
+    bcdDevice=0x0100)
+
+fs_config_desc = usb_descriptors.ConfigurationDescriptor(
+    bmAttributes=0xC0,
+    MaxPower=50)
+
+fs_interface_desc = usb_descriptors.InterfaceDescriptor(
+    bInterfaceNumber=0
+)
+fs_config_desc.AddInterface(fs_interface_desc)
+
+fs_bulk_in_endpoint_desc = usb_descriptors.EndpointDescriptor(
+    bEndpointAddress=0x01,
+    bmAttributes=usb_constants.TransferType.BULK,
+    wMaxPacketSize=64,
+    bInterval=0
+)
+fs_interface_desc.AddEndpoint(fs_bulk_in_endpoint_desc)
+
+fs_bulk_out_endpoint_desc = usb_descriptors.EndpointDescriptor(
+    bEndpointAddress=0x81,
+    bmAttributes=usb_constants.TransferType.BULK,
+    wMaxPacketSize=64,
+    bInterval=0
+)
+fs_interface_desc.AddEndpoint(fs_bulk_out_endpoint_desc)
+
+fs_alt_interface_desc = usb_descriptors.InterfaceDescriptor(
+    bInterfaceNumber=0,
+    bAlternateSetting=1
+)
+fs_config_desc.AddInterface(fs_alt_interface_desc)
+
+fs_interrupt_in_endpoint_desc = usb_descriptors.EndpointDescriptor(
+    bEndpointAddress=0x01,
+    bmAttributes=usb_constants.TransferType.INTERRUPT,
+    wMaxPacketSize=64,
+    bInterval=1
+)
+fs_alt_interface_desc.AddEndpoint(fs_interrupt_in_endpoint_desc)
+
+fs_interrupt_out_endpoint_desc = usb_descriptors.EndpointDescriptor(
+    bEndpointAddress=0x81,
+    bmAttributes=usb_constants.TransferType.INTERRUPT,
+    wMaxPacketSize=64,
+    bInterval=1
+)
+fs_alt_interface_desc.AddEndpoint(fs_interrupt_out_endpoint_desc)
+
+hs_config_desc = usb_descriptors.ConfigurationDescriptor(
+    bmAttributes=0xC0,
+    MaxPower=50)
+
+hs_interface_desc = usb_descriptors.InterfaceDescriptor(
+    bInterfaceNumber=0
+)
+hs_config_desc.AddInterface(hs_interface_desc)
+
+hs_bulk_in_endpoint_desc = usb_descriptors.EndpointDescriptor(
+    bEndpointAddress=0x01,
+    bmAttributes=usb_constants.TransferType.BULK,
+    wMaxPacketSize=512,
+    bInterval=0
+)
+hs_interface_desc.AddEndpoint(hs_bulk_in_endpoint_desc)
+
+hs_bulk_out_endpoint_desc = usb_descriptors.EndpointDescriptor(
+    bEndpointAddress=0x81,
+    bmAttributes=usb_constants.TransferType.BULK,
+    wMaxPacketSize=512,
+    bInterval=0
+)
+hs_interface_desc.AddEndpoint(hs_bulk_out_endpoint_desc)
+
+hs_alt_interface_desc = usb_descriptors.InterfaceDescriptor(
+    bInterfaceNumber=0,
+    bAlternateSetting=1
+)
+hs_config_desc.AddInterface(hs_alt_interface_desc)
+
+hs_interrupt_in_endpoint_desc = usb_descriptors.EndpointDescriptor(
+    bEndpointAddress=0x01,
+    bmAttributes=usb_constants.TransferType.INTERRUPT,
+    wMaxPacketSize=256,
+    bInterval=1
+)
+hs_alt_interface_desc.AddEndpoint(hs_interrupt_in_endpoint_desc)
+
+hs_interrupt_out_endpoint_desc = usb_descriptors.EndpointDescriptor(
+    bEndpointAddress=0x81,
+    bmAttributes=usb_constants.TransferType.INTERRUPT,
+    wMaxPacketSize=256,
+    bInterval=1
+)
+hs_alt_interface_desc.AddEndpoint(hs_interrupt_out_endpoint_desc)
+
+
+class GadgetTest(unittest.TestCase):
+
+  def test_get_descriptors(self):
+    g = gadget.Gadget(device_desc, fs_config_desc, hs_config_desc)
+    self.assertEquals(g.GetDeviceDescriptor(), device_desc)
+    self.assertEquals(g.GetFullSpeedConfigurationDescriptor(), fs_config_desc)
+    self.assertEquals(g.GetHighSpeedConfigurationDescriptor(), hs_config_desc)
+    with self.assertRaisesRegexp(RuntimeError, 'not connected'):
+      g.GetConfigurationDescriptor()
+
+  def test_connect_full_speed(self):
+    g = gadget.Gadget(device_desc, fs_config_desc, hs_config_desc)
+    g.Connected(mock.Mock(), usb_constants.Speed.FULL)
+    self.assertTrue(g.IsConnected())
+    self.assertEquals(g.GetSpeed(), usb_constants.Speed.FULL)
+    self.assertEquals(g.GetConfigurationDescriptor(), fs_config_desc)
+    g.Disconnected()
+    self.assertFalse(g.IsConnected())
+
+  def test_connect_high_speed(self):
+    g = gadget.Gadget(device_desc, fs_config_desc, hs_config_desc)
+    g.Connected(mock.Mock(), usb_constants.Speed.HIGH)
+    self.assertTrue(g.IsConnected())
+    self.assertEquals(g.GetSpeed(), usb_constants.Speed.HIGH)
+    self.assertEquals(g.GetConfigurationDescriptor(), hs_config_desc)
+    g.Disconnected()
+    self.assertFalse(g.IsConnected())
+
+  def test_string_index_out_of_range(self):
+    g = gadget.Gadget(device_desc, fs_config_desc, hs_config_desc)
+    with self.assertRaisesRegexp(ValueError, 'index out of range'):
+      g.AddStringDescriptor(0, 'Hello world!')
+
+  def test_language_id_out_of_range(self):
+    g = gadget.Gadget(device_desc, fs_config_desc, hs_config_desc)
+    with self.assertRaisesRegexp(ValueError, 'language code out of range'):
+      g.AddStringDescriptor(1, 'Hello world!', lang=-1)
+
+  def test_get_languages(self):
+    g = gadget.Gadget(device_desc, fs_config_desc, hs_config_desc)
+    g.AddStringDescriptor(1, 'Hello world!')
+    desc = g.ControlRead(0x80, 6, 0x0300, 0, 255)
+    self.assertEquals(desc, '\x04\x03\x09\x04')
+
+  def test_get_string_descriptor(self):
+    g = gadget.Gadget(device_desc, fs_config_desc, hs_config_desc)
+    g.AddStringDescriptor(1, 'Hello world!')
+    desc = g.ControlRead(0x80, 6, 0x0301, 0x0409, 255)
+    self.assertEquals(desc, '\x1A\x03H\0e\0l\0l\0o\0 \0w\0o\0r\0l\0d\0!\0')
+
+  def test_get_missing_string_descriptor(self):
+    g = gadget.Gadget(device_desc, fs_config_desc, hs_config_desc)
+    g.AddStringDescriptor(1, 'Hello world!')
+    desc = g.ControlRead(0x80, 6, 0x0302, 0x0409, 255)
+    self.assertEquals(desc, None)
+
+  def test_get_missing_string_language(self):
+    g = gadget.Gadget(device_desc, fs_config_desc, hs_config_desc)
+    g.AddStringDescriptor(1, 'Hello world!')
+    desc = g.ControlRead(0x80, 6, 0x0301, 0x040C, 255)
+    self.assertEquals(desc, None)
+
+  def test_class_and_vendor_transfers(self):
+    g = gadget.Gadget(device_desc, fs_config_desc, hs_config_desc)
+    self.assertIsNone(g.ControlRead(0xA0, 0, 0, 0, 0))
+    self.assertIsNone(g.ControlRead(0xC0, 0, 0, 0, 0))
+    self.assertIsNone(g.ControlWrite(0x20, 0, 0, 0, ''))
+    self.assertIsNone(g.ControlWrite(0x40, 0, 0, 0, ''))
+
+  def test_set_configuration(self):
+    g = gadget.Gadget(device_desc, fs_config_desc, hs_config_desc)
+    chip = mock.Mock()
+    g.Connected(chip, usb_constants.Speed.HIGH)
+    g.ControlWrite(0, 9, 1, 0, 0)
+    chip.StartEndpoint.assert_has_calls([
+        mock.call(hs_bulk_in_endpoint_desc),
+        mock.call(hs_bulk_out_endpoint_desc)
+    ])
+
+  def test_set_configuration_zero(self):
+    g = gadget.Gadget(device_desc, fs_config_desc, hs_config_desc)
+    chip = mock.Mock()
+    g.Connected(chip, usb_constants.Speed.HIGH)
+    g.ControlWrite(0, 9, 1, 0, 0)
+    chip.StartEndpoint.reset_mock()
+    g.ControlWrite(0, 9, 0, 0, 0)
+    chip.StopEndpoint.assert_has_calls([
+        mock.call(0x01),
+        mock.call(0x81)
+    ])
+
+  def test_set_bad_configuration(self):
+    g = gadget.Gadget(device_desc, fs_config_desc, hs_config_desc)
+    g.Connected(mock.Mock(), usb_constants.Speed.HIGH)
+    self.assertIsNone(g.ControlWrite(0, 9, 2, 0, 0))
+
+  def test_set_interface(self):
+    g = gadget.Gadget(device_desc, fs_config_desc, hs_config_desc)
+    chip = mock.Mock()
+    g.Connected(chip, usb_constants.Speed.HIGH)
+    self.assertTrue(g.ControlWrite(0, 9, 1, 0, 0))
+    chip.reset_mock()
+    self.assertTrue(g.ControlWrite(1, 11, 1, 0, 0))
+    chip.StopEndpoint.assert_has_calls([
+        mock.call(0x01),
+        mock.call(0x81)
+    ])
+    chip.StartEndpoint.assert_has_calls([
+        mock.call(hs_interrupt_in_endpoint_desc),
+        mock.call(hs_interrupt_out_endpoint_desc)
+    ])
+    chip.reset_mock()
+    self.assertTrue(g.ControlWrite(1, 11, 0, 0, 0))
+    chip.StopEndpoint.assert_has_calls([
+        mock.call(0x01),
+        mock.call(0x81)
+    ])
+    chip.StartEndpoint.assert_has_calls([
+        mock.call(hs_bulk_in_endpoint_desc),
+        mock.call(hs_bulk_out_endpoint_desc)
+    ])
+
+  def test_set_bad_interface(self):
+    g = gadget.Gadget(device_desc, fs_config_desc, hs_config_desc)
+    g.Connected(mock.Mock(), usb_constants.Speed.HIGH)
+    self.assertTrue(g.ControlWrite(0, 9, 1, 0, 0))
+    self.assertIsNone(g.ControlWrite(1, 11, 0, 1, 0))
+
+  def test_send_packet(self):
+    g = gadget.Gadget(device_desc, fs_config_desc, hs_config_desc)
+    chip = mock.Mock()
+    g.Connected(chip, usb_constants.Speed.HIGH)
+    g.SendPacket(0x81, 'Hello world!')
+    chip.SendPacket.assert_called_once_with(0x81, 'Hello world!')
+
+  def test_send_packet_disconnected(self):
+    g = gadget.Gadget(device_desc, fs_config_desc, hs_config_desc)
+    with self.assertRaisesRegexp(RuntimeError, 'not connected'):
+      g.SendPacket(0x81, 'Hello world!')
+    g.Connected(mock.Mock(), usb_constants.Speed.HIGH)
+    g.SendPacket(0x81, 'Hello world!')
+    g.Disconnected()
+    with self.assertRaisesRegexp(RuntimeError, 'not connected'):
+      g.SendPacket(0x81, 'Hello world!')
+
+  def test_send_invalid_endpoint(self):
+    g = gadget.Gadget(device_desc, fs_config_desc, hs_config_desc)
+    chip = mock.Mock()
+    g.Connected(chip, usb_constants.Speed.HIGH)
+    with self.assertRaisesRegexp(ValueError, 'non-input endpoint'):
+      g.SendPacket(0x01, 'Hello world!')
+
+  def test_receive_packet(self):
+    g = gadget.Gadget(device_desc, fs_config_desc, hs_config_desc)
+    self.assertIsNone(g.ReceivePacket(0x01, 'Hello world!'))
+
+  def test_halt_endpoint(self):
+    g = gadget.Gadget(device_desc, fs_config_desc, hs_config_desc)
+    chip = mock.Mock()
+    g.Connected(chip, usb_constants.Speed.HIGH)
+    g.HaltEndpoint(0x01)
+    chip.HaltEndpoint.assert_called_once_with(0x01)
+
+
+if __name__ == '__main__':
+  unittest.main()