tinyproject-monorepo/prg/sjaur/tests/test_cache.py

"""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"