Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
R-Larocque wants to merge 3 commits into from
Closed

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
732856d
Tests must be fixed. Pwease.
R-Larocque May 7, 2026
205d3b3
FIXED! Tests works, and LSBLK is replaced w/Python native tooling1
R-Larocque May 7, 2026
f77fd1c
Issue #178 fixed.
R-Larocque May 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 7 additions & 7 deletions src/lufus/drives/find_usb.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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:
Expand Down
28 changes: 11 additions & 17 deletions src/lufus/drives/get_usb_info.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import psutil
import os
import subprocess
import pyudev
from typing import TypedDict
from lufus.lufus_logging import get_logger

Expand All @@ -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()

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)
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)

# 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)

Expand All @@ -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
45 changes: 38 additions & 7 deletions src/lufus/writing/flash_usb.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,8 @@ def _status(msg: str) -> None:
try:
iso_size = os.path.getsize(iso_path)
_status(f"File size: {iso_size:,} bytes ({iso_size / (1024**3):.2f} GiB)")
if progress_cb:
progress_cb(2)

if iso_path.lower().endswith(".iso"):
_status(f"Validating ISO9660 signature for: {iso_path}")
Expand All @@ -62,8 +64,15 @@ def _status(msg: str) -> None:
else:
_status(f"Not an ISO file ({os.path.basename(iso_path)}), skipping ISO signature check")

if progress_cb:
progress_cb(5)

_status("Checking if image contains installation markers...")
if is_windows_iso(iso_path):
iso_type = detect_iso_type(iso_path)
if progress_cb:
progress_cb(8)

if iso_type == IsoType.WINDOWS:
_status("Windows Installation media detected, routing to flash_windows (ISO mode)")
return flash_windows(
device,
Expand All @@ -73,12 +82,14 @@ def _status(msg: str) -> None:
status_cb=status_cb,
)

iso_type = detect_iso_type(iso_path)
if iso_type == IsoType.LINUX:
_status("Linux Installation media detected, will use dd for flashing")
else:
_status("Generic or unknown image, will use dd for flashing")

if progress_cb:
progress_cb(10)

