Ok

 Please enter the commit message for your changes. Lines starting
 with '' will be ignored, and an empty message aborts the commit.

 On branch less-deps-feature-win-tweaks
 Your branch is up to date with 'origin/less-deps-feature-win-tweaks'.

 Changes to be committed:
	modified:   src/lufus/drives/find_usb.py
	modified:   src/lufus/drives/get_usb_info.py
	modified:   src/lufus/writing/flash_usb.py
	modified:   tests/test_flash_usb_and_find_usb_fixes.py
	modified:   tests/test_get_usb_info.py

 Untracked files:
	tests/conftest.py

Author: R-Larocque

src/lufus/drives/find_usb.py

@@ -35,6 +35,16 @@ def _media_directories() -> list[str]:
 
 
 ### USB RECOGNITION ###
+def _get_device_from_node(context: pyudev.Context, device_node: str) -> pyudev.Device | None:
+    """Helper to resolve a device node to a pyudev device."""
+    try:
+        st = os.stat(device_node)
+        return pyudev.Devices.from_device_number(context, "block", st.st_rdev)
+    except Exception as err:
+        log.debug("Could not resolve device %s via udev: %s", device_node, err)
+        return None
+
+
 def find_usb() -> dict[str, str]:
     """Return a mapping of mount-path -> volume-label for detected USB drives."""
     usbdict = {}  # DICTIONARY WHERE USB MOUNT PATH IS KEY AND LABEL IS VALUE
@@ -52,14 +62,8 @@ def find_usb() -> dict[str, str]:
         if not device_node:
             continue
 
-        label = None
-        try:
-            # 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
+        device = _get_device_from_node(context, device_node)
+        label = device.get("ID_FS_LABEL") if device else None
 
         if not label:
             label = os.path.basename(mount_path)

src/lufus/drives/get_usb_info.py

@@ -13,6 +13,16 @@ class USBDeviceInfo(TypedDict):
     mount_path: str
 
 
+def _get_device_from_node(context: pyudev.Context, device_node: str) -> pyudev.Device | None:
+    """Helper to resolve a device node to a pyudev device."""
+    try:
+        st = os.stat(device_node)
+        return pyudev.Devices.from_device_number(context, "block", st.st_rdev)
+    except Exception as err:
+        log.debug("Could not resolve device %s via udev: %s", device_node, err)
+        return None
+
+
 def get_usb_info(usb_path: str) -> USBDeviceInfo | None:
     try:
         normalized_usb_path = os.path.normpath(usb_path)
@@ -26,13 +36,20 @@ def get_usb_info(usb_path: str) -> USBDeviceInfo | None:
             return None
 
         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)
+        device = _get_device_from_node(context, device_node)
+        if not device:
+            return None
 
         # 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
+        try:
+            usb_size = int(size_attr) * 512 if size_attr is not None else 0
+        except (ValueError, TypeError):
+            log.warning(
+                "Unexpected non-numeric udev size attribute %r; defaulting USB size to 0",
+                size_attr,
+            )
+            usb_size = 0
 
         label = device.get("ID_FS_LABEL")
 

src/lufus/writing/flash_usb.py

@@ -116,19 +116,31 @@ def _status(msg: str) -> None:
 
         _status(f"dd process started with PID {process.pid}")
 
+        # Use non-blocking read for stderr to handle \r updates without blocking
+        os.set_blocking(process.stderr.fileno(), False)
+
         buf = b""
         last_pct = -1
         while True:
-            # Read in small chunks to handle \r progress updates from dd without blocking
-            # until a newline (\n) is received. status=progress usually emits \r.
             try:
-                chunk = process.stderr.read(128)
+                chunk = process.stderr.read(4096)
             except Exception as e:
                 log.warning("Error reading dd stderr: %s", e)
                 break
 
+            if chunk is None:
+                # No data yet, wait a bit
+                import time
+                time.sleep(0.1)
+                if process.poll() is not None:
+                    break
+                continue
+
             if not chunk:
-                break
+                if process.poll() is not None:
+                    break
+                continue
+
             buf += chunk
             # Split by \r or \n to catch all progress updates
             parts = re.split(rb"[\r\n]", buf)

tests/test_flash_usb_and_find_usb_fixes.py

@@ -10,6 +10,7 @@
 import sys
 from pathlib import Path
 from types import SimpleNamespace
+from unittest.mock import MagicMock
 
 ROOT = Path(__file__).resolve().parents[1]
 SRC = ROOT / "src"
@@ -145,10 +146,25 @@ def __init__(self, args, **kwargs):
             def wait(self):
                 pass
 
+            def poll(self):
+                return 0
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, exc_type, exc_val, exc_tb):
+                pass
+
         class FakePipe:
             def readline(self):
                 return b""
 
+            def fileno(self):
+                return 3
+
+            def read(self, size):
+                return b""
+
         monkeypatch.setattr(flash_usb_module.subprocess, "Popen", FakeProcess)
 
         flash_usb("/dev/nvme0n1p1", str(iso))
@@ -362,3 +378,132 @@ def test_find_dn_returns_device_node(self, monkeypatch):
         )
 
         assert find_usb_module.find_device_node() == "/dev/sdd1"
