Removed LSBLK dependency, fixed progress bar. (Issue #178, partially #171)

.github/workflows/ci.yml

@@ -34,7 +34,7 @@ jobs:
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          pip install psutil pytest
+          pip install psutil pyudev pytest
 
       - name: Install package (no-deps to avoid GUI reqs)
         run: pip install -e . --no-deps

.github/workflows/python-compatibility.yml

@@ -9,12 +9,12 @@ on:
 
 jobs:
   test:
-    name: "Test — Python ${{ matrix.python-version }}"
+    name: "Run Tests (Python 3.10+)"
     runs-on: ubuntu-latest
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.10", "3.11", "3.12", "3.13", "3.14-dev"]
+        python-version: ["3.10", "3.13"]
 
     steps:
       - uses: actions/checkout@v4
@@ -24,25 +24,12 @@ jobs:
         with:
           python-version: ${{ matrix.python-version }}
 
-      - name: Cache pip
-        uses: actions/cache@v4
-        with:
-          path: ~/.cache/pip
-          key: ${{ runner.os }}-pip-${{ matrix.python-version }}-${{ hashFiles('pyproject.toml') }}
-          restore-keys: |
-            ${{ runner.os }}-pip-${{ matrix.python-version }}-
-
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          pip install psutil pytest packaging platformdirs
-
-      - name: Install package (no-deps to avoid GUI reqs)
-        run: pip install -e . --no-deps
+          pip install -r requirements-python.txt
+          pip install pytest
 
       - name: Run tests
         run: |
-          pytest --maxfail=1 -q \
-            --ignore=src/lufus/gui \
-            --ignore=src/lufus/drives/autodetect_usb.py \
-            tests/
+          pytest --maxfail=1 -q tests/

README.md

@@ -11,7 +11,7 @@
 
 ## Beta Release Disclaimer
 
-**Lufus** is currently in **Beta**. 
+**Lufus** is currently in **Beta**! 
 
 Lufus is a physical drive imaging and formatting utility written in Python, inspired by **Rufus** on Windows, with the goal of delivering a greater experience for Linux users.
 While core functionality has been implemented, the project is still under active development. Users should expect bugs, incomplete features, and ongoing structural changes.

src/lufus/drives/find_usb.py

@@ -1,6 +1,6 @@
 import psutil
 import os
-import subprocess
+import pyudev
 import getpass
 from lufus import state
 from lufus.lufus_logging import get_logger
@@ -43,6 +43,7 @@ def find_usb() -> dict[str, str]:
     dir_set = set(all_directories)
 
     # Check each partition to see if it matches our potential mount points
+    context = pyudev.Context()
     for part in psutil.disk_partitions(all=True):
         if part.mountpoint not in dir_set:
             continue
@@ -53,12 +54,11 @@ def find_usb() -> dict[str, str]:
 
         label = None
         try:
-            label = subprocess.check_output(
-                ["lsblk", "-d", "-n", "-o", "LABEL", device_node],
-                text=True,
-                timeout=5,
-            ).strip()
-        except (subprocess.CalledProcessError, subprocess.TimeoutExpired):
+            # Using os.stat to get device number as per requirements
+            st = os.stat(device_node)
+            device = pyudev.Devices.from_device_number(context, "block", st.st_rdev)
+            label = device.get("ID_FS_LABEL")
+        except Exception:
             pass
 
         if not label:

src/lufus/drives/get_usb_info.py

@@ -1,6 +1,6 @@
 import psutil
 import os
-import subprocess
+import pyudev
 from typing import TypedDict
 from lufus.lufus_logging import get_logger
 
@@ -25,20 +25,20 @@ def get_usb_info(usb_path: str) -> USBDeviceInfo | None:
             log.warning("Could not find device node for USB path: %s", usb_path)
             return None
 
-        size_output = subprocess.check_output(
-            ["lsblk", "-d", "-n", "-b", "-o", "SIZE", device_node],
-            text=True,
-            timeout=5,
-        ).strip()
+        context = pyudev.Context()
+        # Using os.stat to get device number as per requirements
+        st = os.stat(device_node)
+        device = pyudev.Devices.from_device_number(context, "block", st.st_rdev)
 
-        usb_size = int(size_output) if size_output.isdigit() else 0
-        if not size_output.isdigit():
-            log.warning("Could not parse device size: %r", size_output)
+        # Size in bytes: udev attributes 'size' is in 512-byte sectors
+        size_attr = device.attributes.get("size")
+        usb_size = int(size_attr) * 512 if size_attr else 0
+
+        label = device.get("ID_FS_LABEL")
 
         if usb_size > 32 * 1024**3:
             log.warning("USB device is large (%d bytes); confirm before flashing.", usb_size)
 
-        label = subprocess.check_output(["lsblk", "-d", "-n", "-o", "LABEL", device_node], text=True, timeout=5).strip()
         if not label:
             label = os.path.basename(usb_path)
 
@@ -49,15 +49,9 @@ def get_usb_info(usb_path: str) -> USBDeviceInfo | None:
         }
         log.info("USB Info: %s", usb_info)
         return usb_info
