Merge pull request #1523 from Bastian-Krause/bst/ruff

ruff: maintain configuration, format examples and labgrid.protocols

Author: Emantor

.github/workflows/reusable-unit-tests.yml

@@ -55,7 +55,7 @@ jobs:
         pylint labgrid
     - name: Format with ruff
       run: |
-        ruff format --check --diff labgrid/remote/
+        ruff format --check --diff
     - name: Test with pytest
       run: |
         pytest -r a --cov-config .coveragerc --cov=labgrid --local-sshmanager --ssh-username runner -k "not test_docker_with_daemon"

examples/barebox/conftest.py

@@ -1,8 +1,8 @@
 import pytest
 
 
-@pytest.fixture(scope='session')
+@pytest.fixture(scope="session")
 def command(target):
-    barebox = target.get_driver('CommandProtocol')
+    barebox = target.get_driver("CommandProtocol")
     target.activate(barebox)
     return barebox

examples/barebox/test_barebox.py

@@ -1,11 +1,11 @@
 def test_barebox(command):
-    stdout, stderr, returncode = command.run('version')
+    stdout, stderr, returncode = command.run("version")
     assert returncode == 0
     assert stdout
     assert not stderr
-    assert 'barebox' in '\n'.join(stdout)
+    assert "barebox" in "\n".join(stdout)
 
-    stdout, stderr, returncode = command.run('false')
+    stdout, stderr, returncode = command.run("false")
     assert returncode == 1
     assert not stdout
     assert not stderr

examples/barebox/test_bootchooser.py

@@ -2,10 +2,10 @@
 
 
 def test_bootchooser(command):
-    stdout, stderr, returncode = command.run('bootchooser -i')
+    stdout, stderr, returncode = command.run("bootchooser -i")
     if returncode == 127:
         pytest.skip("bootchooser command not available")
     assert returncode == 0
     assert not stderr
-    assert stdout[0].startswith('Good targets')
-    assert stdout[1] != 'none'
+    assert stdout[0].startswith("Good targets")
+    assert stdout[1] != "none"

examples/barebox/test_sleep.py

@@ -6,14 +6,14 @@
 def test_sleep(command):
     # measure the round-trip-time
     timestamp = monotonic()
-    stdout, stderr, returncode = command.run('true')
+    stdout, stderr, returncode = command.run("true")
     elapsed_true = monotonic() - timestamp
     assert returncode == 0
     assert not stdout
     assert not stderr
 
     timestamp = monotonic()
-    stdout, stderr, returncode = command.run('sleep 1')
+    stdout, stderr, returncode = command.run("sleep 1")
     elapsed_sleep = monotonic() - timestamp
     assert returncode == 0
     assert not stdout

examples/barebox/test_state.py

@@ -2,10 +2,10 @@
 
 
 def test_state(command):
-    stdout, stderr, returncode = command.run('state')
+    stdout, stderr, returncode = command.run("state")
     if returncode == 127:
         pytest.skip("state command not available")
     assert returncode == 0
     assert not stderr
-    assert stdout[0] == 'registered state instances:'
+    assert stdout[0] == "registered state instances:"
     assert len(stdout) > 1

examples/barebox/test_watchdog.py

@@ -2,7 +2,7 @@
 
 
 def test_watchdog(command):
-    stdout, stderr, returncode = command.run('wd 1')
+    stdout, stderr, returncode = command.run("wd 1")
     if returncode == 127:
         pytest.skip("wd command not available")
     assert returncode == 0
@@ -11,6 +11,6 @@ def test_watchdog(command):
 
     command._await_prompt()
 
-    stdout = command.run_check('echo ${global.system.reset}')
+    stdout = command.run_check("echo ${global.system.reset}")
     assert len(stdout) == 1
-    assert stdout[0] == 'WDG'
+    assert stdout[0] == "WDG"

examples/deditec-relais8/deditec.py

@@ -12,7 +12,7 @@
 # log labgrid steps
 StepLogger.start()
 
-t = Target('main')
+t = Target("main")
 r = DeditecRelais8(t, name=None, index=1)
 d = DeditecRelaisDriver(t, name=None)
 

examples/deditec-relais8/deditec_remote.py

@@ -10,7 +10,7 @@
 # show labgrid steps on the console
 StepLogger.start()
 
-e = Environment('import-dedicontrol.yaml')
+e = Environment("import-dedicontrol.yaml")
 t = e.get_target()
 
 p = t.get_driver("DigitalOutputProtocol")

examples/docker/conftest.py

@@ -1,9 +1,9 @@
 import pytest
 
-@pytest.fixture(scope='session')
+
+@pytest.fixture(scope="session")
 def command(target):
