[core] Skip PlatformIO's private-package authorization probe

This commit is contained in:
J. Nick Koston
2026-08-27 10:35:55 -05:00
parent a8c8827a5b
commit 61c0f84ed2
4 changed files with 112 additions and 1 deletions
+5
View File
@@ -577,6 +577,11 @@ def _make_registry_client() -> Any:
"""
from platformio.package.manager._registry import PackageManagerRegistryMixin
from esphome.platformio.runner import patch_registry_private_packages
# Every lookup would otherwise sleep ~500 ms in PlatformIO's account probe
patch_registry_private_packages()
class _Registry(PackageManagerRegistryMixin):
def __init__(self) -> None:
self._registry_client = None
+25 -1
View File
@@ -2,7 +2,8 @@
Invoked via ``python -m esphome.platformio.runner`` instead of
``python -m platformio`` so that the patches (incremental rebuild
preservation, download retries) apply inside the subprocess. Running
preservation, download retries, skipping the private-package probe) apply
inside the subprocess. Running
PlatformIO in a subprocess keeps its ``sys.path`` mutations and other
global state from leaking into the ESPHome process.
"""
@@ -105,6 +106,28 @@ def patch_file_downloader() -> None:
FileDownloader.__init__ = patched_init
def patch_registry_private_packages() -> None:
"""Answer PlatformIO's private-package probe without the network.
``RegistryClient.get_package()`` calls ``allowed_private_packages()``
before it checks its own HTTP cache. That probe goes through
``AccountClient``'s throttled ``send_request``, which sleeps up to 500 ms
to space out requests and then fails at once without a PlatformIO login,
so every registry lookup costs half a second of sleep for nothing.
ESPHome never uses private registry packages, so answer False directly.
"""
from platformio.registry.client import RegistryClient
if getattr(RegistryClient.allowed_private_packages, "_esphome_patched", False):
return
def no_private_packages() -> bool:
return False
no_private_packages._esphome_patched = True # type: ignore[attr-defined] # pylint: disable=protected-access
RegistryClient.allowed_private_packages = staticmethod(no_private_packages) # type: ignore[method-assign]
_IGNORE_LIB_WARNINGS = "(?:Hash|Update)"
# Regex patterns matched against each line of PlatformIO output. Lines that
# match are dropped by RedirectText before they reach the parent process.
@@ -152,6 +175,7 @@ FILTER_PLATFORMIO_LINES = [
def main() -> int:
patch_structhash()
patch_file_downloader()
patch_registry_private_packages()
# Wrap stdout/stderr with RedirectText before PlatformIO runs:
#
@@ -224,6 +224,28 @@ def test_resolve_registry_version_raises_without_pkg_file(monkeypatch):
_resolve_registry_version("owner", "pkg", set())
def test_make_registry_client_skips_private_package_probe(monkeypatch):
"""Resolving through our client never sleeps in PlatformIO's account probe."""
from platformio.account.client import AccountClient
from platformio.registry.client import RegistryClient
# Undo the class-level patch after the test so other tests see the original
monkeypatch.setattr(
RegistryClient,
"allowed_private_packages",
RegistryClient.__dict__["allowed_private_packages"],
)
def fail(*_args, **_kwargs):
raise AssertionError("account probe must not run")
monkeypatch.setattr(AccountClient, "get_account_info", fail)
client = lib._make_registry_client().get_registry_client_instance()
assert client.allowed_private_packages() is False
def _patch_registry_resolve(monkeypatch: pytest.MonkeyPatch) -> None:
"""Stub the registry lookup so tests never touch the network."""
monkeypatch.setattr(
@@ -30,6 +30,7 @@ def _prepare_main(
monkeypatch.setattr(sys, "stderr", stream)
monkeypatch.setattr(runner, "patch_structhash", lambda: None)
monkeypatch.setattr(runner, "patch_file_downloader", lambda: None)
monkeypatch.setattr(runner, "patch_registry_private_packages", lambda: None)
platformio = ModuleType("platformio")
platformio_main = ModuleType("platformio.__main__")
@@ -91,3 +92,62 @@ def test_main_still_filters_a_drained_partial_line(
assert runner.main() == 0
assert buf.getvalue() == b""
def test_main_applies_registry_private_packages_patch(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The probe patch is installed before PlatformIO gets control."""
order: list[str] = []
_prepare_main(monkeypatch, lambda: order.append("pio") or 0)
monkeypatch.setattr(
runner, "patch_registry_private_packages", lambda: order.append("patch")
)
assert runner.main() == 0
assert order == ["patch", "pio"]
def _restore_registry_probe(monkeypatch: pytest.MonkeyPatch) -> None:
"""Undo the class-level patch after the test so other tests see PlatformIO's own probe."""
from platformio.registry.client import RegistryClient
monkeypatch.setattr(
RegistryClient,
"allowed_private_packages",
RegistryClient.__dict__["allowed_private_packages"],
)
def test_patch_registry_private_packages_skips_account_probe(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The patched probe answers False without touching the account client."""
from platformio.account.client import AccountClient
from platformio.registry.client import RegistryClient
_restore_registry_probe(monkeypatch)
def fail(*_args, **_kwargs):
raise AssertionError("account probe must not run")
monkeypatch.setattr(AccountClient, "get_account_info", fail)
runner.patch_registry_private_packages()
assert RegistryClient.allowed_private_packages() is False
assert RegistryClient().allowed_private_packages() is False
def test_patch_registry_private_packages_is_idempotent(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from platformio.registry.client import RegistryClient
_restore_registry_probe(monkeypatch)
runner.patch_registry_private_packages()
patched = RegistryClient.allowed_private_packages
runner.patch_registry_private_packages()
assert RegistryClient.allowed_private_packages is patched