upbeatBytes/tests/test_newsimg.py

"""News image cache (hardened): SSRF-safe cycle-only fetch + downscale to WebP, reject
SVG/non-raster/private-host/redirect-to-private, negative-cache failures, LRU eviction,
and the cache-HITS-ONLY /api/img endpoint (restricted to accepted, canonical articles)."""
import io
import os

import pytest
from PIL import Image
from fastapi.testclient import TestClient

from goodnews import newsimg
from goodnews.db import connect, init_db

# The real SSRF-safe fetch, captured before any fixture stubs it — the SSRF tests below
# restore it so they exercise the genuine host/redirect validation.
_REAL_SAFE_FETCH = newsimg._safe_fetch


def _png(w=1600, h=1000, color=(40, 130, 173)):
    out = io.BytesIO()
    Image.new("RGB", (w, h), color).save(out, format="PNG")
    return out.getvalue()


@pytest.fixture
def cache(tmp_path, monkeypatch):
    monkeypatch.setenv("GOODNEWS_IMG_CACHE", str(tmp_path / "img_cache"))
    # default: treat hosts as public + a happy PNG fetch; individual tests override.
    monkeypatch.setattr(newsimg, "_host_is_public", lambda h: True)
    monkeypatch.setattr(newsimg, "_safe_fetch", lambda url, timeout=12: (_png(), "image/png"))
    return tmp_path / "img_cache"


def test_fetch_and_cache_downscales_to_webp(cache):
    p = newsimg.fetch_and_cache("https://example.com/big.png")
    assert p and p.exists() and p.suffix == ".webp"
    with Image.open(p) as im:
        assert im.width == newsimg.DISPLAY_WIDTH and im.format == "WEBP"
    assert newsimg.path_for("https://example.com/big.png") == p   # pure cache lookup


def test_bad_scheme_and_empty_are_rejected(cache):
    assert newsimg.fetch_and_cache(None) is None
    assert newsimg.fetch_and_cache("ftp://example.com/x.png") is None
    assert not list(cache.glob("*.webp"))


def test_svg_and_nonimage_rejected_and_negative_cached(cache, monkeypatch):
    monkeypatch.setattr(newsimg, "_safe_fetch",
                        lambda url, timeout=12: (b"<svg xmlns='http://www.w3.org/2000/svg'></svg>" + b" " * 600,
                                                 "image/svg+xml"))
    url = "https://example.com/evil.svg"
    assert newsimg.fetch_and_cache(url) is None          # decoded-raster only → SVG refused
    assert not list(cache.glob("*.webp"))                # nothing stored
    assert newsimg._failed_recently(url)                 # negative-cached (won't refetch)


def test_private_host_refused_and_negative_cached(cache, monkeypatch):
    monkeypatch.setattr(newsimg, "_safe_fetch", _REAL_SAFE_FETCH)   # exercise real validation
    monkeypatch.setattr(newsimg, "_host_is_public", lambda h: False)
    url = "http://169.254.169.254/latest/meta-data/"
    assert newsimg.fetch_and_cache(url) is None
    assert newsimg._failed_recently(url)                 # SSRF target → permanent fail


def test_redirect_to_private_is_refused(cache, monkeypatch):
    monkeypatch.setattr(newsimg, "_safe_fetch", _REAL_SAFE_FETCH)   # exercise real redirect re-validation
    hosts = {"good.example": True, "evil.internal": False}
    monkeypatch.setattr(newsimg, "_host_is_public", lambda h: hosts.get(h, False))

    class _Resp:
        def __init__(self, status, headers, body=b""):
            self.status, self.headers, self._b = status, headers, body
        def read(self, n=-1):
            return self._b
        def close(self):
            pass

    class _Opener:
        def open(self, req, timeout=None):
            if "good.example" in req.full_url:                       # public → 302 to private
                return _Resp(302, {"Location": "http://evil.internal/x.png"})
            return _Resp(200, {"Content-Type": "image/png"}, _png())  # should never be reached
    monkeypatch.setattr(newsimg.urllib.request, "build_opener", lambda *a: _Opener())

    url = "http://good.example/p.png"
    assert newsimg.fetch_and_cache(url) is None          # redirect hop re-validated → refused
    assert newsimg._failed_recently(url)


def test_transient_failure_is_not_negative_cached(cache, monkeypatch):
    def boom(url, timeout=12):
        raise newsimg._FetchError("network down", permanent=False)
    monkeypatch.setattr(newsimg, "_safe_fetch", boom)
    url = "https://example.com/x.jpg"
    assert newsimg.fetch_and_cache(url) is None
    assert not newsimg._failed_recently(url)             # transient → retried next cycle


