tests: cover GPIO digital input support

Document GpioDigitalInputDriver and add coverage for sysfsgpio input
direction handling, cached direction reconfiguration, input driver
reads, protocol compatibility, and labgrid-client io get routing.

Co-developed-by: Felix Zwettler <Felix.Zwettler@duagon.com>
Signed-off-by: Felix Zwettler <Felix.Zwettler@duagon.com>
Signed-off-by: Ozan Durgut <ozan.durgut@analog.com>

Author: ozan956

doc/configuration.rst

@@ -651,6 +651,7 @@ Arguments:
   - invert (bool, default=False): optional, whether the logic level is inverted (active-low)
 
 Used by:
+  - `GpioDigitalInputDriver`_
   - `GpioDigitalOutputDriver`_
 
 NetworkSysfsGPIO
@@ -684,6 +685,7 @@ Arguments:
   - invert (bool, default=False): optional, whether the logic level is inverted (active-low)
 
 Used by:
+  - `GpioDigitalInputDriver`_
   - `GpioDigitalOutputDriver`_
 
 NetworkService
@@ -2559,6 +2561,30 @@ Implements:
 Arguments:
   - delay (float, default=2.0): delay in seconds between off and on
 
+GpioDigitalInputDriver
+~~~~~~~~~~~~~~~~~~~~~~
+The :any:`GpioDigitalInputDriver` reads a digital signal from a GPIO line.
+
+This driver configures GPIO lines via
+`the sysfs kernel interface <https://www.kernel.org/doc/html/latest/gpio/sysfs.html>`__
+as an input.
+
+Binds to:
+  gpio:
+    - `SysfsGPIO`_
+    - `MatchedSysfsGPIO`_
+    - `NetworkSysfsGPIO`_
+
+Implements:
+  - :any:`DigitalInputProtocol`
+
+.. code-block:: yaml
+
+   GpioDigitalInputDriver: {}
+
+Arguments:
+  - None
+
 GpioDigitalOutputDriver
 ~~~~~~~~~~~~~~~~~~~~~~~
 The :any:`GpioDigitalOutputDriver` writes a digital signal to a GPIO line.

tests/test_gpiodriver.py

@@ -0,0 +1,92 @@
+import types
+
+import labgrid.driver.gpiodriver as gpiodriver
+from labgrid.driver.gpiodriver import GpioDigitalInputDriver, GpioDigitalOutputDriver
+from labgrid.remote.client import ClientSession
+from labgrid.resource.common import ResourceManager
+from labgrid.resource.remote import NetworkSysfsGPIO
+from labgrid.resource import SysfsGPIO
+
+
+class FakeWrapper:
+    def __init__(self, host, proxy):
+        self.host = host
+        self.proxy = proxy
+
+    def load(self, name):
+        assert name == 'sysfsgpio'
+        return self.proxy
+
+    def close(self):
+        pass
+
+
+def test_gpio_input_driver_get(target, monkeypatch):
+    proxy = types.SimpleNamespace(calls=[])
+
+    def proxy_get(index, invert, direction):
+        proxy.calls.append((index, invert, direction))
+        return True
+
+    proxy.get = proxy_get
+
+    monkeypatch.setattr(gpiodriver, "AgentWrapper", lambda host: FakeWrapper(host, proxy))
+
+    SysfsGPIO(target, name=None, index=13, invert=True)
+    driver = GpioDigitalInputDriver(target, name=None)
+
+    target.activate(driver)
+
+    assert driver.get() is True
+    assert proxy.calls == [(13, True, 'in')]
+
+    target.deactivate(driver)
+
+
+def test_gpio_output_driver_implements_digital_input_protocol(target):
+    SysfsGPIO(target, name=None, index=13, invert=False)
+    driver = GpioDigitalOutputDriver(target, name=None)
+
+    assert target.get_driver("DigitalInputProtocol", activate=False) is driver
+
+
+def test_client_io_get_uses_configured_gpio_input_driver(target, monkeypatch, capsys):
+    proxy = types.SimpleNamespace(calls=[])
+
+    def proxy_get(index, invert, direction):
+        proxy.calls.append((index, invert, direction))
+        return True
+
+    proxy.get = proxy_get
+
+    monkeypatch.setattr(gpiodriver, "AgentWrapper", lambda host: FakeWrapper(host, proxy))
+
+    SysfsGPIO(target, name="gpio_in", index=13, invert=False)
+    GpioDigitalInputDriver(target, name="gpio_in")
+
+    session = object.__new__(ClientSession)
+    session.args = types.SimpleNamespace(action="get", name="gpio_in")
+    session.get_acquired_place = lambda: types.SimpleNamespace(name="test")
+    session._get_target = lambda place: target
+
+    session.digital_io()
+
+    assert "digital IO gpio_in for place test is high" in capsys.readouterr().out
+    assert proxy.calls == [(13, False, 'in')]
+
+
+def test_client_io_get_keeps_network_sysfs_output_fallback(target, monkeypatch, mocker):
+    monkeypatch.setattr(NetworkSysfsGPIO, "manager_cls", ResourceManager)
+    driver = types.SimpleNamespace(get=mocker.MagicMock(return_value=False))
+    session = object.__new__(ClientSession)
+    session.args = types.SimpleNamespace(action="get", name="gpio")
+    session.get_acquired_place = lambda: types.SimpleNamespace(name="test")
+    session._get_target = lambda place: target
+    session._get_driver_or_new = mocker.MagicMock(return_value=driver)
+
+    NetworkSysfsGPIO(target, name="gpio", host="exporter", index=13, invert=False)
+
+    session.digital_io()
+
+    session._get_driver_or_new.assert_called_once_with(target, "GpioDigitalOutputDriver", name="gpio")
+    driver.get.assert_called_once_with()

