98 lines
3.7 KiB
Python
98 lines
3.7 KiB
Python
"""SpaceDevs caches survive empty files and failed disk writes."""
|
|
|
|
import errno
|
|
import json
|
|
import typing
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from agenda import thespacedevs
|
|
|
|
|
|
@pytest.mark.parametrize("contents", ["", "{", "[]", '{"detail": "throttled"}'])
|
|
def test_launch_cache_falls_back_to_usable_file(tmp_path: Path, contents: str) -> None:
|
|
old = datetime.now() - timedelta(days=2)
|
|
(tmp_path / old.strftime("%Y-%m-%d_%H:%M:%S.json")).write_text('{"results": []}')
|
|
(tmp_path / datetime.now().strftime("%Y-%m-%d_%H:%M:%S.json")).write_text(contents)
|
|
assert thespacedevs.load_cached_launches(str(tmp_path)) == {"results": []}
|
|
assert thespacedevs.read_cached_launches(str(tmp_path)) == []
|
|
assert not thespacedevs.is_launches_cache_fresh(str(tmp_path))
|
|
|
|
|
|
def test_empty_file_is_not_parsed(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
filename = tmp_path / "empty.json"
|
|
filename.touch()
|
|
parser = Mock(side_effect=AssertionError("empty file must not be parsed"))
|
|
monkeypatch.setattr("agenda.thespacedevs.json.load", parser)
|
|
assert thespacedevs.read_json_cache(str(filename)) is None
|
|
parser.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("existing", [False, True])
|
|
@pytest.mark.parametrize("failure", ["fsync", "replace"])
|
|
def test_disk_failure_preserves_cache(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, existing: bool, failure: str
|
|
) -> None:
|
|
filename = tmp_path / "cache.json"
|
|
original = '{"results": ["old"]}'
|
|
if existing:
|
|
filename.write_text(original)
|
|
fail = Mock(side_effect=OSError(errno.ENOSPC, "No space left on device"))
|
|
monkeypatch.setattr(f"agenda.thespacedevs.os.{failure}", fail)
|
|
with pytest.raises(OSError, match="No space left"):
|
|
thespacedevs.write_json_cache(str(filename), {"results": ["new"]})
|
|
if existing:
|
|
assert filename.read_text() == original
|
|
else:
|
|
assert not filename.exists()
|
|
assert not list(tmp_path.glob(".cache-*"))
|
|
|
|
|
|
def test_successful_write_replaces_empty_file(tmp_path: Path) -> None:
|
|
filename = tmp_path / "cache.json"
|
|
filename.touch()
|
|
thespacedevs.write_json_cache(str(filename), {"results": []})
|
|
assert json.loads(filename.read_text()) == {"results": []}
|
|
assert not list(tmp_path.glob(".cache-*"))
|
|
|
|
|
|
def test_all_empty_launch_files_return_no_cache(tmp_path: Path) -> None:
|
|
(tmp_path / "2026-09-30_12:00:00.json").touch()
|
|
assert thespacedevs.load_cached_launches(str(tmp_path)) is None
|
|
assert not thespacedevs.is_launches_cache_fresh(str(tmp_path))
|
|
|
|
|
|
@pytest.mark.parametrize("existing_mode", [None, 0o600, 0o644])
|
|
def test_published_cache_is_readable_by_web_server(
|
|
tmp_path: Path, existing_mode: int | None
|
|
) -> None:
|
|
filename = tmp_path / "cache.json"
|
|
if existing_mode is not None:
|
|
filename.touch(mode=existing_mode)
|
|
thespacedevs.write_json_cache(str(filename), {"results": []})
|
|
assert filename.stat().st_mode & 0o777 == 0o644
|
|
|
|
|
|
def test_unreadable_cache_falls_back(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
older = tmp_path / "2026-09-29_12:00:00.json"
|
|
newer = tmp_path / "2026-09-30_12:00:00.json"
|
|
older.write_text('{"results": []}')
|
|
newer.write_text('{"results": []}')
|
|
import builtins
|
|
|
|
original_open = builtins.open
|
|
|
|
def open_cache(filename: str) -> typing.TextIO:
|
|
if filename == str(newer):
|
|
raise PermissionError(errno.EACCES, "Permission denied", filename)
|
|
return original_open(filename)
|
|
|
|
monkeypatch.setattr("agenda.thespacedevs.open", open_cache, raising=False)
|
|
assert thespacedevs.load_cached_launches(str(tmp_path)) == {"results": []}
|