def test_from_name_nonexisting_subsystem(self, context):
     with pytest.raises(DeviceNotFoundByNameError) as exc_info:
         Device.from_name(context, "no_such_subsystem", "foobar")
     error = exc_info.value
     assert error.subsystem == "no_such_subsystem"
     assert error.sys_name == "foobar"
     assert str(error) == "No device {0!r} in {1!r}".format(error.sys_name, error.subsystem)

 def test_from_device_file_no_device_file(self, context, tmpdir):
     filename = tmpdir.join("test")
     filename.ensure(file=True)
     with pytest.raises(ValueError) as excinfo:
         Device.from_device_file(context, str(filename))
     message = "not a device file: {0!r}".format(str(filename))
     assert str(excinfo.value) == message

 def test_from_sys_path_device_not_found(self, context):
     sys_path = "there_will_not_be_such_a_device"
     with pytest.raises(DeviceNotFoundAtPathError) as exc_info:
         Device.from_sys_path(context, sys_path)
     error = exc_info.value
     assert error.sys_path == sys_path
     assert str(error) == "No device at {0!r}".format(sys_path)

 def test_from_name_no_device_in_existing_subsystem(self, context):
     with pytest.raises(DeviceNotFoundByNameError) as exc_info:
         Device.from_name(context, 'block', 'foobar')
     error = exc_info.value
     assert error.subsystem == 'block'
     assert error.sys_name == 'foobar'
     assert str(error) == 'No device {0!r} in {1!r}'.format(
         error.sys_name, error.subsystem)

    def _hp_ld_to_lsm_vol(hp_ld, pool_id, sys_id, ctrl_num, array_num, hp_ld_name):
        """
        raises DeviceNotFoundError
        """
        ld_num = hp_ld_name[len("Logical Drive: ") :]
        vpd83 = hp_ld["Unique Identifier"].lower()
        # No document or command output indicate block size
        # of volume. So we try to read from linux kernel, if failed
        # try 512 and roughly calculate the sector count.
        device = Device.from_device_file(_CONTEXT, hp_ld["Disk Name"])
        vol_name = "%s: /dev/%s" % (hp_ld_name, device.sys_name)
        attributes = device.attributes
        try:
            block_size = attributes.asint("queue/logical_block_size")
            num_of_blocks = attributes.asint("size")
        except (KeyError, UnicodeDecodeError, ValueError):
            block_size = 512
            num_of_blocks = int(_hp_size_to_lsm(hp_ld["Size"]) / block_size)

        if "Failed" in hp_ld["Status"]:
            admin_status = Volume.ADMIN_STATE_DISABLED
        else:
            admin_status = Volume.ADMIN_STATE_ENABLED
        plugin_data = "%s:%s:%s" % (ctrl_num, array_num, ld_num)

        # HP SmartArray does not allow disabling volume.
        return Volume(vpd83, vol_name, vpd83, block_size, num_of_blocks, admin_status, sys_id, pool_id, plugin_data)

def get_indexed_string(device_handle, index):
	"""Get a string from a HID device, based on its string index.

	Note: currently not working in the ``hidraw`` native implementation.

	:param device_handle: a device handle returned by open() or open_path().
	:param index: the index of the string to get.
	"""
	if index not in _DEVICE_STRINGS:
		return None

	assert device_handle
	stat = _os.fstat(device_handle)
	dev = _Device.from_device_number(_Context(), 'char', stat.st_rdev)
	if dev:
		hid_dev = dev.find_parent('hid')
		if hid_dev:
			assert 'HID_ID' in hid_dev
			bus, _ignore, _ignore = hid_dev['HID_ID'].split(':')

			if bus == '0003':  # USB
				usb_dev = dev.find_parent('usb', 'usb_device')
				assert usb_dev
				key = _DEVICE_STRINGS[index]
				attrs = usb_dev.attributes
				if key in attrs:
					return attrs[key]

			elif bus == '0005':  # BLUETOOTH
				# TODO
				pass

