commit local changes.

2026-10-02 15:29:35 +02:00
parent 92e5b6e363
commit 95af3a84af
24 changed files with 3187 additions and 78 deletions
@@ -0,0 +1,277 @@
+"""Tests for the cache manager.
+
+The downloader is monkeypatched, so no test touches the network. Atomic
+replacement, freshness handling, corruption recovery, and stale fallback are
+all covered.
+"""
+
+from __future__ import annotations
+
+import gzip
+import json
+import os
+import time
+from pathlib import Path
+
+import pytest
+
+from sjaur.cache import CacheManager, download_file, humanize_delta
+from sjaur.errors import CacheCorruptError, DownloadError, SjaurError
+
+DATA_V1 = [{"Name": "firefox"}, {"Name": "python"}]
+DATA_V2 = [{"Name": "firefox"}, {"Name": "python"}, {"Name": "neovim"}]
+
+
+def gzipped(data) -> bytes:
+    return gzip.compress(json.dumps(data).encode())
+
+
+def install_downloader(monkeypatch, data) -> None:
+    """Make ``download_file`` produce a valid dump instead of using network."""
+
+    def fake_download(url, dest, progress=None, timeout=None):
+        dest.write_bytes(gzipped(data))
+
+    monkeypatch.setattr("sjaur.cache.download_file", fake_download)
+
+
+def install_failing_downloader(monkeypatch, message="Connection timed out.") -> None:
+    def fake_download(url, dest, progress=None, timeout=None):
+        raise DownloadError(message)
+
+    monkeypatch.setattr("sjaur.cache.download_file", fake_download)
+
+
+def write_cache_file(cache: CacheManager, content: bytes) -> None:
+    cache.cache_dir.mkdir(parents=True, exist_ok=True)
+    cache.metadata_path.write_bytes(content)
+
+
+def make_cache_stale(cache: CacheManager, hours: float = 25.0) -> None:
+    old = time.time() - hours * 3600
+    os.utime(cache.metadata_path, (old, old))
+
+
+def no_tmp_files(cache: CacheManager) -> bool:
+    return list(cache.cache_dir.glob("*.tmp")) == []
+
+
+# -- fresh/missing/expired ----------------------------------------------------
+
+
+def test_missing_cache_is_downloaded(tmp_path, monkeypatch):
+    cache = CacheManager(tmp_path / "cache")
+    install_downloader(monkeypatch, DATA_V1)
+
+    bundle = cache.ensure_fresh()
+
+    assert bundle.source == "downloaded"
+    assert bundle.data == DATA_V1
+    assert bundle.warning is None
+    assert cache.metadata_path.is_file()
+    assert no_tmp_files(cache)
+
+
+def test_fresh_cache_is_used_without_download(tmp_path, monkeypatch):
+    cache = CacheManager(tmp_path / "cache")
+    write_cache_file(cache, json.dumps(DATA_V1).encode())
+
+    def boom(*args, **kwargs):
+        raise AssertionError("download must not run for a fresh cache")
+
+    monkeypatch.setattr("sjaur.cache.download_file", boom)
+    bundle = cache.ensure_fresh()
+
+    assert bundle.source == "cache"
+    assert bundle.data == DATA_V1
+    assert cache.cache_age() is not None and cache.cache_age().total_seconds() < 60
+
+
+def test_expired_cache_is_replaced(tmp_path, monkeypatch):
+    cache = CacheManager(tmp_path / "cache")
+    write_cache_file(cache, json.dumps(DATA_V1).encode())
+    make_cache_stale(cache)
+    install_downloader(monkeypatch, DATA_V2)
+
+    bundle = cache.ensure_fresh()
+
+    assert bundle.source == "downloaded"
+    assert bundle.data == DATA_V2
+    assert json.loads(cache.metadata_path.read_text()) == DATA_V2
+    assert cache.cache_age().total_seconds() < 60
+    assert no_tmp_files(cache)
+
+
+def test_force_bypasses_freshness(tmp_path, monkeypatch):
+    cache = CacheManager(tmp_path / "cache")
+    write_cache_file(cache, json.dumps(DATA_V1).encode())
+    install_downloader(monkeypatch, DATA_V2)
+
+    bundle = cache.ensure_fresh(force=True)
+
+    assert bundle.source == "downloaded"
+    assert bundle.data == DATA_V2
+
+
+# -- download failures ---------------------------------------------------------
+
+
+def test_failed_download_keeps_stale_cache(tmp_path, monkeypatch):
+    cache = CacheManager(tmp_path / "cache")
+    write_cache_file(cache, json.dumps(DATA_V1).encode())
+    make_cache_stale(cache)
+    install_failing_downloader(monkeypatch, "Connection timed out.")
+
+    bundle = cache.ensure_fresh()
+
+    assert bundle.source == "stale-cache"
+    assert bundle.data == DATA_V1
+    assert bundle.warning is not None
+    assert "Connection timed out." in bundle.warning
+    assert "Using existing cached metadata" in bundle.warning
+    # Old cache untouched, no leftovers.
+    assert json.loads(cache.metadata_path.read_text()) == DATA_V1
+    assert no_tmp_files(cache)
+
+
+def test_failed_download_without_cache_raises(tmp_path, monkeypatch):
+    cache = CacheManager(tmp_path / "cache")
+    install_failing_downloader(monkeypatch)
+
+    with pytest.raises(SjaurError) as excinfo:
+        cache.ensure_fresh()
+    assert "timed out" in str(excinfo.value)
+    assert not cache.metadata_path.exists()
+
+
+def test_forced_failed_download_falls_back_to_fresh_cache(tmp_path, monkeypatch):
+    cache = CacheManager(tmp_path / "cache")
+    write_cache_file(cache, json.dumps(DATA_V1).encode())
+    install_failing_downloader(monkeypatch)
+
+    bundle = cache.ensure_fresh(force=True)
+
+    assert bundle.source == "stale-cache"
+    assert bundle.data == DATA_V1
+
+
+# -- corruption handling --------------------------------------------------------
+
+
+def test_corrupt_gzip_download_is_rejected_without_touching_cache(tmp_path, monkeypatch):
+    cache = CacheManager(tmp_path / "cache")
+    write_cache_file(cache, json.dumps(DATA_V1).encode())
+    make_cache_stale(cache)
+
+    def corrupt_download(url, dest, progress=None, timeout=None):
+        dest.write_bytes(b"this is not gzip data")
+
+    monkeypatch.setattr("sjaur.cache.download_file", corrupt_download)
+
+    bundle = cache.ensure_fresh()
+
+    assert bundle.source == "stale-cache"
+    assert bundle.data == DATA_V1
+    assert json.loads(cache.metadata_path.read_text()) == DATA_V1
+    assert no_tmp_files(cache)
+
+
+def test_fresh_but_corrupt_cache_is_replaced(tmp_path, monkeypatch):
+    cache = CacheManager(tmp_path / "cache")
+    write_cache_file(cache, b"{not valid json")
+    install_downloader(monkeypatch, DATA_V2)
+
+    bundle = cache.ensure_fresh()
+
+    assert bundle.source == "downloaded"
+    assert bundle.data == DATA_V2
+
+
+def test_fresh_but_corrupt_cache_and_failed_download_raises(tmp_path, monkeypatch):
+    cache = CacheManager(tmp_path / "cache")
+    write_cache_file(cache, b"{not valid json")
+    install_failing_downloader(monkeypatch)
+
+    with pytest.raises(SjaurError):
+        cache.ensure_fresh()
+
+
+def test_structurally_wrong_cache_is_rejected(tmp_path):
+    cache = CacheManager(tmp_path / "cache")
+    write_cache_file(cache, json.dumps({"not": "a list"}).encode())
+
+    with pytest.raises(CacheCorruptError):
+        cache._load()
+
+
+def test_empty_dump_is_valid_but_yields_no_packages(tmp_path, monkeypatch):
+    cache = CacheManager(tmp_path / "cache")
+    install_downloader(monkeypatch, [])
+
+    bundle = cache.ensure_fresh()
+
+    assert bundle.data == []
+
+
+def test_unwritable_cache_dir_raises_friendly_error(tmp_path):
+    # A *file* where the cache directory should be triggers a mkdir failure.
+    blocker = tmp_path / "blocker"
+    blocker.write_text("I am a file, not a directory")
+    cache = CacheManager(blocker)
+
+    with pytest.raises(SjaurError) as excinfo:
+        cache.ensure_fresh()
+    assert "cache directory" in str(excinfo.value)
+
+
+# -- download_file unit tests (local file:// URL, no network) --------------------
+
+
+def test_download_file_copies_content(tmp_path):
+    source = tmp_path / "source.gz"
+    source.write_bytes(gzipped(DATA_V1))
+    dest = tmp_path / "dest.gz"
+
+    download_file(source.as_uri(), dest)
+
+    assert dest.read_bytes() == source.read_bytes()
+
+
+def test_download_file_missing_source_raises(tmp_path):
+    dest = tmp_path / "dest.gz"
+    with pytest.raises(DownloadError):
+        download_file((tmp_path / "missing.gz").as_uri(), dest)
+
+
+def test_progress_callback_reports_bytes(tmp_path):
+    source = tmp_path / "source.gz"
+    source.write_bytes(gzipped(DATA_V1))
+    dest = tmp_path / "dest.gz"
+    seen = []
+
+    download_file(source.as_uri(), dest, progress=lambda done, total: seen.append((done, total)))
+
+    assert seen and seen[-1][0] == source.stat().st_size
+
+
+# -- helpers ---------------------------------------------------------------------
+
+
+def test_humanize_delta():
+    from datetime import timedelta
+
+    assert humanize_delta(timedelta(seconds=10)) == "10 seconds"
+    assert humanize_delta(timedelta(minutes=5)) == "5 minutes"
+    assert "hours" in humanize_delta(timedelta(hours=3))
+    assert "days" in humanize_delta(timedelta(days=3))
+
+
+def test_cache_age_missing_file(tmp_path):
+    cache = CacheManager(tmp_path / "cache")
+    assert cache.cache_age() is None
+    assert not cache.is_fresh()
+
+
+def test_metadata_filename(tmp_path):
+    cache = CacheManager(tmp_path)
+    assert cache.metadata_path == Path(tmp_path) / "packages-meta-ext-v1.json"