mirror of
https://github.com/esphome/esphome.git
synced 2026-07-10 17:05:36 +00:00
Simplify _read_etag, walrus the etag/new_etag binds, fixturize requests + helper patches
- _read_etag collapses into a single try/except that handles both stat and read failures (and the unlink) uniformly. - _mtime_seconds() helper compares mtimes at whole-second resolution instead of nanoseconds. Sub-second precision varies by filesystem (FAT does 2s, NTFS 100ns, APFS/ext4 ns); whole seconds is the only resolution every supported filesystem can preserve through a set+read round-trip. - has_remote_file_changed: walrus-assign `etag` and `new_etag` instead of an explicit binding + immediate check. - New test fixtures `mock_requests_head`, `mock_requests_get`, `mock_has_remote_file_changed`, `mock_write_file` replace the per-test `@patch(...)` decorator stacks. Tests are simpler and the fixtures can be reused by future tests in the file.
This commit is contained in:
+26
-30
@@ -31,37 +31,35 @@ def _etag_sidecar_path(local_file_path: Path) -> Path:
|
||||
return local_file_path.parent / f".{local_file_path.name}.etag"
|
||||
|
||||
|
||||
def _read_etag(local_file_path: Path) -> str | None:
|
||||
"""Return the cached ETag if it still describes the current cached file.
|
||||
def _mtime_seconds(path: Path) -> int:
|
||||
"""Return `path`'s mtime as integer seconds.
|
||||
|
||||
The sidecar's mtime is kept in sync with the cache file's mtime by
|
||||
`_write_etag`. If they disagree, the cache file was modified out-of-band
|
||||
(manual edit, restored from backup, replaced by another tool) and the
|
||||
ETag no longer corresponds to its contents, so treat the sidecar as
|
||||
stale and delete it.
|
||||
Whole seconds is the common-denominator resolution across all
|
||||
filesystems we run on (FAT/exFAT 2s, NTFS 100ns, APFS/ext4 ns), so
|
||||
comparisons survive setting+reading round-trips that would lose
|
||||
sub-second precision on lower-resolution filesystems.
|
||||
"""
|
||||
return int(path.stat().st_mtime)
|
||||
|
||||
|
||||
def _read_etag(local_file_path: Path) -> str | None:
|
||||
"""Return the cached ETag if its sidecar's mtime still matches the cache
|
||||
file's. A mismatch means the cache file was modified out-of-band, so the
|
||||
ETag no longer describes its contents -- delete the stale sidecar and
|
||||
return None.
|
||||
"""
|
||||
etag_path = _etag_sidecar_path(local_file_path)
|
||||
try:
|
||||
etag_mtime_ns = etag_path.stat().st_mtime_ns
|
||||
file_mtime_ns = local_file_path.stat().st_mtime_ns
|
||||
except OSError:
|
||||
return None
|
||||
if etag_mtime_ns != file_mtime_ns:
|
||||
_LOGGER.debug(
|
||||
"ETag sidecar mtime (%d) does not match cached file mtime (%d) at %s; "
|
||||
"treating ETag as stale",
|
||||
etag_mtime_ns,
|
||||
file_mtime_ns,
|
||||
local_file_path,
|
||||
)
|
||||
with contextlib.suppress(OSError):
|
||||
if _mtime_seconds(etag_path) != _mtime_seconds(local_file_path):
|
||||
_LOGGER.debug(
|
||||
"ETag sidecar mtime mismatch at %s; treating as stale",
|
||||
local_file_path,
|
||||
)
|
||||
etag_path.unlink()
|
||||
return None
|
||||
try:
|
||||
etag = etag_path.read_text().strip()
|
||||
return None
|
||||
return etag_path.read_text().strip() or None
|
||||
except OSError:
|
||||
return None
|
||||
return etag or None
|
||||
|
||||
|
||||
def _write_etag(local_file_path: Path, etag: str | None) -> None:
|
||||
@@ -79,8 +77,8 @@ def _write_etag(local_file_path: Path, etag: str | None) -> None:
|
||||
# Pin the sidecar's mtime to the cache file's mtime. _read_etag relies on
|
||||
# this match to detect out-of-band edits to the cache file.
|
||||
try:
|
||||
file_mtime_ns = local_file_path.stat().st_mtime_ns
|
||||
os.utime(etag_path, ns=(file_mtime_ns, file_mtime_ns))
|
||||
file_mtime = _mtime_seconds(local_file_path)
|
||||
os.utime(etag_path, (file_mtime, file_mtime))
|
||||
except OSError as e:
|
||||
_LOGGER.debug(
|
||||
"Could not sync ETag sidecar mtime for %s: %s", local_file_path, e
|
||||
@@ -100,8 +98,7 @@ def has_remote_file_changed(url: str, local_file_path: Path) -> bool:
|
||||
IF_MODIFIED_SINCE: local_modification_time_str,
|
||||
CACHE_CONTROL: CACHE_CONTROL_MAX_AGE + "3600",
|
||||
}
|
||||
etag = _read_etag(local_file_path)
|
||||
if etag:
|
||||
if etag := _read_etag(local_file_path):
|
||||
headers[IF_NONE_MATCH] = etag
|
||||
response = requests.head(
|
||||
url, headers=headers, timeout=NETWORK_TIMEOUT, allow_redirects=True
|
||||
@@ -120,8 +117,7 @@ def has_remote_file_changed(url: str, local_file_path: Path) -> bool:
|
||||
"has_remote_file_changed: File not modified since %s",
|
||||
local_modification_time_str,
|
||||
)
|
||||
new_etag = response.headers.get(ETAG)
|
||||
if new_etag and new_etag != etag:
|
||||
if (new_etag := response.headers.get(ETAG)) and new_etag != etag:
|
||||
_write_etag(local_file_path, new_etag)
|
||||
return False
|
||||
_LOGGER.debug("has_remote_file_changed: File modified")
|
||||
|
||||
@@ -19,11 +19,47 @@ def _seed_etag(cache_file: Path, etag: str) -> Path:
|
||||
"""
|
||||
sidecar = external_files._etag_sidecar_path(cache_file)
|
||||
sidecar.write_text(etag)
|
||||
file_mtime_ns = cache_file.stat().st_mtime_ns
|
||||
os.utime(sidecar, ns=(file_mtime_ns, file_mtime_ns))
|
||||
file_mtime = int(cache_file.stat().st_mtime)
|
||||
os.utime(sidecar, (file_mtime, file_mtime))
|
||||
return sidecar
|
||||
|
||||
|
||||
@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.
|
||||
"""
|
||||
with patch("esphome.external_files.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.
|
||||
"""
|
||||
with patch("esphome.external_files.requests.get") as m:
|
||||
yield m
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_has_remote_file_changed() -> MagicMock:
|
||||
"""Patch `external_files.has_remote_file_changed` so download tests can
|
||||
control the conditional check independently from the GET path.
|
||||
"""
|
||||
with patch("esphome.external_files.has_remote_file_changed") as m:
|
||||
yield m
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_write_file() -> MagicMock:
|
||||
"""Patch `external_files.write_file` so atomic-write failures can be
|
||||
injected without involving the real filesystem helper.
|
||||
"""
|
||||
with patch("esphome.external_files.write_file") as m:
|
||||
yield m
|
||||
|
||||
|
||||
def test_compute_local_file_dir(setup_core: Path) -> None:
|
||||
"""Test compute_local_file_dir creates and returns correct path."""
|
||||
domain = "font"
|
||||
@@ -100,9 +136,8 @@ def test_is_file_recent_with_zero_refresh(setup_core: Path) -> None:
|
||||
assert result is False
|
||||
|
||||
|
||||
@patch("esphome.external_files.requests.head")
|
||||
def test_has_remote_file_changed_not_modified(
|
||||
mock_head: MagicMock, setup_core: Path
|
||||
mock_requests_head: MagicMock, setup_core: Path
|
||||
) -> None:
|
||||
"""Test has_remote_file_changed returns False when file not modified."""
|
||||
test_file = setup_core / "cached.txt"
|
||||
@@ -111,23 +146,22 @@ def test_has_remote_file_changed_not_modified(
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 304
|
||||
mock_response.headers = {}
|
||||
mock_head.return_value = mock_response
|
||||
mock_requests_head.return_value = mock_response
|
||||
|
||||
url = "https://example.com/file.txt"
|
||||
result = external_files.has_remote_file_changed(url, test_file)
|
||||
|
||||
assert result is False
|
||||
mock_head.assert_called_once()
|
||||
mock_requests_head.assert_called_once()
|
||||
|
||||
call_args = mock_head.call_args
|
||||
call_args = mock_requests_head.call_args
|
||||
headers = call_args[1]["headers"]
|
||||
assert external_files.IF_MODIFIED_SINCE in headers
|
||||
assert external_files.CACHE_CONTROL in headers
|
||||
|
||||
|
||||
@patch("esphome.external_files.requests.head")
|
||||
def test_has_remote_file_changed_modified(
|
||||
mock_head: MagicMock, setup_core: Path
|
||||
mock_requests_head: MagicMock, setup_core: Path
|
||||
) -> None:
|
||||
"""Test has_remote_file_changed returns True when file modified."""
|
||||
test_file = setup_core / "cached.txt"
|
||||
@@ -136,7 +170,7 @@ def test_has_remote_file_changed_modified(
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.headers = {}
|
||||
mock_head.return_value = mock_response
|
||||
mock_requests_head.return_value = mock_response
|
||||
|
||||
url = "https://example.com/file.txt"
|
||||
result = external_files.has_remote_file_changed(url, test_file)
|
||||
@@ -154,15 +188,16 @@ def test_has_remote_file_changed_no_local_file(setup_core: Path) -> None:
|
||||
assert result is True
|
||||
|
||||
|
||||
@patch("esphome.external_files.requests.head")
|
||||
def test_has_remote_file_changed_network_error(
|
||||
mock_head: MagicMock, setup_core: Path
|
||||
mock_requests_head: MagicMock, setup_core: Path
|
||||
) -> None:
|
||||
"""Test has_remote_file_changed returns False on network error when file is cached."""
|
||||
test_file = setup_core / "cached.txt"
|
||||
test_file.write_text("cached content")
|
||||
|
||||
mock_head.side_effect = requests.exceptions.RequestException("Network error")
|
||||
mock_requests_head.side_effect = requests.exceptions.RequestException(
|
||||
"Network error"
|
||||
)
|
||||
|
||||
url = "https://example.com/file.txt"
|
||||
result = external_files.has_remote_file_changed(url, test_file)
|
||||
@@ -170,9 +205,8 @@ def test_has_remote_file_changed_network_error(
|
||||
assert result is False
|
||||
|
||||
|
||||
@patch("esphome.external_files.requests.head")
|
||||
def test_has_remote_file_changed_timeout(
|
||||
mock_head: MagicMock, setup_core: Path
|
||||
mock_requests_head: MagicMock, setup_core: Path
|
||||
) -> None:
|
||||
"""Test has_remote_file_changed respects timeout."""
|
||||
test_file = setup_core / "cached.txt"
|
||||
@@ -181,18 +215,17 @@ def test_has_remote_file_changed_timeout(
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 304
|
||||
mock_response.headers = {}
|
||||
mock_head.return_value = mock_response
|
||||
mock_requests_head.return_value = mock_response
|
||||
|
||||
url = "https://example.com/file.txt"
|
||||
external_files.has_remote_file_changed(url, test_file)
|
||||
|
||||
call_args = mock_head.call_args
|
||||
call_args = mock_requests_head.call_args
|
||||
assert call_args[1]["timeout"] == external_files.NETWORK_TIMEOUT
|
||||
|
||||
|
||||
@patch("esphome.external_files.requests.head")
|
||||
def test_has_remote_file_changed_uses_etag(
|
||||
mock_head: MagicMock, setup_core: Path
|
||||
mock_requests_head: MagicMock, setup_core: Path
|
||||
) -> None:
|
||||
"""Test has_remote_file_changed sends If-None-Match when ETag is cached."""
|
||||
test_file = setup_core / "cached.txt"
|
||||
@@ -202,19 +235,18 @@ def test_has_remote_file_changed_uses_etag(
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 304
|
||||
mock_response.headers = {}
|
||||
mock_head.return_value = mock_response
|
||||
mock_requests_head.return_value = mock_response
|
||||
|
||||
url = "https://example.com/file.txt"
|
||||
result = external_files.has_remote_file_changed(url, test_file)
|
||||
|
||||
assert result is False
|
||||
headers = mock_head.call_args[1]["headers"]
|
||||
headers = mock_requests_head.call_args[1]["headers"]
|
||||
assert headers[external_files.IF_NONE_MATCH] == '"abc123"'
|
||||
|
||||
|
||||
@patch("esphome.external_files.requests.head")
|
||||
def test_has_remote_file_changed_no_etag_no_if_none_match(
|
||||
mock_head: MagicMock, setup_core: Path
|
||||
mock_requests_head: MagicMock, setup_core: Path
|
||||
) -> None:
|
||||
"""Test has_remote_file_changed omits If-None-Match when no ETag is cached."""
|
||||
test_file = setup_core / "cached.txt"
|
||||
@@ -223,18 +255,17 @@ def test_has_remote_file_changed_no_etag_no_if_none_match(
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 304
|
||||
mock_response.headers = {}
|
||||
mock_head.return_value = mock_response
|
||||
mock_requests_head.return_value = mock_response
|
||||
|
||||
url = "https://example.com/file.txt"
|
||||
external_files.has_remote_file_changed(url, test_file)
|
||||
|
||||
headers = mock_head.call_args[1]["headers"]
|
||||
headers = mock_requests_head.call_args[1]["headers"]
|
||||
assert external_files.IF_NONE_MATCH not in headers
|
||||
|
||||
|
||||
@patch("esphome.external_files.requests.head")
|
||||
def test_has_remote_file_changed_refreshes_etag_on_304(
|
||||
mock_head: MagicMock, setup_core: Path
|
||||
mock_requests_head: MagicMock, setup_core: Path
|
||||
) -> None:
|
||||
"""Test has_remote_file_changed updates the cached ETag when the 304 sends a new one."""
|
||||
test_file = setup_core / "cached.txt"
|
||||
@@ -244,7 +275,7 @@ def test_has_remote_file_changed_refreshes_etag_on_304(
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 304
|
||||
mock_response.headers = {external_files.ETAG: '"new"'}
|
||||
mock_head.return_value = mock_response
|
||||
mock_requests_head.return_value = mock_response
|
||||
|
||||
url = "https://example.com/file.txt"
|
||||
external_files.has_remote_file_changed(url, test_file)
|
||||
@@ -252,9 +283,8 @@ def test_has_remote_file_changed_refreshes_etag_on_304(
|
||||
assert external_files._etag_sidecar_path(test_file).read_text() == '"new"'
|
||||
|
||||
|
||||
@patch("esphome.external_files.requests.head")
|
||||
def test_has_remote_file_changed_ignores_etag_when_mtime_diverges(
|
||||
mock_head: MagicMock, setup_core: Path
|
||||
mock_requests_head: MagicMock, setup_core: Path
|
||||
) -> None:
|
||||
"""If the cache file was edited out-of-band (mtime no longer matches the
|
||||
sidecar's), the cached ETag must not be used -- it no longer describes the
|
||||
@@ -264,49 +294,46 @@ def test_has_remote_file_changed_ignores_etag_when_mtime_diverges(
|
||||
test_file.write_text("cached content")
|
||||
sidecar = _seed_etag(test_file, '"abc123"')
|
||||
|
||||
# Simulate an out-of-band edit to the cache file -- mtime advances but the
|
||||
# sidecar is left untouched, so the recorded ETag is now stale.
|
||||
# Simulate an out-of-band edit to the cache file -- mtime advances by a
|
||||
# full second (so it diverges at whole-second resolution) but the sidecar
|
||||
# is left untouched, so the recorded ETag is now stale.
|
||||
file_stat = test_file.stat()
|
||||
os.utime(
|
||||
test_file, ns=(file_stat.st_atime_ns, file_stat.st_mtime_ns + 1_000_000_000)
|
||||
)
|
||||
os.utime(test_file, (file_stat.st_atime, file_stat.st_mtime + 1))
|
||||
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 304
|
||||
mock_response.headers = {}
|
||||
mock_head.return_value = mock_response
|
||||
mock_requests_head.return_value = mock_response
|
||||
|
||||
external_files.has_remote_file_changed("https://example.com/file.txt", test_file)
|
||||
|
||||
headers = mock_head.call_args[1]["headers"]
|
||||
headers = mock_requests_head.call_args[1]["headers"]
|
||||
assert external_files.IF_NONE_MATCH not in headers
|
||||
# Stale sidecar should be removed so future calls don't keep paying the
|
||||
# mtime-comparison cost on a known-bad sidecar.
|
||||
assert not sidecar.exists()
|
||||
|
||||
|
||||
@patch("esphome.external_files.requests.get")
|
||||
@patch("esphome.external_files.has_remote_file_changed")
|
||||
def test_download_content_pins_etag_mtime_to_file_mtime(
|
||||
mock_has_changed: MagicMock,
|
||||
mock_get: MagicMock,
|
||||
mock_has_remote_file_changed: MagicMock,
|
||||
mock_requests_get: MagicMock,
|
||||
setup_core: Path,
|
||||
) -> None:
|
||||
"""After a successful download, the sidecar's mtime must equal the cache
|
||||
file's mtime so `_read_etag` accepts it on the next call.
|
||||
"""
|
||||
test_file = setup_core / "fresh.txt"
|
||||
mock_has_changed.return_value = True
|
||||
mock_has_remote_file_changed.return_value = True
|
||||
mock_response = MagicMock()
|
||||
mock_response.content = b"fresh content"
|
||||
mock_response.headers = {external_files.ETAG: '"deadbeef"'}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
mock_get.return_value = mock_response
|
||||
mock_requests_get.return_value = mock_response
|
||||
|
||||
external_files.download_content("https://example.com/file.txt", test_file)
|
||||
|
||||
sidecar = external_files._etag_sidecar_path(test_file)
|
||||
assert sidecar.stat().st_mtime_ns == test_file.stat().st_mtime_ns
|
||||
assert int(sidecar.stat().st_mtime) == int(test_file.stat().st_mtime)
|
||||
|
||||
|
||||
def test_compute_local_file_dir_creates_parent_dirs(setup_core: Path) -> None:
|
||||
@@ -334,10 +361,10 @@ def test_is_file_recent_handles_float_seconds(setup_core: Path) -> None:
|
||||
assert result is True
|
||||
|
||||
|
||||
@patch("esphome.external_files.requests.get")
|
||||
@patch("esphome.external_files.has_remote_file_changed")
|
||||
def test_download_content_with_network_error_uses_cache(
|
||||
mock_has_changed: MagicMock, mock_get: MagicMock, setup_core: Path
|
||||
mock_has_remote_file_changed: MagicMock,
|
||||
mock_requests_get: MagicMock,
|
||||
setup_core: Path,
|
||||
) -> None:
|
||||
"""Test download_content uses cached file when network fails."""
|
||||
test_file = setup_core / "cached.txt"
|
||||
@@ -345,8 +372,10 @@ def test_download_content_with_network_error_uses_cache(
|
||||
test_file.write_bytes(cached_content)
|
||||
|
||||
# Simulate file has changed, so it tries to download
|
||||
mock_has_changed.return_value = True
|
||||
mock_get.side_effect = requests.exceptions.RequestException("Network error")
|
||||
mock_has_remote_file_changed.return_value = True
|
||||
mock_requests_get.side_effect = requests.exceptions.RequestException(
|
||||
"Network error"
|
||||
)
|
||||
|
||||
url = "https://example.com/file.txt"
|
||||
result = external_files.download_content(url, test_file)
|
||||
@@ -354,17 +383,19 @@ def test_download_content_with_network_error_uses_cache(
|
||||
assert result == cached_content
|
||||
|
||||
|
||||
@patch("esphome.external_files.requests.get")
|
||||
@patch("esphome.external_files.has_remote_file_changed")
|
||||
def test_download_content_with_network_error_no_cache_fails(
|
||||
mock_has_changed: MagicMock, mock_get: MagicMock, setup_core: Path
|
||||
mock_has_remote_file_changed: MagicMock,
|
||||
mock_requests_get: MagicMock,
|
||||
setup_core: Path,
|
||||
) -> None:
|
||||
"""Test download_content raises error when network fails and no cache exists."""
|
||||
test_file = setup_core / "nonexistent.txt"
|
||||
|
||||
# Simulate file has changed (doesn't exist), so it tries to download
|
||||
mock_has_changed.return_value = True
|
||||
mock_get.side_effect = requests.exceptions.RequestException("Network error")
|
||||
mock_has_remote_file_changed.return_value = True
|
||||
mock_requests_get.side_effect = requests.exceptions.RequestException(
|
||||
"Network error"
|
||||
)
|
||||
|
||||
url = "https://example.com/file.txt"
|
||||
|
||||
@@ -372,11 +403,9 @@ def test_download_content_with_network_error_no_cache_fails(
|
||||
external_files.download_content(url, test_file)
|
||||
|
||||
|
||||
@patch("esphome.external_files.requests.get")
|
||||
@patch("esphome.external_files.has_remote_file_changed")
|
||||
def test_download_content_skip_external_update_uses_cache(
|
||||
mock_has_changed: MagicMock,
|
||||
mock_get: MagicMock,
|
||||
mock_has_remote_file_changed: MagicMock,
|
||||
mock_requests_get: MagicMock,
|
||||
setup_core: Path,
|
||||
) -> None:
|
||||
"""Test download_content skips network checks when CORE.skip_external_update is set."""
|
||||
@@ -389,27 +418,25 @@ def test_download_content_skip_external_update_uses_cache(
|
||||
result = external_files.download_content(url, test_file)
|
||||
|
||||
assert result == cached_content
|
||||
mock_has_changed.assert_not_called()
|
||||
mock_get.assert_not_called()
|
||||
mock_has_remote_file_changed.assert_not_called()
|
||||
mock_requests_get.assert_not_called()
|
||||
|
||||
|
||||
@patch("esphome.external_files.requests.get")
|
||||
@patch("esphome.external_files.has_remote_file_changed")
|
||||
def test_download_content_skip_external_update_downloads_when_missing(
|
||||
mock_has_changed: MagicMock,
|
||||
mock_get: MagicMock,
|
||||
mock_has_remote_file_changed: MagicMock,
|
||||
mock_requests_get: MagicMock,
|
||||
setup_core: Path,
|
||||
) -> None:
|
||||
"""Test download_content still downloads when file is missing, even with skip_external_update."""
|
||||
test_file = setup_core / "missing.txt"
|
||||
new_content = b"fresh content"
|
||||
|
||||
mock_has_changed.return_value = True
|
||||
mock_has_remote_file_changed.return_value = True
|
||||
mock_response = MagicMock()
|
||||
mock_response.content = new_content
|
||||
mock_response.headers = {}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
mock_get.return_value = mock_response
|
||||
mock_requests_get.return_value = mock_response
|
||||
|
||||
CORE.skip_external_update = True
|
||||
url = "https://example.com/file.txt"
|
||||
@@ -419,23 +446,21 @@ def test_download_content_skip_external_update_downloads_when_missing(
|
||||
assert test_file.read_bytes() == new_content
|
||||
|
||||
|
||||
@patch("esphome.external_files.requests.get")
|
||||
@patch("esphome.external_files.has_remote_file_changed")
|
||||
def test_download_content_saves_etag(
|
||||
mock_has_changed: MagicMock,
|
||||
mock_get: MagicMock,
|
||||
mock_has_remote_file_changed: MagicMock,
|
||||
mock_requests_get: MagicMock,
|
||||
setup_core: Path,
|
||||
) -> None:
|
||||
"""Test download_content writes the ETag sidecar after a successful download."""
|
||||
test_file = setup_core / "fresh.txt"
|
||||
new_content = b"fresh content"
|
||||
|
||||
mock_has_changed.return_value = True
|
||||
mock_has_remote_file_changed.return_value = True
|
||||
mock_response = MagicMock()
|
||||
mock_response.content = new_content
|
||||
mock_response.headers = {external_files.ETAG: '"deadbeef"'}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
mock_get.return_value = mock_response
|
||||
mock_requests_get.return_value = mock_response
|
||||
|
||||
url = "https://example.com/file.txt"
|
||||
external_files.download_content(url, test_file)
|
||||
@@ -443,12 +468,9 @@ def test_download_content_saves_etag(
|
||||
assert external_files._etag_sidecar_path(test_file).read_text() == '"deadbeef"'
|
||||
|
||||
|
||||
@patch("esphome.external_files.write_file")
|
||||
@patch("esphome.external_files.requests.get")
|
||||
@patch("esphome.external_files.has_remote_file_changed")
|
||||
def test_download_content_atomic_write_no_partial_on_failure(
|
||||
mock_has_changed: MagicMock,
|
||||
mock_get: MagicMock,
|
||||
mock_has_remote_file_changed: MagicMock,
|
||||
mock_requests_get: MagicMock,
|
||||
mock_write_file: MagicMock,
|
||||
setup_core: Path,
|
||||
) -> None:
|
||||
@@ -463,12 +485,12 @@ def test_download_content_atomic_write_no_partial_on_failure(
|
||||
original_content = b"original content"
|
||||
test_file.write_bytes(original_content)
|
||||
|
||||
mock_has_changed.return_value = True
|
||||
mock_has_remote_file_changed.return_value = True
|
||||
mock_response = MagicMock()
|
||||
mock_response.content = b"new content"
|
||||
mock_response.headers = {}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
mock_get.return_value = mock_response
|
||||
mock_requests_get.return_value = mock_response
|
||||
|
||||
mock_write_file.side_effect = EsphomeError("disk full")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user