-    strategy = target.get_driver('DockerStrategy')
+    strategy = target.get_driver("DockerStrategy")
     strategy.transition("accessible")
-    shell = target.get_driver('CommandProtocol')
+    shell = target.get_driver("CommandProtocol")
     return shell
-

examples/docker/test_shell.py

@@ -1,11 +1,11 @@
 def test_shell(command):
-    stdout, stderr, returncode = command.run('cat /proc/version')
+    stdout, stderr, returncode = command.run("cat /proc/version")
     assert returncode == 0
     assert len(stdout) > 0
     assert len(stderr) == 0
-    assert 'Linux' in stdout[0]
+    assert "Linux" in stdout[0]
 
-    stdout, stderr, returncode = command.run('false')
+    stdout, stderr, returncode = command.run("false")
     assert returncode != 0
     assert len(stdout) == 0
     assert len(stderr) == 0

examples/library/test.py

@@ -15,15 +15,17 @@
 # log labgrid steps
 StepLogger.start()
 
+
 def run_once(target):
-    s = target.get_driver('BareboxStrategy')
+    s = target.get_driver("BareboxStrategy")
     s.status = Status.unknown  # force a power-cycle
-    s.transition('barebox')
-    cmd = target['CommandProtocol']
-    cmd.run_check('test -e /dev/nand0')
+    s.transition("barebox")
+    cmd = target["CommandProtocol"]
+    cmd.run_check("test -e /dev/nand0")
     target.deactivate(cmd)
 
+
 env = Environment(sys.argv[1])
-target = env.get_target('main')
+target = env.get_target("main")
 while True:
     run_once(target)

examples/modbusrtu/conftest.py

@@ -1,7 +1,7 @@
 import pytest
 
 
-@pytest.fixture(scope='session')
+@pytest.fixture(scope="session")
 def instrument(target):
-    _modbus = target.get_driver('ModbusRTUDriver')
+    _modbus = target.get_driver("ModbusRTUDriver")
     return _modbus

examples/network-test/pkg-replay-record.py

@@ -11,6 +11,7 @@
 from labgrid import Environment
 from labgrid.logging import basicConfig, StepLogger
 
+
 def generate_frame():
     frame = Ether(dst="11:22:33:44:55:66", src="66:55:44:33:22:11", type=0x9000)
     padding = "\x00" * (conf.min_pkt_size - len(frame))

examples/networkmanager/nm.py

@@ -12,72 +12,71 @@
 StepLogger.start()
 
 
-e = Environment('nm.env')
+e = Environment("nm.env")
 t = e.get_target()