def test_prune_evicts_least_recently_used_over_cap(cache):
    paths = [newsimg.fetch_and_cache(f"https://example.com/{i}.png") for i in range(5)]
    for i, p in enumerate(paths):                        # 0 = oldest/LRU, 4 = newest
        os.utime(p, (1000 + i, 1000 + i))
    sizes = [p.stat().st_size for p in paths]
    cap = sum(sizes) - sizes[0] - sizes[1] + 1            # room for the 3 newest only
    r = newsimg.prune(cap)
    assert r["removed"] == 2 and r["after"] <= cap
    assert not paths[0].exists() and not paths[1].exists()
    assert paths[2].exists() and paths[4].exists()


def test_warm_skips_cached_and_failed(cache, monkeypatch):
    conn = connect(":memory:"); init_db(conn)
    conn.execute("INSERT INTO sources (id,name,feed_url) VALUES (1,'S','http://s/f')")
    for aid, img, acc, dup in ((1, "https://x/1.jpg", 1, None), (2, "https://x/2.jpg", 1, None),
                               (3, "https://x/3.jpg", 0, None), (4, "https://x/4.jpg", 1, 1)):
        conn.execute("INSERT INTO articles (id,source_id,canonical_url,title,url_hash,image_url,duplicate_of) "
                     "VALUES (?,1,?,?,?,?,?)", (aid, f"http://s/{aid}", f"t{aid}", f"h{aid}", img, dup))
        conn.execute("INSERT INTO article_scores (article_id, accepted) VALUES (?,?)", (aid, acc))
    conn.commit()
    assert newsimg.warm(conn) == 2                        # only the 2 accepted, canonical, with image
    assert newsimg.warm(conn) == 0                        # idempotent — already cached


@pytest.fixture
def client(tmp_path, monkeypatch):
    monkeypatch.setenv("GOODNEWS_DB", str(tmp_path / "t.sqlite3"))
    monkeypatch.setenv("GOODNEWS_IMG_CACHE", str(tmp_path / "img_cache"))
    monkeypatch.setattr(newsimg, "_host_is_public", lambda h: True)
    monkeypatch.setattr(newsimg, "_safe_fetch", lambda url, timeout=12: (_png(), "image/png"))
    conn = connect(tmp_path / "t.sqlite3"); init_db(conn)
    conn.execute("INSERT INTO sources (id,name,feed_url) VALUES (1,'S','http://s/f')")
    # 1 = accepted+canonical+image, 2 = no image, 3 = NOT accepted, 4 = duplicate
    rows = ((1, "https://x/pic.jpg", 1, None), (2, "", 1, None),
            (3, "https://x/p3.jpg", 0, None), (4, "https://x/p4.jpg", 1, 1))
    for aid, img, acc, dup in rows:
        conn.execute("INSERT INTO articles (id,source_id,canonical_url,title,url_hash,image_url,duplicate_of) "
                     "VALUES (?,1,?,?,?,?,?)", (aid, f"http://s/{aid}", f"t{aid}", f"h{aid}", img, dup))
        conn.execute("INSERT INTO article_scores (article_id, accepted) VALUES (?,?)", (aid, acc))
    conn.commit()
    # the cycle owns fetching — pre-warm so the endpoint has a hit to serve
    newsimg.warm(conn)
    conn.close()
    from goodnews.api import create_app
    return TestClient(create_app())


def test_img_endpoint_serves_cached_and_restricts(client):
    r = client.get("/api/img/1")
    assert r.status_code == 200 and r.headers["content-type"] == "image/webp"
    assert "immutable" not in r.headers.get("cache-control", "")   # no 1-year immutable
    assert client.get("/api/img/2").status_code == 404             # no image
    assert client.get("/api/img/3").status_code == 404             # not accepted
    assert client.get("/api/img/4").status_code == 404             # duplicate (non-canonical)
    assert client.get("/api/img/999").status_code == 404           # unknown id


def test_img_endpoint_never_fetches(client, monkeypatch):
    # An accepted article whose image was never warmed must 404, not trigger a fetch.
    called = {"n": 0}
    monkeypatch.setattr(newsimg, "fetch_and_cache", lambda *a, **k: called.__setitem__("n", called["n"] + 1))
    # /api/img/1 is cached (warmed in fixture) → still served without any fetch
    assert client.get("/api/img/1").status_code == 200
    assert called["n"] == 0