mirror of
https://github.com/esphome/esphome.git
synced 2026-06-30 12:36:08 +00:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 19280e03ad | |||
| 1ff519446b | |||
| ee1fffb062 |
@@ -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()
|
||||
|
||||
@@ -11,8 +11,6 @@ import sys
|
||||
import time
|
||||
from typing import IO
|
||||
|
||||
import requests
|
||||
|
||||
from esphome.helpers import ProgressBar, rmtree
|
||||
|
||||
PathType = str | os.PathLike
|
||||
@@ -635,6 +633,10 @@ def download_from_mirrors(
|
||||
ValueError: If mirrors list is empty.
|
||||
Exception: If all download attempts fail.
|
||||
"""
|
||||
# Imported lazily: requests is a heavy import (~85ms) and is only needed
|
||||
# when actually downloading a toolchain, never during config validation.
|
||||
import requests
|
||||
|
||||
# 1. Open target file for writing if path given
|
||||
with ExitStack() as stack:
|
||||
if isinstance(target, (str, os.PathLike)):
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -526,7 +526,7 @@ class TestDownloadFromMirrors:
|
||||
def test_success_returns_url_and_writes_content(self, tmp_path: Path) -> None:
|
||||
target = tmp_path / "out.bin"
|
||||
with patch(
|
||||
"esphome.framework_helpers.requests.get",
|
||||
"requests.get",
|
||||
return_value=_mock_response(b"filedata"),
|
||||
):
|
||||
url = download_from_mirrors(["https://example.com/f"], {}, target)
|
||||
@@ -535,7 +535,7 @@ class TestDownloadFromMirrors:
|
||||
|
||||
def test_substitutions_applied_to_url(self, tmp_path: Path) -> None:
|
||||
with patch(
|
||||
"esphome.framework_helpers.requests.get",
|
||||
"requests.get",
|
||||
return_value=_mock_response(b"x"),
|
||||
) as mock_get:
|
||||
download_from_mirrors(
|
||||
@@ -547,7 +547,7 @@ class TestDownloadFromMirrors:
|
||||
|
||||
def test_falls_back_to_second_mirror(self, tmp_path: Path) -> None:
|
||||
with patch(
|
||||
"esphome.framework_helpers.requests.get",
|
||||
"requests.get",
|
||||
side_effect=[_mock_response(b"", ok=False), _mock_response(b"second")],
|
||||
):
|
||||
url = download_from_mirrors(
|
||||
@@ -561,7 +561,7 @@ class TestDownloadFromMirrors:
|
||||
def test_all_mirrors_fail_reraises_last_exception(self, tmp_path: Path) -> None:
|
||||
with (
|
||||
patch(
|
||||
"esphome.framework_helpers.requests.get",
|
||||
"requests.get",
|
||||
return_value=_mock_response(b"", ok=False),
|
||||
),
|
||||
pytest.raises(req.HTTPError),
|
||||
@@ -579,7 +579,7 @@ class TestDownloadFromMirrors:
|
||||
def test_file_like_target_written(self) -> None:
|
||||
buf = io.BytesIO()
|
||||
with patch(
|
||||
"esphome.framework_helpers.requests.get",
|
||||
"requests.get",
|
||||
return_value=_mock_response(b"bytes"),
|
||||
):
|
||||
download_from_mirrors(["https://example.com/f"], {}, buf)
|
||||
@@ -590,7 +590,7 @@ class TestDownloadFromMirrors:
|
||||
r = _mock_response(b"1234567890")
|
||||
r.headers = {"content-length": "10"}
|
||||
with (
|
||||
patch("esphome.framework_helpers.requests.get", return_value=r),
|
||||
patch("requests.get", return_value=r),
|
||||
patch("esphome.framework_helpers.ProgressBar") as mock_pb,
|
||||
):
|
||||
download_from_mirrors(["https://example.com/f"], {}, tmp_path / "out.bin")
|
||||
@@ -606,12 +606,35 @@ class TestDownloadFromMirrors:
|
||||
r.headers = {"content-length": "0"}
|
||||
r.iter_content.return_value = [b""] # one empty chunk
|
||||
target = tmp_path / "out.bin"
|
||||
with patch("esphome.framework_helpers.requests.get", return_value=r):
|
||||
with patch("requests.get", return_value=r):
|
||||
download_from_mirrors(["https://example.com/f"], {}, target)
|
||||
assert target.exists()
|
||||
assert target.read_bytes() == b""
|
||||
|
||||
|
||||
def test_importing_framework_helpers_does_not_import_requests() -> None:
|
||||
"""Importing framework_helpers must not drag in requests.
|
||||
|
||||
requests is a heavy import (~85ms) only needed by download_from_mirrors to
|
||||
fetch toolchains during a build. framework_helpers is loaded during config
|
||||
validation (esp-idf framework, host platform), so the import is deferred to
|
||||
the function that uses it. A fresh interpreter is required because the test
|
||||
process has already imported requests.
|
||||
"""
|
||||
result = subprocess.run(
|
||||
[
|
||||
sys.executable,
|
||||
"-c",
|
||||
"import sys\nimport esphome.framework_helpers\n"
|
||||
"print('\\n'.join(sys.modules))",
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
assert "requests" not in result.stdout.split()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_python_env_executable_path — Windows branch
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user