tests/test_sysfsgpioagent.py

@@ -1,6 +1,7 @@
 import pytest
 
 import os
+from labgrid.util.agents import sysfsgpio
 from labgrid.util.agents.sysfsgpio import GpioDigitalOutput
 from tempfile import TemporaryDirectory
 
@@ -11,7 +12,6 @@ def __init__(self, **kwargs):
             index = kwargs['index']
             self.sysfs_mock_directory = TemporaryDirectory()
             GpioDigitalOutput._gpio_sysfs_path_prefix = self.sysfs_mock_directory.name
-            GpioDigitalOutput._buffered_file_access = True
             export_file_path = os.path.join(self.sysfs_mock_directory.name, 'export')
             os.mknod(export_file_path)
             # Since there is no real device, writing to `export` does not create a corresponding
@@ -23,7 +23,7 @@ def __init__(self, **kwargs):
                 assert export_content == str(index)
             self.gpio_line_directory = os.path.join(self.sysfs_mock_directory.name, f'gpio{index}')
             os.mkdir(self.gpio_line_directory)
-            for control_file in ['direction', 'value']:
+            for control_file in ['active_low', 'direction', 'value']:
                 control_file_path = os.path.join(self.gpio_line_directory, control_file)
                 print(f'creating control file `{control_file_path}`')
                 os.mknod(control_file_path)
@@ -44,3 +44,43 @@ def test_set(self):
         for val in [True, False, True, False]:
             gpio_line.set(val)
             assert gpio_line.get() == val
+
+    def test_output_direction(self):
+        gpio_line = TestGpioAgent.GpioDigitalOutputMock(index=13, invert=False)
+        direction_file_path = os.path.join(gpio_line.gpio_line_directory, 'direction')
+        with open(direction_file_path, mode='rb') as direction_file:
+            assert direction_file.read() == b'out'
+
+    def test_input_direction(self):
+        gpio_line = TestGpioAgent.GpioDigitalOutputMock(index=13, invert=False, direction='in')
+        direction_file_path = os.path.join(gpio_line.gpio_line_directory, 'direction')
+        value_file_path = os.path.join(gpio_line.gpio_line_directory, 'value')
+
+        with open(direction_file_path, mode='rb') as direction_file:
+            assert direction_file.read() == b'in'
+
+        with open(value_file_path, mode='wb') as value_file:
+            value_file.write(b'1')
+
+        assert gpio_line.get() is True
+
+    def test_invalid_direction(self):
+        with pytest.raises(ValueError, match='direction'):
+            GpioDigitalOutput(index=13, invert=False, direction='invalid')
+
+    def test_cached_line_reconfigures_direction(self):
+        with TemporaryDirectory() as sysfs_mock_directory:
+            GpioDigitalOutput._gpio_sysfs_path_prefix = sysfs_mock_directory
+            sysfsgpio._gpios.clear()
+            gpio_line_directory = os.path.join(sysfs_mock_directory, 'gpio13')
+            os.mkdir(gpio_line_directory)
+            for control_file in ['active_low', 'direction', 'value']:
+                os.mknod(os.path.join(gpio_line_directory, control_file))
+
+            sysfsgpio.handle_set(13, False, True)
+            sysfsgpio.handle_get(13, False, 'in')
+            sysfsgpio.handle_set(13, False, False)
+
+            direction_file_path = os.path.join(gpio_line_directory, 'direction')
+            with open(direction_file_path, mode='rb') as direction_file:
+                assert direction_file.read() == b'out'