dd_args = [
"dd",
f"if={iso_path}",
Expand All @@ -93,7 +104,11 @@ def _status(msg: str) -> None:
_status(f"Writing {iso_size:,} bytes to {shlex.quote(device)}, this may take several minutes...")

try:
process = subprocess.Popen(dd_args, stderr=subprocess.PIPE, stdout=subprocess.DEVNULL)
# Use LC_ALL=C to ensure "bytes" is the keyword for progress parsing
# and set a consistent output format across different locales.
env = os.environ.copy()
env["LC_ALL"] = "C"
process = subprocess.Popen(dd_args, stderr=subprocess.PIPE, stdout=subprocess.DEVNULL, env=env)
except FileNotFoundError:
log.error("Flash failed: 'dd' utility not found. Install coreutils.")
_status("Flash failed: 'dd' utility not found. Install coreutils.")
Expand All @@ -104,27 +119,43 @@ def _status(msg: str) -> None:
buf = b""
last_pct = -1
while True:
chunk = process.stderr.readline()
# 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)
except Exception as e:
log.warning("Error reading dd stderr: %s", e)
break

if not chunk:
break
buf += chunk
# Split by \r or \n to catch all progress updates
parts = re.split(rb"[\r\n]", buf)
# The last part might be incomplete, keep it in the buffer
buf = parts[-1]

for line in parts[:-1]:
line = line.strip()
if not line:
continue
m = re.match(rb"^(\d+)\s+bytes", line)
if m and iso_size > 0:
bytes_done = int(m.group(1))
pct = min(int(bytes_done * 100 / iso_size), 99)
# Scale progress to 10-95% range to leave room for early steps and final sync
pct_raw = min(int(bytes_done * 100 / iso_size), 100)
pct = 10 + int(pct_raw * 0.85)

if pct != last_pct:
_status(f"dd progress: {bytes_done:,} / {iso_size:,} bytes ({pct}%)")
_status(f"dd progress: {bytes_done:,} / {iso_size:,} bytes ({pct_raw}%)")
last_pct = pct
if progress_cb:
progress_cb(pct)
else:
log.warning("dd stderr: %s", line.decode("utf-8", errors="replace"))
# Filter out common dd output lines to avoid logging noise
line_str = line.decode("utf-8", errors="replace")
if not any(x in line_str for x in ["records in", "records out", "copied"]):
log.warning("dd stderr: %s", line_str)

process.wait()
_status(f"dd process exited with return code {process.returncode}")
Expand Down
51 changes: 38 additions & 13 deletions tests/test_find_usb.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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"
Expand All @@ -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(
Expand All @@ -35,21 +37,34 @@ 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(
Expand All @@ -70,13 +85,23 @@ 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")

monkeypatch.setattr(find_usb_module.subprocess, "check_output", raise_lsblk_error)
# 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)

# 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"}
Expand Down
27 changes: 20 additions & 7 deletions tests/test_flash_usb_and_find_usb_fixes.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Comment on lines -288 to 293

@sourcery-ai sourcery-ai Bot May 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion (testing): New flash_usb progress and dd stderr parsing behaviour is not exercised; add tests to cover progress_cb milestones and LC_ALL handling.

The new flash_usb logic (early progress_cb milestones, dd progress scaling, and LC_ALL + stderr parsing) isn’t covered by tests. Please add focused tests (e.g. a TestFlashUsbProgress suite) that:

  1. Use a fake subprocess.Popen with controlled stderr to verify:
    • progress_cb receives the expected early milestones and correctly scaled dd progress values.
    • Status messages still report the raw dd percentage while progress_cb sees the scaled percentage.
  2. Assert subprocess.Popen is invoked with an env that sets LC_ALL="C" without mutating the global environment.
  3. Confirm non-progress lines like "records in/out" and "copied" are ignored, while other unexpected stderr lines still produce a warning.

This will guard against regressions in progress reporting and locale-dependent parsing.

Suggested implementation:

            find_usb_module.psutil,
            "disk_partitions",
        )


class TestFlashUsbProgress:
    def _fake_popen_factory(self, stderr_lines, 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.returncode = 0
                # simulate a text-mode stderr iterable as dd would produce
                self.stderr = iter(line.encode("utf-8") for line in stderr_lines)

            def wait(self):
                return self.returncode

        return FakePopen

    def test_progress_cb_receives_milestones_and_scaled_dd_progress(self, monkeypatch):
        progress_values = []
        status_messages = []

        def progress_cb(progress, status):
            progress_values.append(progress)
            status_messages.append(status)

        stderr_lines = [
            "0+0 records in\n",
            "0+0 records out\n",
            "10% completed\n",
            "12345+0 records in\n",
            "12345+0 records out\n",
            "50% completed\n",
            "copied, 1.23 s, 4.56 MB/s\n",
            "100% completed\n",
        ]

        popen_calls = []
        envs = []

        fake_popen = self._fake_popen_factory(stderr_lines, popen_calls, envs)
        monkeypatch.setattr(flash_usb_module.subprocess, "Popen", fake_popen)

        # Capture a copy of the current environment to ensure it is not mutated
        import os

        original_environ = dict(os.environ)

        # Call the function under test; adjust arguments to match your flash_usb signature
        flash_usb_module.flash_usb("/dev/sdb", "/path/to/image", progress_cb=progress_cb)

        # Assert LC_ALL is set to "C" in the spawned process environment
        assert envs, "subprocess.Popen should have been called at least once"
        env = envs[0]
        assert env is not None
        assert env.get("LC_ALL") == "C"
        # Ensure the global environment is not mutated
        assert dict(os.environ) == original_environ

        # progress_cb should be called with early milestones as well as scaled dd progress
        # Adjust these assertions if your implementation uses different milestone values.
        assert progress_values[0] == 0.0
        assert progress_values[-1] == 1.0
        assert sorted(progress_values) == progress_values

        # Status messages should contain the raw dd progress lines
        assert any("10% completed" in msg for msg in status_messages)
        assert any("50% completed" in msg for msg in status_messages)
        assert any("100% completed" in msg for msg in status_messages)

    def test_non_progress_and_unexpected_stderr_lines_handling(self, monkeypatch, caplog):
        progress_values = []
        status_messages = []

        def progress_cb(progress, status):
            progress_values.append(progress)
            status_messages.append(status)

        stderr_lines = [
            "1+0 records in\n",
            "1+0 records out\n",
            "copied, 1.23 s, 4.56 MB/s\n",
            "some unexpected warning from dd\n",
        ]

        popen_calls = []
        envs = []

        fake_popen = self._fake_popen_factory(stderr_lines, popen_calls, envs)
        monkeypatch.setattr(flash_usb_module.subprocess, "Popen", fake_popen)

        with caplog.at_level("WARNING"):
            flash_usb_module.flash_usb("/dev/sdb", "/path/to/image", progress_cb=progress_cb)

        # Non-progress bookkeeping lines should be ignored and not result in progress_cb calls
        assert progress_values == []

        # But unexpected stderr lines should still be surfaced in logs
        unexpected_logs = [
            record for record in caplog.records if "some unexpected warning from dd" in record.getMessage()
        ]
        assert unexpected_logs, "Unexpected stderr lines should result in a warning log entry"
  1. Ensure flash_usb_module is imported at the top of tests/test_flash_usb_and_find_usb_fixes.py, for example:
    import flash_usb_module or from <your_package> import flash_usb as flash_usb_module, matching how find_usb_module is imported in this file.
  2. Align the flash_usb_module.flash_usb call signature with your implementation. If the function uses different parameter names or order (e.g. image_path then device or additional options), update the tests accordingly.
  3. If flash_usb emits progress milestones with specific numeric values (e.g. 0.05, 0.1, etc.), update the assert statements on progress_values to check for those exact milestones instead of only checking the first and last values and monotonicity.
  4. If your logging uses a logger other than the root logger, adjust the caplog.at_level(...) call to pass the correct logger name (e.g. caplog.at_level("WARNING", logger="your.module.logger")) so that the warning assertions observe the expected records.
  5. If subprocess.Popen is accessed differently (e.g. from subprocess import Popen in flash_usb_module), modify the monkeypatch.setattr target accordingly so that the fake Popen is actually used by flash_usb.

Comment on lines 287 to 294

@sourcery-ai sourcery-ai Bot May 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion (testing): New flash_usb progress and dd stderr-parsing behavior is not covered by tests; consider adding focused tests here.

These changes introduce several new behaviors in flash_usb (early progress_cb calls, LC_ALL=C in the dd env, chunked stderr reads with \r, scaled progress, and filtering of common dd output), but none appear to be exercised by tests. Please add tests that, for example:

  • Mock subprocess.Popen to assert dd is invoked with LC_ALL="C".
  • Feed simulated dd stderr with \r-separated/incomplete progress lines and verify progress_cb is monotonic within 10–95, and that the status line uses the raw percentage while progress_cb sees the scaled value.
  • Ensure "records in", "records out", and "copied" lines are not logged as warnings, but unrelated stderr lines still are.
  • Optionally verify early progress_cb invocations (2, 5, 8) occur in the correct order for Windows vs non-Windows ISO paths.

This will help prevent regressions in the new progress and parsing logic.

Suggested implementation:

            find_usb_module.psutil,
            "disk_partitions",

class TestFlashUsbProgressAndParsing:
    def _install_fake_popen(self, monkeypatch, stderr_bytes, captured_popen_args):
        """
        Install a fake Popen on flash_usb_module.subprocess.Popen that:
        - Records args and env for assertions.
        - Exposes a bytes-like stderr stream containing stderr_bytes.
        - Behaves like a finished process with returncode=0.
        """
        import io

        class FakePopen:
            def __init__(self, args, **kwargs):
                # Record arguments for assertions
                captured_popen_args["args"] = args
                captured_popen_args["kwargs"] = kwargs
                self.args = args
                self.kwargs = kwargs
                self.env = kwargs.get("env") or {}
                # Simulated stderr stream; flash_usb should read from this
                self.stderr = io.BytesIO(stderr_bytes)
                self.returncode = 0

            def poll(self):
                return self.returncode

            def wait(self, timeout=None):
                return self.returncode

            def communicate(self, input=None, timeout=None):
                # flash_usb shouldn't rely on this, but provide it just in case
                return (b"", stderr_bytes)

        monkeypatch.setattr(flash_usb_module.subprocess, "Popen", FakePopen)

    def test_dd_invoked_with_lc_all_c(self, monkeypatch):
        """
        Ensure that flash_usb invokes dd with LC_ALL set to 'C'.
        """
        captured_popen_args = {}

        # stderr doesn't matter for this test
        self._install_fake_popen(monkeypatch, stderr_bytes=b"", captured_popen_args=captured_popen_args)

        progress_calls = []

        def progress_cb(value, status=None):
            progress_calls.append((value, status))

        # Use a dummy ISO path and device path; adjust if your API differs
        flash_usb_module.flash_usb("/tmp/test.iso", "/dev/sdz", progress_cb=progress_cb)

        # Assert Popen was invoked
        assert "args" in captured_popen_args
        args = captured_popen_args["args"]
        kwargs = captured_popen_args["kwargs"]

        # dd should be part of the invoked command
        assert any("dd" in str(part) for part in (args[0] if isinstance(args[0], (list, tuple)) else args)), args

        # LC_ALL must be set to "C" in the environment
        env = kwargs.get("env") or {}
        assert env.get("LC_ALL") == "C"

    def test_progress_parsing_monotonic_and_scaled(self, monkeypatch, caplog):
        """
        Feed simulated dd stderr with \\r-separated/incomplete progress lines and verify:
        - progress_cb is monotonic and stays within 10–95.
        - status line uses raw percentage while progress_cb sees scaled value.
        """
        caplog.set_level("INFO")

        # Simulate dd-style progress output with carriage returns and partial lines
        stderr_bytes = (
            b"12345 bytes (12 MB) copied, 1.0 s, 12.3 MB/s\r"
            b"12% done, 1.0 s, 12.3 MB/s\r"
            b"33% done, 2.0 s, 11.0 MB/s\r"
            b"95% done, 3.0 s, 10.0 MB/s\r"
        )

        captured_popen_args = {}
        self._install_fake_popen(monkeypatch, stderr_bytes=stderr_bytes, captured_popen_args=captured_popen_args)

        progress_values = []
        status_values = []

        def progress_cb(value, status=None):
            progress_values.append(value)
            status_values.append(status)

        flash_usb_module.flash_usb("/tmp/test.iso", "/dev/sdz", progress_cb=progress_cb)

        # Filter out any early "pre-dd" progress calls like 2/5/8
        meaningful_progress = [v for v in progress_values if 10 <= v <= 95]
        assert meaningful_progress, "Expected scaled progress values between 10 and 95"

        # Monotonic non-decreasing within the 10–95 window
        assert meaningful_progress == sorted(meaningful_progress)

        # Status messages should contain the raw dd percentages (12%, 33%, 95%)
        # while progress_cb receives scaled values.
        # We do not assert the exact scaling function, just that raw percentages appear in status.
        raw_percent_fragments = ["12%", "33%", "95%"]
        status_str = " ".join(str(s) for s in status_values if s is not None)
        for fragment in raw_percent_fragments:
            assert fragment in status_str

        # Also check log messages contain raw percentages (status line text)
        log_text = " ".join(rec.getMessage() for rec in caplog.records)
        for fragment in raw_percent_fragments:
            assert fragment in log_text

    def test_dd_common_output_not_warned_but_other_stderr_is(self, monkeypatch, caplog):
        """
        Ensure 'records in', 'records out', and 'copied' lines are not logged as warnings,
        but unrelated stderr lines still are.
        """
        caplog.set_level("WARNING")

        stderr_lines = [
            b"123+0 records in\n",
            b"123+0 records out\n",
            b"123456 bytes (123 MB) copied, 1.23 s, 100 MB/s\n",
            b"weird dd error: something unexpected\n",
        ]
        stderr_bytes = b"".join(stderr_lines)

        captured_popen_args = {}
        self._install_fake_popen(monkeypatch, stderr_bytes=stderr_bytes, captured_popen_args=captured_popen_args)

        def progress_cb(value, status=None):
            # progress_cb is irrelevant for this test
            pass

        flash_usb_module.flash_usb("/tmp/test.iso", "/dev/sdz", progress_cb=progress_cb)

        warning_messages = [rec.getMessage() for rec in caplog.records if rec.levelname == "WARNING"]

        # Common dd noise should not be warned
        assert not any("records in" in msg for msg in warning_messages)
        assert not any("records out" in msg for msg in warning_messages)
        assert not any("copied" in msg for msg in warning_messages)

        # But unrelated stderr lines should still show up as warnings
        assert any("weird dd error" in msg for msg in warning_messages)

    def test_early_progress_calls_for_windows_iso(self, monkeypatch):
        """
        Verify early progress_cb invocations (e.g. 2, 5, 8) occur in the correct order
        for a Windows ISO path.
        """
        captured_popen_args = {}
        # No need for real progress; dd won't be heavily exercised here
        self._install_fake_popen(monkeypatch, stderr_bytes=b"", captured_popen_args=captured_popen_args)

        progress_values = []

        def progress_cb(value, status=None):
            progress_values.append(value)

        # Heuristic: treat a path ending with '.iso' and containing backslashes as "Windows ISO"
        flash_usb_module.flash_usb(r"C:\\isos\\windows-server.iso", "/dev/sdz", progress_cb=progress_cb)

        # Expect specific early progress hints; adjust expected sequence to match implementation
        early = [v for v in progress_values if v < 10]
        assert early == sorted(early), "Early progress should be increasing for Windows ISO"
        # At least contain the milestones (implementation-dependent; see additional_changes)
        for milestone in (2, 5, 8):
            assert milestone in early

    def test_early_progress_calls_for_non_windows_iso(self, monkeypatch):
        """
        Verify early progress_cb invocations occur in correct order for a non-Windows ISO path.
        """
        captured_popen_args = {}
        self._install_fake_popen(monkeypatch, stderr_bytes=b"", captured_popen_args=captured_popen_args)

        progress_values = []

        def progress_cb(value, status=None):
            progress_values.append(value)

        flash_usb_module.flash_usb("/isos/linux-distro.iso", "/dev/sdz", progress_cb=progress_cb)

        early = [v for v in progress_values if v < 10]
        assert early == sorted(early), "Early progress should be increasing for non-Windows ISO"
        # Non-Windows images may skip the earliest hint; adjust as needed
        assert early, "Expected at least one early progress value for non-Windows ISO"
  1. Ensure flash_usb_module is imported in tests/test_flash_usb_and_find_usb_fixes.py (for example, import flash_usb as flash_usb_module) if it is not already.
  2. The helper _install_fake_popen uses io.BytesIO; add import io at the top of the test file if not already present.
  3. The exact scaling from raw dd percentage to progress values (within 10–95) and the exact early progress milestones (2, 5, 8) are inferred from your description; adjust the expected ranges/milestones and any assertions to match the actual behavior of flash_usb if they differ.
  4. If flash_usb has a different signature (e.g., keyword names for ISO/device paths or additional required parameters), update the test calls to flash_usb_module.flash_usb(...) accordingly.
  5. If flash_usb logs via a logger other than the root logger, you may need to configure caplog to capture that logger (e.g., caplog.set_level("INFO", logger="your.module.logger")) and adjust the tests.

Expand All @@ -308,14 +309,26 @@ 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"}

Expand Down
Loading
Loading