Merge branch 'esp8266-native-library-converter' into esp8266-native-shared-helpers

esphome/espidf/framework.py

@@ -17,11 +17,11 @@ import platformdirs
 from esphome.core import CORE, Version
 from esphome.framework_helpers import (
     PathType,
-    _failure_reason,
     archive_extract_all,
     create_venv,
     download_from_mirrors,
     download_with_resume,
+    failure_reason,
     get_python_env_executable_path,
     get_system_python_path,
     rmdir,
@@ -788,8 +788,8 @@ def _prefetch_idf_tool_archives(
             ],
         )
         for name, e in failures:
-            # _failure_reason: a message-less exception must not log blank
-            _LOGGER.warning("Could not prefetch %s: %s", name, _failure_reason(e))
+            # failure_reason: a message-less exception must not log blank
+            _LOGGER.warning("Could not prefetch %s: %s", name, failure_reason(e))
             _LOGGER.debug("Prefetch failure detail", exc_info=e)
     except Exception as e:  # noqa: BLE001  # pylint: disable=broad-exception-caught
         # The installer downloads anything missing itself; never let the

esphome/framework_helpers.py

@@ -1058,11 +1058,11 @@ def download_with_resume(
 
     raise EsphomeError(
         f"Failed to download {url} after {attempts} attempts: "
-        f"{_failure_reason(last_error)}"
+        f"{failure_reason(last_error)}"
     ) from last_error
 
 
-def _failure_reason(e: BaseException) -> str:
+def failure_reason(e: BaseException) -> str:
     """Format a download exception for the aggregated error message.
 
     ``requests`` appends " for url: <url>" to HTTP errors; the URL is already
@@ -1078,7 +1078,7 @@ def _spent_attempts_error(e: Exception, attempts: int) -> Exception:
     the sweep classifies it as permanent."""
     from esphome.core import EsphomeError
 
-    err = EsphomeError(f"failed after {attempts} attempts: {_failure_reason(e)}")
+    err = EsphomeError(f"failed after {attempts} attempts: {failure_reason(e)}")
     err.__cause__ = e
     return err
 
@@ -1304,7 +1304,7 @@ def download_from_mirrors(
             _LOGGER.warning(
                 "Download of %s failed (%s); retrying in %d seconds (attempt %d/%d)",
                 transient[0],
-                _failure_reason(transient[1]),
+                failure_reason(transient[1]),
                 delay,
                 sweep + 1,
                 _MIRROR_SWEEP_ATTEMPTS,
@@ -1325,7 +1325,7 @@ def download_from_mirrors(
         seen: set[tuple[str, str]] = set()
         attempts = ""
         for url, e in failures:
-            reason = _failure_reason(e)
+            reason = failure_reason(e)
             if (url, reason) not in seen:
                 seen.add((url, reason))
                 attempts += f"\n  {url}\n    {reason}"

esphome/platformio/library.py

@@ -32,9 +32,9 @@ from urllib.request import url2pathname
 from esphome import git
 from esphome.core import CORE, EsphomeError, Library
 from esphome.framework_helpers import (
-    _failure_reason,
     archive_extract_all,
     download_from_mirrors,
+    failure_reason,
     rmdir,
     run_batch_downloads,
 )
@@ -954,7 +954,7 @@ def _prefetch_wave(
             _LOGGER.warning(
                 "Prefetch of %s failed (retrying sequentially): %s",
                 name,
-                _failure_reason(err),
+                failure_reason(err),
             )
             _LOGGER.debug("Prefetch failure detail", exc_info=err)
     except Exception as err:  # noqa: BLE001  # pylint: disable=broad-exception-caught

tests/unit_tests/test_helpers.py

@@ -1117,6 +1117,20 @@ def test_progressbar_enabled_on_pipe_with_dashboard(monkeypatch) -> None:
     assert bar.enabled is True
 
 
+def test_progressbar_interrupt_keeps_finished_bar_done(monkeypatch) -> None:
+    """interrupt() on a bar whose 100% frame already ended its own line
+    must not reset it, or the next tick would redraw a second Done row."""
+    stream = MagicMock(spec=io.TextIOWrapper)
+    stream.isatty.return_value = True
+    monkeypatch.setattr(CORE, "dashboard", False)
+
+    bar = ProgressBar("Uploading", stream=stream)
+    bar.update(1)
+    assert bar.last_progress == 100
+    bar.interrupt()
+    assert bar.last_progress == 100
+
+
 @pytest.mark.parametrize(
     ("seconds", "expected"),
     [