def get_indexed_string(device_handle, index):
	"""Get a string from a HID device, based on its string index.

	Note: currently not working in the ``hidraw`` native implementation.

	:param device_handle: a device handle returned by open() or open_path().
	:param index: the index of the string to get.
	:returns: the value corresponding to index, or None if no value found
	:rtype: bytes or NoneType
	"""
	try:
	    key = _DEVICE_STRINGS[index]
	except KeyError:
	    return None

	assert device_handle
	stat = _os.fstat(device_handle)
	try:
		dev = _Device.from_device_number(_Context(), 'char', stat.st_rdev)
	except (DeviceNotFoundError, ValueError):
		return None

	hid_dev = dev.find_parent('hid')
	if hid_dev:
		assert 'HID_ID' in hid_dev
		bus, _ignore, _ignore = hid_dev['HID_ID'].split(':')

		if bus == '0003':  # USB
			usb_dev = dev.find_parent('usb', 'usb_device')
			assert usb_dev
			return usb_dev.attributes.get(key)

		elif bus == '0005':  # BLUETOOTH
			# TODO
			pass

    def test_events_real(self, context, monitor):
        # make sure that the module is unloaded initially
        pytest.unload_dummy()
        monitor.filter_by('net')
        monitor.start()
        self.prepare_test(monitor)
        # setup signal handlers
        event_callback = Mock(side_effect=self.stop_when_done)
        added_callback = Mock(side_effect=self.stop_when_done)
        removed_callback = Mock(side_effect=self.stop_when_done)
        self.connect_signal(event_callback)
        self.connect_signal(added_callback, action='add')
        self.connect_signal(removed_callback, action='remove')

        # test add event
        self.start_event_loop(pytest.load_dummy)
        device = Device.from_path(context, '/devices/virtual/net/dummy0')
        event_callback.assert_called_with('add', device)
        added_callback.assert_called_with(device)
        assert not removed_callback.called

        for mock in (event_callback, added_callback, removed_callback):
            mock.reset_mock()

        self.start_event_loop(pytest.unload_dummy)
        event_callback.assert_called_with('remove', device)
        assert not added_callback.called
        removed_callback.assert_called_with(device)

 def test_asstring(self, a_context, device_datum):
     """
     Test that string value agrees with cli value and is unicode.
     """
     device = Device.from_path(a_context, device_datum.device_path)
     for key, value in non_volatile_attributes(device_datum.attributes):
         assert is_unicode_string(device.attributes.asstring(key))
         assert device.attributes.asstring(key) == value

 def test_device_ordering(self, context, operator):
     try:
         device = Device.from_path(context, "/devices/platform")
     except DeviceNotFoundAtPathError:
         pytest.skip("device not found")
     with pytest.raises(TypeError) as exc_info:
         operator(device, device)
     assert str(exc_info.value) == "Device not orderable"

 def test_from_device_file_links(self, context, device_data):
     if not device_data.device_links:
         pytest.skip("no device links")
     for link in device_data.device_links:
         link = os.path.join(context.device_path, link)
         device = Device.from_device_file(context, link)
         assert device.device_path == device_data.device_path
         assert link in device.device_links

 def test_match_parent(self, context, device_data):
     device = Device.from_path(context, device_data.device_path)
     parent = device.parent
     if parent is None:
         pytest.skip('Device {0!r} has no parent'.format(device))
     else:
         children = list(context.list_devices().match_parent(parent))
         assert device in children
         assert parent in children

 def test_from_device_number(self, context, device_data):
     if not device_data.device_node:
         pytest.skip("no device node, no device number")
     mode = os.stat(device_data.device_node).st_mode
     type = "block" if stat.S_ISBLK(mode) else "char"
     device = Device.from_device_number(context, type, device_data.device_number)
     assert device.device_number == device_data.device_number
     # make sure, we are really referring to the same device
     assert device.device_path == device_data.device_path

 def test_getitem(self, a_context, device_datum):
     """
     Test that attribute value is the same as datum attribute value and
     is instance of bytes.
     """
     device = Device.from_path(a_context, device_datum.device_path)
     for key, value in non_volatile_attributes(device_datum.attributes):
         raw_value = value.encode(sys.getfilesystemencoding())
         assert isinstance(device.attributes.get(key), bytes)
         assert device.attributes.get(key) == raw_value

 def test_asint(self, a_context, device_datum):
     device = Device.from_path(a_context, device_datum.device_path)
     for key, value in self.non_volatile_items(device_datum.attributes):
         try:
             value = int(value)
         except ValueError:
             with pytest.raises(ValueError):
                 device.attributes.asint(key)
         else:
             assert device.attributes.asint(key) == value

 def test_from_device_number(self, context, device_path, device_number,
                             device_node):
     if not device_node:
         pytest.skip('no device node, no device number')
     mode = os.stat(device_node).st_mode
     type = 'block' if stat.S_ISBLK(mode) else 'char'
     device = Device.from_device_number(context, type, device_number)
     assert device.device_number == device_number
     # make sure, we are really referring to the same device
     assert device.device_path == device_path

 def test_asbool(self, a_context, device_datum):
     device = Device.from_path(a_context, device_datum.device_path)
     for key, value in self.non_volatile_items(device_datum.attributes):
         if value == '1':
             assert device.attributes.asbool(key)
         elif value == '0':
             assert not device.attributes.asbool(key)
         else:
             with pytest.raises(ValueError) as exc_info:
                 device.attributes.asbool(key)
             message = 'Not a boolean value:'
             assert str(exc_info.value).startswith(message)

 def test_asint(self, a_context, device_datum):
     """
     Test that integer result is an int or ValueError raised.
     """
     device = Device.from_path(a_context, device_datum.device_path)
     for key, value in non_volatile_attributes(device_datum.attributes):
         try:
             value = int(value)
         except ValueError:
             with pytest.raises(ValueError):
                 device.attributes.asint(key)
         else:
             assert device.attributes.asint(key) == value

 def test_from_device_number_wrong_type(self, context, device_data):
     if not device_data.device_node:
         pytest.skip("no device node, no device number")
     mode = os.stat(device_data.device_node).st_mode
     # deliberately use the wrong type here to cause either failure or at
     # least device mismatch
     type = "char" if stat.S_ISBLK(mode) else "block"
     try:
         # this either fails, in which case the caught exception is raised,
         # or succeeds, but returns a wrong device (device numbers are not
         # unique across device types)
         device = Device.from_device_number(context, type, device_data.device_number)
         # if it succeeds, the resulting device must not match the one, we
         # are actually looking for!
         assert device.device_path != device_data.device_path
     except DeviceNotFoundByNumberError as error:
         # check the correctness of the exception attributes
         assert error.device_type == type
         assert error.device_number == device_data.device_number

    def test_events_real(self, context, monitor):
        # make sure that the module is unloaded initially
        pytest.unload_dummy()
        monitor.filter_by('net')
        monitor.start()
        self.prepare_test(monitor)
        # setup signal handlers
        event_callback = mock.Mock(
            side_effect=lambda *args: self.stop_event_loop())
        self.connect_signal(event_callback)

        # test add event
        self.start_event_loop(pytest.load_dummy)
        device = Device.from_path(context, '/devices/virtual/net/dummy0')
        event_callback.assert_called_with(device)

        event_callback.reset_mock()

        self.start_event_loop(pytest.unload_dummy)
        event_callback.assert_called_with(device)