From 19280e03ad1bc6422fdf434b5b64601c40f48ee7 Mon Sep 17 00:00:00 2001 From: Franck Nijhof Date: Thu, 25 Jun 2026 20:42:35 +0000 Subject: [PATCH] [core] Defer requests import in external_files and web_server_ota Same pattern as framework_helpers: both modules imported requests at module top level but only use it inside functions that perform actual network I/O (downloading fonts/images, web_server OTA upload), never during config validation. external_files is loaded by the font and audio_file components. Defer the imports into the functions that use them, and update the tests to patch requests. directly (the modules no longer hold a requests attribute). Adds regression tests guarding against re-introducing the top-level imports. --- esphome/external_files.py | 10 +++- esphome/web_server_ota.py | 8 +-- tests/unit_tests/test_external_files.py | 37 +++++++++++--- tests/unit_tests/test_web_server_ota.py | 66 ++++++++++++++++--------- 4 files changed, 88 insertions(+), 33 deletions(-) diff --git a/esphome/external_files.py b/esphome/external_files.py index 4e73c8dc21..6ccf054bb3 100644 --- a/esphome/external_files.py +++ b/esphome/external_files.py @@ -9,8 +9,6 @@ import os from pathlib import Path import time -import requests - import esphome.config_validation as cv from esphome.const import CONF_FILE, CONF_TYPE, CONF_URL, __version__ from esphome.core import CORE, EsphomeError, TimePeriodSeconds @@ -92,6 +90,10 @@ def _write_etag(local_file_path: Path, etag: str | None) -> None: def has_remote_file_changed( url: str, local_file_path: Path, timeout: int = NETWORK_TIMEOUT ) -> bool: + # Imported lazily: requests is a heavy import (~85ms) only needed when + # actually checking remote files, never during config validation. + import requests + if local_file_path.exists(): _LOGGER.debug("has_remote_file_changed: File exists at %s", local_file_path) try: @@ -158,6 +160,10 @@ def compute_local_file_dir(domain: str) -> Path: def download_content(url: str, path: Path, timeout: int = NETWORK_TIMEOUT) -> bytes: + # Imported lazily: requests is a heavy import (~85ms) only needed when + # actually downloading, never during config validation. + import requests + if CORE.skip_external_update and path.exists(): _LOGGER.debug("Skipping update for %s (refresh disabled)", url) return path.read_bytes() diff --git a/esphome/web_server_ota.py b/esphome/web_server_ota.py index 8d0fdeecff..4b3b81e68c 100644 --- a/esphome/web_server_ota.py +++ b/esphome/web_server_ota.py @@ -15,9 +15,6 @@ import secrets import socket from typing import BinaryIO -import requests -from requests.auth import HTTPBasicAuth - from esphome.core import EsphomeError from esphome.helpers import ProgressBar, resolve_ip_address @@ -92,6 +89,11 @@ def _try_upload( password: str | None, filename: Path, ) -> tuple[int, str | None]: + # Imported lazily: requests is a heavy import (~85ms) only needed when + # actually performing a web_server OTA upload, never during config validation. + import requests + from requests.auth import HTTPBasicAuth + from esphome.core import CORE try: diff --git a/tests/unit_tests/test_external_files.py b/tests/unit_tests/test_external_files.py index 16cee9564f..a84424bd46 100644 --- a/tests/unit_tests/test_external_files.py +++ b/tests/unit_tests/test_external_files.py @@ -2,6 +2,8 @@ import os from pathlib import Path +import subprocess +import sys import time from unittest.mock import MagicMock, patch @@ -26,19 +28,21 @@ def _seed_etag(cache_file: Path, etag: str) -> Path: @pytest.fixture def mock_requests_head() -> MagicMock: - """Patch `external_files.requests.head` so the conditional HEAD-request - validator can be tested without doing real HTTP. + """Patch `requests.head` so the conditional HEAD-request validator can be + tested without doing real HTTP. external_files imports requests lazily, so + the global module is patched rather than a module-level attribute. """ - with patch("esphome.external_files.requests.head") as m: + with patch("requests.head") as m: yield m @pytest.fixture def mock_requests_get() -> MagicMock: - """Patch `external_files.requests.get` so the download path can be - tested without doing real HTTP. + """Patch `requests.get` so the download path can be tested without doing + real HTTP. external_files imports requests lazily, so the global module is + patched rather than a module-level attribute. """ - with patch("esphome.external_files.requests.get") as m: + with patch("requests.get") as m: yield m @@ -799,3 +803,24 @@ def test_download_content_atomic_write_no_partial_on_failure( # into the cache directory either way. leftover_tmps = list(setup_core.glob("tmp*")) assert leftover_tmps == [] + + +def test_importing_external_files_does_not_import_requests() -> None: + """Importing external_files must not drag in requests. + + requests is a heavy import (~85ms) only needed when actually fetching remote + files. external_files is imported during config validation (font, audio_file + components), so the import is deferred to the functions that use it. A fresh + interpreter is required because the test process has already imported it. + """ + result = subprocess.run( + [ + sys.executable, + "-c", + "import sys\nimport esphome.external_files\nprint('\\n'.join(sys.modules))", + ], + capture_output=True, + text=True, + check=True, + ) + assert "requests" not in result.stdout.split() diff --git a/tests/unit_tests/test_web_server_ota.py b/tests/unit_tests/test_web_server_ota.py index 606905e36e..25eae8dfda 100644 --- a/tests/unit_tests/test_web_server_ota.py +++ b/tests/unit_tests/test_web_server_ota.py @@ -6,6 +6,8 @@ import io import logging from pathlib import Path import socket +import subprocess +import sys from unittest.mock import MagicMock, patch import pytest @@ -173,7 +175,7 @@ def test_run_ota_success(monkeypatch: pytest.MonkeyPatch, firmware: Path) -> Non _patch_resolve(monkeypatch, [("192.168.1.50", 80)]) with patch( - "esphome.web_server_ota.requests.post", + "requests.post", return_value=_make_response(200, "Update Successful!"), ) as post: exit_code, host = run_ota(["device.local"], 80, None, None, firmware) @@ -199,7 +201,7 @@ def test_run_ota_logs_device_response_body( caplog.set_level(logging.INFO, logger="esphome.web_server_ota") with patch( - "esphome.web_server_ota.requests.post", + "requests.post", return_value=_make_response(200, "Update Successful!"), ): run_ota(["192.168.1.50"], 80, None, None, firmware) @@ -216,7 +218,7 @@ def test_run_ota_log_says_via_web_server( caplog.set_level(logging.INFO, logger="esphome.web_server_ota") with patch( - "esphome.web_server_ota.requests.post", + "requests.post", return_value=_make_response(200, "Update Successful!"), ): run_ota(["192.168.1.50"], 80, None, None, firmware) @@ -230,7 +232,7 @@ def test_run_ota_sends_basic_auth( _patch_resolve(monkeypatch, [("192.168.1.50", 80)]) with patch( - "esphome.web_server_ota.requests.post", + "requests.post", return_value=_make_response(200, "Update Successful!"), ) as post: exit_code, _ = run_ota(["192.168.1.50"], 80, "admin", "secret", firmware) @@ -248,7 +250,7 @@ def test_run_ota_skips_auth_when_no_credentials( _patch_resolve(monkeypatch, [("192.168.1.50", 80)]) with patch( - "esphome.web_server_ota.requests.post", + "requests.post", return_value=_make_response(200, "Update Successful!"), ) as post: run_ota(["192.168.1.50"], 80, None, None, firmware) @@ -263,7 +265,7 @@ def test_run_ota_skips_auth_when_only_username( _patch_resolve(monkeypatch, [("192.168.1.50", 80)]) with patch( - "esphome.web_server_ota.requests.post", + "requests.post", return_value=_make_response(200, "Update Successful!"), ) as post: run_ota(["192.168.1.50"], 80, "admin", None, firmware) @@ -277,7 +279,7 @@ def test_run_ota_uses_update_url( _patch_resolve(monkeypatch, [("192.168.1.50", 8080)]) with patch( - "esphome.web_server_ota.requests.post", + "requests.post", return_value=_make_response(200, "Update Successful!"), ) as post: run_ota(["192.168.1.50"], 8080, None, None, firmware) @@ -293,7 +295,7 @@ def test_run_ota_failure_response( _patch_resolve(monkeypatch, [("192.168.1.50", 80)]) with patch( - "esphome.web_server_ota.requests.post", + "requests.post", return_value=_make_response(200, "Update Failed!"), ): exit_code, host = run_ota(["192.168.1.50"], 80, None, None, firmware) @@ -309,7 +311,7 @@ def test_run_ota_failure_response_empty_body( _patch_resolve(monkeypatch, [("192.168.1.50", 80)]) with patch( - "esphome.web_server_ota.requests.post", + "requests.post", return_value=_make_response(200, ""), ): exit_code, host = run_ota(["192.168.1.50"], 80, None, None, firmware) @@ -325,7 +327,7 @@ def test_run_ota_auth_failed( _patch_resolve(monkeypatch, [("192.168.1.50", 80)]) with patch( - "esphome.web_server_ota.requests.post", + "requests.post", return_value=_make_response(401, "Unauthorized"), ): exit_code, host = run_ota(["192.168.1.50"], 80, "user", "wrong", firmware) @@ -341,7 +343,7 @@ def test_run_ota_unexpected_status_code( _patch_resolve(monkeypatch, [("192.168.1.50", 80)]) with patch( - "esphome.web_server_ota.requests.post", + "requests.post", return_value=_make_response(500, "Internal Error"), ): exit_code, host = run_ota(["192.168.1.50"], 80, None, None, firmware) @@ -361,7 +363,7 @@ def test_run_ota_unexpected_status_empty_body_falls_back( response.reason = "Service Unavailable" with patch( - "esphome.web_server_ota.requests.post", + "requests.post", return_value=response, ): exit_code, host = run_ota(["192.168.1.50"], 80, None, None, firmware) @@ -381,7 +383,7 @@ def test_run_ota_unexpected_status_no_body_no_reason( response.reason = "" with patch( - "esphome.web_server_ota.requests.post", + "requests.post", return_value=response, ): run_ota(["192.168.1.50"], 80, None, None, firmware) @@ -399,7 +401,7 @@ def test_run_ota_connection_error_then_success( ) with patch( - "esphome.web_server_ota.requests.post", + "requests.post", side_effect=[ requests.ConnectionError("refused"), _make_response(200, "Update Successful!"), @@ -422,7 +424,7 @@ def test_run_ota_request_exception_falls_through( ) with patch( - "esphome.web_server_ota.requests.post", + "requests.post", side_effect=[ requests.Timeout("read timeout"), _make_response(200, "Update Successful!"), @@ -444,7 +446,7 @@ def test_run_ota_all_addresses_unreachable( ) with patch( - "esphome.web_server_ota.requests.post", + "requests.post", side_effect=requests.ConnectionError("refused"), ): exit_code, host = run_ota(["device.local"], 80, None, None, firmware) @@ -516,7 +518,7 @@ def test_run_ota_string_host_accepted( _patch_resolve(monkeypatch, [("10.0.0.5", 80)]) with patch( - "esphome.web_server_ota.requests.post", + "requests.post", return_value=_make_response(200, "Update Successful!"), ): exit_code, host = run_ota("10.0.0.5", 80, None, None, firmware) @@ -544,7 +546,7 @@ def test_run_ota_multiple_hosts_first_fails( monkeypatch.setattr("esphome.web_server_ota.resolve_ip_address", _resolve) with patch( - "esphome.web_server_ota.requests.post", + "requests.post", side_effect=[ requests.ConnectionError("refused"), _make_response(200, "Update Successful!"), @@ -595,7 +597,7 @@ def test_run_ota_finalizes_progress_bar_on_success( with ( patch( - "esphome.web_server_ota.requests.post", + "requests.post", return_value=_make_response(200, "Update Successful!"), ), patch.object(ProgressBar, "done", lambda self: done_called.append(True)), @@ -615,7 +617,7 @@ def test_run_ota_finalizes_progress_bar_on_failure( with ( patch( - "esphome.web_server_ota.requests.post", + "requests.post", side_effect=requests.ConnectionError("boom"), ), patch.object(ProgressBar, "done", lambda self: done_called.append(True)), @@ -637,7 +639,7 @@ def test_run_ota_ipv6_url_brackets_host( ) with patch( - "esphome.web_server_ota.requests.post", + "requests.post", return_value=_make_response(200, "Update Successful!"), ) as post: exit_code, host = run_ota(["device.local"], 80, None, None, firmware) @@ -660,7 +662,7 @@ def test_run_ota_ipv6_link_local_includes_scope_id( ) with patch( - "esphome.web_server_ota.requests.post", + "requests.post", return_value=_make_response(200, "Update Successful!"), ) as post: exit_code, _ = run_ota(["device.local"], 80, None, None, firmware) @@ -668,3 +670,23 @@ def test_run_ota_ipv6_link_local_includes_scope_id( assert exit_code == 0 url = post.call_args.args[0] assert url == f"http://[fe80::1%253]:80{OTA_PATH}" + + +def test_importing_web_server_ota_does_not_import_requests() -> None: + """Importing web_server_ota must not drag in requests. + + requests is a heavy import (~85ms) only needed when actually performing a + web_server OTA upload, so the import is deferred into _try_upload. A fresh + interpreter is required because the test process has already imported it. + """ + result = subprocess.run( + [ + sys.executable, + "-c", + "import sys\nimport esphome.web_server_ota\nprint('\\n'.join(sys.modules))", + ], + capture_output=True, + text=True, + check=True, + ) + assert "requests" not in result.stdout.split()