-d = t.get_driver('NetworkInterfaceDriver')
+d = t.get_driver("NetworkInterfaceDriver")
 
 # based on https://developer.gnome.org/NetworkManager/stable/ch01.html, but adapted to python dicts
 s_client = {
-    'connection': {
-        'type': "802-11-wireless",
+    "connection": {
+        "type": "802-11-wireless",
     },
-    '802-11-wireless': {
-        'mode': "infrastructure",
-        'ssid': "local-rpi",
+    "802-11-wireless": {
+        "mode": "infrastructure",
+        "ssid": "local-rpi",
     },
-    '802-11-wireless-security': {
-        'key-mgmt': "wpa-psk",
-        'psk': "obMinwyurArc5",
+    "802-11-wireless-security": {
+        "key-mgmt": "wpa-psk",
+        "psk": "obMinwyurArc5",
     },
-    'ipv4': {
-        'method': "auto",
-        'ignore-auto-dns': True,
-        'ignore-auto-routes': True,
-        'never-default': True,
+    "ipv4": {
+        "method": "auto",
+        "ignore-auto-dns": True,
+        "ignore-auto-routes": True,
+        "never-default": True,
     },
-    'ipv6': {
-        'method': "link-local",
+    "ipv6": {
+        "method": "link-local",
     },
 }
 s_ap = {
-    'connection': {
-        'type': "802-11-wireless",
+    "connection": {
+        "type": "802-11-wireless",
     },
-    '802-11-wireless': {
-        'mode': "ap",
-        'ssid': "local-rpi",
+    "802-11-wireless": {
+        "mode": "ap",
+        "ssid": "local-rpi",
     },
-    '802-11-wireless-security': {
-        'key-mgmt': "wpa-psk",
-        'psk': "obMinwyurArc5",
+    "802-11-wireless-security": {
+        "key-mgmt": "wpa-psk",
+        "psk": "obMinwyurArc5",
     },
-    'ipv4': {
+    "ipv4": {
         #'method': "auto",
         #'method': "link-local",
-        'method': "shared",
-        'addresses': ["172.16.0.2/29"],
+        "method": "shared",
+        "addresses": ["172.16.0.2/29"],
     },
-    'ipv6': {
-        'method': "link-local",
+    "ipv6": {
+        "method": "link-local",
     },
 }
 
 d.disable()
-d.wait_state('disconnected')
+d.wait_state("disconnected")
 print("access points after scan")
 pprint(d.get_access_points())
 
 d.configure(s_ap)
-d.wait_state('activated')
+d.wait_state("activated")
 print("settings in AP mode")
 pprint(d.get_settings())
 print("state in AP mode")
 pprint(d.get_state())
 
-#d.configure(s_client)
-#d.wait_state('activated')
-#print("settings in client mode")
-#pprint(d.get_settings())
-#print("state in client mode")
-#pprint(d.get_state())
-
+# d.configure(s_client)
+# d.wait_state('activated')
+# print("settings in client mode")
+# pprint(d.get_settings())
+# print("state in client mode")
+# pprint(d.get_state())

examples/power/power_example.py

@@ -3,7 +3,7 @@
 
 @pytest.fixture()
 def pdu(target):
-    return target.get_driver('NetworkPowerDriver')
+    return target.get_driver("NetworkPowerDriver")
 
 
 def test_something(pdu):

examples/pyvisa/pyvisa_example.py

@@ -3,13 +3,13 @@
 
 @pytest.fixture()
 def signal_generator(target):
-    return target.get_driver('PyVISADriver').get_session()
+    return target.get_driver("PyVISADriver").get_session()
 
 
 def test_with_signal_generator_example(signal_generator):
-    signal_generator.write('*RST')
+    signal_generator.write("*RST")
 
     # Setup channel 1
-    signal_generator.write('C1:BSWV WVTP,SQUARE,HLEV,5,LLEV,0,DUTY,50')
+    signal_generator.write("C1:BSWV WVTP,SQUARE,HLEV,5,LLEV,0,DUTY,50")
     # Switch on channel 1
-    signal_generator.write('C1:OUTP ON,LOAD,HZ,PLRT,NOR')
+    signal_generator.write("C1:OUTP ON,LOAD,HZ,PLRT,NOR")

examples/qemu-networking/test_qemu_networking.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+
 def test_shell(shell_command):
     shell_command.run("true")
 

examples/remote/test_barebox.py

@@ -1,14 +1,14 @@
 def test_target(target):
-    barebox = target.get_driver('CommandProtocol')
+    barebox = target.get_driver("CommandProtocol")
     target.activate(barebox)
 
-    stdout, stderr, returncode = barebox.run('version')
+    stdout, stderr, returncode = barebox.run("version")
     assert returncode == 0
     assert stdout
     assert not stderr
-    assert 'barebox' in '\n'.join(stdout)
+    assert "barebox" in "\n".join(stdout)
 
-    stdout, stderr, returncode = barebox.run('false')
+    stdout, stderr, returncode = barebox.run("false")
     assert returncode == 1
     assert not stdout
     assert not stderr

examples/shell/conftest.py

@@ -1,8 +1,8 @@
 import pytest
 
 
-@pytest.fixture(scope='session')
+@pytest.fixture(scope="session")
 def command(target):
-    shell = target.get_driver('CommandProtocol')
+    shell = target.get_driver("CommandProtocol")
     target.activate(shell)
     return shell

examples/shell/test_hwclock.py

@@ -3,7 +3,7 @@
 
 def test_hwclock_rate(command):
     """Test that the hardware clock rate is not too inaccurate."""
-    result = command.run_check('hwclock -c | head -n 3')
+    result = command.run_check("hwclock -c | head -n 3")
     hw_time, sys_time, freq_offset_ppm, tick = result[-1].strip().split()
     assert abs(int(freq_offset_ppm)) < 1000
 
@@ -15,11 +15,11 @@ def test_hwclock_value(command):
     """
 
     def get_time():
-        result = command.run_check('hwclock --utc --show')[0].strip()
-        return datetime.strptime(result, '%Y-%m-%d %H:%M:%S.%f+0:00')
+        result = command.run_check("hwclock --utc --show")[0].strip()
+        return datetime.strptime(result, "%Y-%m-%d %H:%M:%S.%f+0:00")
 
     def set_time(time):
-        time = time.strftime('%Y-%m-%d %H:%M:%S.%f+0:00')
+        time = time.strftime("%Y-%m-%d %H:%M:%S.%f+0:00")
         command.run_check(f'hwclock --utc --set --date "{time}"')
 
     offset = abs((get_time() - datetime.utcnow()).total_seconds())