-    except subprocess.TimeoutExpired as e:
-        log.error("Timed out getting USB info for %s: %s", usb_path, e)
-        return None
     except PermissionError:
         log.error("Permission denied when trying to get USB info: %s", usb_path)
         return None
-    except subprocess.CalledProcessError as e:
-        log.error("Error getting USB info: %s", e)
-        return None
     except Exception as err:
         log.error("Unexpected error getting USB info: %s", err)
         return None

tests/test_find_usb.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
-import subprocess
 import sys
+import os
 from pathlib import Path
 from types import SimpleNamespace
+from unittest.mock import MagicMock
 
 ROOT = Path(__file__).resolve().parents[1]
 SRC = ROOT / "src"
@@ -15,6 +16,7 @@
 def test_find_usb_returns_mount_to_label_mapping(monkeypatch) -> None:
     user = "testuser"
     mount_path = f"/media/{user}/MY_USB"
+    device_node = "/dev/sdb1"
 
     monkeypatch.setattr(find_usb_module.getpass, "getuser", lambda: user)
     monkeypatch.setattr(
@@ -35,21 +37,36 @@ def test_find_usb_returns_mount_to_label_mapping(monkeypatch) -> None:
     monkeypatch.setattr(
         find_usb_module.psutil,
         "disk_partitions",
-        lambda *args, **kwargs: [SimpleNamespace(mountpoint=mount_path, device="/dev/sdb1")],
-    )
-    monkeypatch.setattr(
-        find_usb_module.subprocess,
-        "check_output",
-        lambda *args, **kwargs: "lufus_USB\n",
+        lambda *args, **kwargs: [SimpleNamespace(mountpoint=mount_path, device=device_node)],
     )
 
+    # Mock os.stat safely
+    os_stat_orig = os.stat
+
+    def mock_os_stat(p):
+        if str(p).startswith("/dev/"):
+            mock_stat = MagicMock()
+            mock_stat.st_rdev = 1234
+            return mock_stat
+        return os_stat_orig(p)
+
+    monkeypatch.setattr(find_usb_module.os, "stat", mock_os_stat)
+
+    # Mock pyudev
+    mock_context = MagicMock()
+    mock_device = MagicMock()
+    mock_device.get.return_value = "lufus_USB"
+    monkeypatch.setattr(find_usb_module.pyudev, "Context", lambda: mock_context)
+    monkeypatch.setattr(find_usb_module.pyudev.Devices, "from_device_number", lambda ctx, type, num: mock_device)
+
     result = find_usb_module.find_usb()
     assert result == {mount_path: "lufus_USB"}
 
 
-def test_find_usb_falls_back_to_dir_name_when_lsblk_fails(monkeypatch) -> None:
+def test_find_usb_falls_back_to_dir_name_when_pyudev_fails(monkeypatch) -> None:
     user = "testuser"
     mount_path = f"/media/{user}/NO_LABEL"
+    device_node = "/dev/sdc1"
 
     monkeypatch.setattr(find_usb_module.getpass, "getuser", lambda: user)
     monkeypatch.setattr(
@@ -70,13 +87,27 @@ def test_find_usb_falls_back_to_dir_name_when_lsblk_fails(monkeypatch) -> None:
     monkeypatch.setattr(
         find_usb_module.psutil,
         "disk_partitions",
-        lambda *args, **kwargs: [SimpleNamespace(mountpoint=mount_path, device="/dev/sdc1")],
+        lambda *args, **kwargs: [SimpleNamespace(mountpoint=mount_path, device=device_node)],
     )
 
-    def raise_lsblk_error(*args, **kwargs):
-        raise subprocess.CalledProcessError(returncode=1, cmd="lsblk")
+    # Mock os.stat safely
+    os_stat_orig = os.stat
+
+    def mock_os_stat(p):
+        if str(p).startswith("/dev/"):
+            mock_stat = MagicMock()
+            mock_stat.st_rdev = 5678
+            return mock_stat
+        return os_stat_orig(p)
+
+    monkeypatch.setattr(find_usb_module.os, "stat", mock_os_stat)
 
-    monkeypatch.setattr(find_usb_module.subprocess, "check_output", raise_lsblk_error)
+    # Mock pyudev to fail
+    mock_context = MagicMock()
+    monkeypatch.setattr(find_usb_module.pyudev, "Context", lambda: mock_context)
+    monkeypatch.setattr(
+        find_usb_module.pyudev.Devices, "from_device_number", MagicMock(side_effect=Exception("udev fail"))
+    )
 
     result = find_usb_module.find_usb()
     assert result == {mount_path: "NO_LABEL"}

tests/test_flash_usb_and_find_usb_fixes.py

@@ -285,9 +285,10 @@ def test_empty_device_node_does_not_overwrite_states_dn(self, monkeypatch):
 
 
 class TestFindUsbHappyPath:
-    def test_find_usb_returns_label_from_lsblk(self, monkeypatch):
+    def test_find_usb_returns_label_from_udev(self, monkeypatch):
         user = "testuser"
         mount_path = f"/media/{user}/MY_USB"
+        device_node = "/dev/sdb1"
 
         monkeypatch.setattr(find_usb_module.getpass, "getuser", lambda: user)
         monkeypatch.setattr(
@@ -308,14 +309,29 @@ def test_find_usb_returns_label_from_lsblk(self, monkeypatch):
         monkeypatch.setattr(
             find_usb_module.psutil,
             "disk_partitions",
-            lambda *args, **kwargs: [SimpleNamespace(mountpoint=mount_path, device="/dev/sdb1")],
-        )
-        monkeypatch.setattr(
-            find_usb_module.subprocess,
-            "check_output",
-            lambda *a, **kw: "MY_LABEL\n",
+            lambda *args, **kwargs: [SimpleNamespace(mountpoint=mount_path, device=device_node)],
         )
 
+        import os as real_os
+        from unittest.mock import MagicMock
+
+        os_stat_orig = find_usb_module.os.stat
+
+        def mock_os_stat(p):
+            if str(p).startswith("/dev/"):
+                m = MagicMock()
+                m.st_rdev = 1234
+                return m
+            return os_stat_orig(p)
+
+        monkeypatch.setattr(find_usb_module.os, "stat", mock_os_stat)
+
+        mock_context = MagicMock()
+        mock_device = MagicMock()
+        mock_device.get.return_value = "MY_LABEL"
+        monkeypatch.setattr(find_usb_module.pyudev, "Context", lambda: mock_context)
+        monkeypatch.setattr(find_usb_module.pyudev.Devices, "from_device_number", lambda ctx, type, num: mock_device)
+
         result = find_usb_module.find_usb()
         assert result == {mount_path: "MY_LABEL"}
 

tests/test_get_usb_info.py

@@ -1,8 +1,8 @@
 from __future__ import annotations
-import subprocess
 import sys
 from pathlib import Path
 from types import SimpleNamespace
+from unittest.mock import MagicMock
 
 ROOT = Path(__file__).resolve().parents[1]
 SRC = ROOT / "src"
@@ -32,14 +32,25 @@ def test_get_usb_info_returns_expected_dictionary(monkeypatch) -> None:
         lambda *args, **kwargs: [SimpleNamespace(mountpoint=mount_path, device=device_node)],
     )
 
-    def fake_check_output(cmd, text=True, timeout=5):
-        if cmd[-2:] == ["SIZE", device_node]:
-            return str(16 * 1024**3)
-        if cmd[-2:] == ["LABEL", device_node]:
-            return "MYUSB\n"
-        raise AssertionError(f"Unexpected command: {cmd}")
+    # Mock os.stat safely
+    os_stat_orig = get_usb_info_module.os.stat
 
-    monkeypatch.setattr(get_usb_info_module.subprocess, "check_output", fake_check_output)
+    def mock_os_stat(p):
+        if str(p).startswith("/dev/"):
+            mock_stat = MagicMock()
+            mock_stat.st_rdev = 1234
+            return mock_stat
+        return os_stat_orig(p)
+
+    monkeypatch.setattr(get_usb_info_module.os, "stat", mock_os_stat)
+
+    # Mock pyudev
+    mock_context = MagicMock()
+    mock_device = MagicMock()
+    mock_device.attributes = {"size": str((16 * 1024**3) // 512)}
+    mock_device.get.return_value = "MYUSB"
+    monkeypatch.setattr(get_usb_info_module.pyudev, "Context", lambda: mock_context)
+    monkeypatch.setattr(get_usb_info_module.pyudev.Devices, "from_device_number", lambda ctx, type, num: mock_device)
 
     result = get_usb_info_module.get_usb_info(mount_path)
     assert result == {
@@ -59,20 +70,31 @@ def test_get_usb_info_uses_mount_basename_when_label_is_empty(monkeypatch) -> No
         lambda *args, **kwargs: [SimpleNamespace(mountpoint=mount_path, device=device_node)],
     )
 
-    def fake_check_output(cmd, text=True, timeout=5):
-        if cmd[-2:] == ["SIZE", device_node]:
-            return str(8 * 1024**3)
-        if cmd[-2:] == ["LABEL", device_node]:
-            return "\n"
-        raise AssertionError(f"Unexpected command: {cmd}")
+    # Mock os.stat safely
+    os_stat_orig = get_usb_info_module.os.stat
+
+    def mock_os_stat(p):
+        if str(p).startswith("/dev/"):
+            mock_stat = MagicMock()
+            mock_stat.st_rdev = 5678
+            return mock_stat
+        return os_stat_orig(p)
 
-    monkeypatch.setattr(get_usb_info_module.subprocess, "check_output", fake_check_output)
+    monkeypatch.setattr(get_usb_info_module.os, "stat", mock_os_stat)
+
+    # Mock pyudev
+    mock_context = MagicMock()
+    mock_device = MagicMock()
+    mock_device.attributes = {"size": str((8 * 1024**3) // 512)}
+    mock_device.get.return_value = None
+    monkeypatch.setattr(get_usb_info_module.pyudev, "Context", lambda: mock_context)
+    monkeypatch.setattr(get_usb_info_module.pyudev.Devices, "from_device_number", lambda ctx, type, num: mock_device)
 
     result = get_usb_info_module.get_usb_info(mount_path)
     assert result["label"] == "NO_LABEL"
 
 
-def test_get_usb_info_returns_empty_when_lsblk_fails(monkeypatch) -> None:
+def test_get_usb_info_returns_empty_when_pyudev_fails(monkeypatch) -> None:
     mount_path = "/media/testuser/USB"
     device_node = "/dev/sdb1"
 
@@ -82,9 +104,15 @@ def test_get_usb_info_returns_empty_when_lsblk_fails(monkeypatch) -> None:
         lambda *args, **kwargs: [SimpleNamespace(mountpoint=mount_path, device=device_node)],
     )
 
-    def raise_lsblk_error(*args, **kwargs):
-        raise subprocess.CalledProcessError(returncode=1, cmd="lsblk")
+    # Mock os.stat safely
+    os_stat_orig = get_usb_info_module.os.stat
+
+    def mock_os_stat(p):
+        if str(p).startswith("/dev/"):
+            raise Exception("stat fail")
+        return os_stat_orig(p)
 
-    monkeypatch.setattr(get_usb_info_module.subprocess, "check_output", raise_lsblk_error)
+    monkeypatch.setattr(get_usb_info_module.os, "stat", mock_os_stat)
 
+    # This should return None because of the catch-all Exception in get_usb_info
     assert get_usb_info_module.get_usb_info(mount_path) is None