fix: Make arguments depend on uhubctl version

Refactor uhubctl wrapper code and introduce version check, which is used
for conditional arguments (some require a more recent version).

Signed-off-by: Nicolai Buchwitz <nb@tipi-net.de>

Author: nbuchwitz

tests/helper.py

@@ -70,6 +70,7 @@ def cmd(
             stdout += self.__stdout("New", port_filter=port_number)
 
         fp.register(cmd, stdout=stdout)
+        fp.register(["uhubctl", "-v"], stdout="2.4.0-43-ge1e4d450")
 
 
 @pytest.fixture

uhubctl/usb.py

@@ -2,7 +2,7 @@
 import re
 from typing import List, Optional
 
-from .utils import _uhubctl
+from .utils import UHubCtl
 
 
 def discover_hubs() -> List["Hub"]:
@@ -17,7 +17,7 @@ def discover_hubs() -> List["Hub"]:
 
     pattern = re.compile(r"Current status for hub ([\.\d-]+)")
 
-    for line in _uhubctl():
+    for line in UHubCtl.exec():
         regex = pattern.match(line)
 
         if regex:
@@ -96,7 +96,7 @@ def discover_ports(self) -> None:
         """
         pattern = re.compile(r"  Port (\d+): \d{4} ")
 
-        for line in _uhubctl(["-l", self.path]):
+        for line in UHubCtl.exec(["-l", self.path]):
             regex = pattern.match(line)
 
             if regex:
@@ -133,7 +133,7 @@ def status(self) -> bool:
         pattern = re.compile(rf"  Port {self.port_number}: \d{{4}} (power|off|indicator)")
 
         args = ["-l", self.hub.path, "-p", str(self.port_number)]
-        for line in _uhubctl(args):
+        for line in UHubCtl.exec(args):
             reg = pattern.match(line)
 
             if reg:
@@ -153,7 +153,7 @@ def status(self, status: bool) -> None:
         else:
             args.append("off")
 
-        _uhubctl(args)
+        UHubCtl.exec(args)
 
     @staticmethod
     def from_path(path: str):

uhubctl/utils.py

@@ -1,25 +1,45 @@
 """Utilities around uhubctl binary"""
 import subprocess
 
+from packaging import version
+
 UHUBCTL_BINARY = "uhubctl"
 
 
-def _uhubctl(args: list = None) -> list:
-    cmd = UHUBCTL_BINARY.split(" ")
-    cmd.append("-N")
+class UHubCtl:
+    _version = None
+
+    @classmethod
+    def version(cls) -> str:
+        if cls._version is None:
+            cmd = UHUBCTL_BINARY.split(" ")
+            cmd.append("-v")
+
+            result = subprocess.run(cmd, capture_output=True, text=True, check=True)
+
+            cls._version = result.stdout.split("-", maxsplit=1)[0]
+
+        return cls._version
+
+    @classmethod
+    def exec(cls, args: list = None) -> list:
+        cmd = UHUBCTL_BINARY.split(" ")
+
+        if version.parse(cls.version()) > version.parse("2.3.0"):
+            cmd.append("-N")
 
-    if args is not None:
-        cmd += args
+        if args is not None:
+            cmd += args
 
-    try:
-        result = subprocess.run(cmd, capture_output=True, check=True)
-        stdout = result.stdout.decode()
+        try:
+            result = subprocess.run(cmd, capture_output=True, check=True)
+            stdout = result.stdout.decode()
 
-        return stdout.split("\n")
-    except subprocess.CalledProcessError as exc:
-        stderr = exc.stderr.decode()
+            return stdout.split("\n")
+        except subprocess.CalledProcessError as exc:
+            stderr = exc.stderr.decode()
 
-        if stderr.startswith("No compatible devices detected"):
-            return []
+            if stderr.startswith("No compatible devices detected"):
+                return []
 
-        raise Exception(f"uhubctl failed: {stderr}") from exc
+            raise Exception(f"uhubctl failed: {stderr}") from exc