+
+class TestFlashUsbProgress:
+    def _fake_popen_factory(self, stderr_chunks, popen_calls, envs):
+        class FakePopen:
+            def __init__(self, cmd, **kwargs):
+                popen_calls.append((cmd, kwargs))
+                envs.append(kwargs.get("env"))
+                self.cmd = cmd
+                self.kwargs = kwargs
+                self.pid = 999
+                self.returncode = 0
+                self.stderr = MagicMock()
+                self.stderr.fileno.return_value = 3
+                
+                # Generator for non-blocking read simulator
+                def chunk_generator():
+                    for chunk in stderr_chunks:
+                        yield chunk.encode("utf-8")
+                    while True:
+                        yield b""
+                
+                self.gen = chunk_generator()
+
+            def poll(self):
+                return self.returncode
+
+            def wait(self):
+                return self.returncode
+            
+            @property
+            def stderr_mock(self):
+                # This matches the read(4096) call in flash_usb
+                def read_mock(size):
+                    try:
+                        val = next(self.gen)
+                        return val if val else None # simulate non-blocking "None" for empty
+                    except StopIteration:
+                        return b""
+                return read_mock
+
+        return FakePopen
+
+    def test_progress_cb_receives_milestones_and_scaled_dd_progress(self, monkeypatch):
+        progress_values = []
+        status_messages = []
+
+        def progress_cb(progress):
+            progress_values.append(progress)
+
+        def status_cb(status):
+            status_messages.append(status)
+
+        # 100MB ISO
+        iso_size = 100 * 1024 * 1024
+        stderr_chunks = [
+            "0+0 records in\r",
+            "0+0 records out\r",
+            "10485760 bytes (10 MB) copied\r", # 10%
+            "52428800 bytes (52 MB) copied\r", # 50%
+            "104857600 bytes (105 MB) copied\n", # 100%
+        ]
+
+        popen_calls = []
+        envs = []
+
+        # We need to mock os.path.getsize, check_iso_signature, detect_iso_type, os.set_blocking
+        monkeypatch.setattr(flash_usb_module.os.path, "getsize", lambda p: iso_size)
+        monkeypatch.setattr(flash_usb_module, "check_iso_signature", lambda p: True)
+        monkeypatch.setattr(flash_usb_module, "detect_iso_type", lambda p: flash_usb_module.IsoType.LINUX)
+        monkeypatch.setattr(flash_usb_module.os, "set_blocking", lambda fd, block: None)
+
+        fake_popen_class = self._fake_popen_factory(stderr_chunks, popen_calls, envs)
+        
+        # We must patch subprocess.Popen where it is used in flash_usb_module
+        monkeypatch.setattr(flash_usb_module.subprocess, "Popen", 
+                            lambda cmd, **kwargs: (p := fake_popen_class(cmd, **kwargs), 
+                                                 setattr(p.stderr, 'read', p.stderr_mock), p)[2])
+
+        import os
+        original_environ = dict(os.environ)
+
+        flash_usb("/dev/sdb", "/path/to/image.iso", progress_cb=progress_cb, status_cb=status_cb)
+
+        assert envs[0].get("LC_ALL") == "C"
+        assert dict(os.environ) == original_environ
+
+        # Milestones: 2, 5, 8, 10
+        assert 2 in progress_values
+        assert 5 in progress_values
+        assert 8 in progress_values
+        assert 10 in progress_values
+        
+        # 10% raw -> 10 + 10*0.85 = 18.5 -> 18
+        # 50% raw -> 10 + 50*0.85 = 52.5 -> 52
+        # 100% raw -> 10 + 100*0.85 = 95
+        assert 18 in progress_values
+        assert 52 in progress_values
+        assert 95 in progress_values
+
+        assert any("10%" in msg for msg in status_messages)
+        assert any("50%" in msg for msg in status_messages)
+
+    def test_non_progress_and_unexpected_stderr_lines_handling(self, monkeypatch, caplog):
+        def progress_cb(p): pass
+        def status_cb(s): pass
+
+        iso_size = 1000
+        stderr_chunks = [
+            "1+0 records in\n",
+            "some unexpected warning from dd\n",
+        ]
+
+        popen_calls = []
+        envs = []
+
+        monkeypatch.setattr(flash_usb_module.os.path, "getsize", lambda p: iso_size)
+        monkeypatch.setattr(flash_usb_module, "check_iso_signature", lambda p: True)
+        monkeypatch.setattr(flash_usb_module, "detect_iso_type", lambda p: flash_usb_module.IsoType.LINUX)
+        monkeypatch.setattr(flash_usb_module.os, "set_blocking", lambda fd, block: None)
+
+        fake_popen_class = self._fake_popen_factory(stderr_chunks, popen_calls, envs)
+        monkeypatch.setattr(flash_usb_module.subprocess, "Popen", 
+                            lambda cmd, **kwargs: (p := fake_popen_class(cmd, **kwargs), 
+                                                 setattr(p.stderr, 'read', p.stderr_mock), p)[2])
+
+        with caplog.at_level("WARNING"):
+            flash_usb("/dev/sdb", "/path/to/image.iso", progress_cb=progress_cb, status_cb=status_cb)
+
+        assert any("some unexpected warning from dd" in record.getMessage() for record in caplog.records)