state-hub/tests/test_edge_read_cache.py

import httpx
import pytest
from httpx import ASGITransport, AsyncClient

from api.edge.read_cache import ReadCacheStore, build_cache_key, is_cacheable_get
from api.edge.relay import create_app


class FailingAsyncClient:
    def __init__(self, *args, **kwargs):
        pass

    async def __aenter__(self):
        return self

    async def __aexit__(self, *exc_info):
        return False

    async def request(self, *args, **kwargs):
        raise httpx.ConnectError("upstream down")

    async def get(self, *args, **kwargs):
        raise httpx.ConnectError("upstream down")


class SuccessAsyncClient:
    def __init__(self, *args, **kwargs):
        pass

    async def __aenter__(self):
        return self

    async def __aexit__(self, *exc_info):
        return False

    async def request(self, method, path, **kwargs):
        request = httpx.Request(method, f"http://upstream{path}")
        return httpx.Response(
            200,
            json={"path": path, "ok": True},
            request=request,
            headers={"content-type": "application/json"},
        )


def test_build_cache_key_sorts_query_params() -> None:
    key = build_cache_key("GET", "/state/summary", [("b", "2"), ("a", "1")])
    assert key == "GET:/state/summary?a=1&b=2"


def test_is_cacheable_get_allows_activity_core_reads() -> None:
    assert is_cacheable_get("/state/summary")
    assert is_cacheable_get("/workplans/index")
    assert is_cacheable_get("/tasks/")
    assert is_cacheable_get("/edge/health") is False


def test_read_cache_store_round_trip(tmp_path) -> None:
    store = ReadCacheStore(tmp_path / "cache.sqlite3")
    store.put(
        cache_key="GET:/state/summary",
        status_code=200,
        content=b'{"ok": true}',
        content_type="application/json",
    )
    entry = store.get("GET:/state/summary")
    assert entry is not None
    assert entry.status_code == 200
    assert entry.content == b'{"ok": true}'


@pytest.mark.asyncio
async def test_relay_caches_successful_get_and_serves_stale_on_outage(tmp_path, monkeypatch):
    from api.edge import relay

    monkeypatch.setattr(relay.httpx, "AsyncClient", SuccessAsyncClient)
    cache_path = tmp_path / "read-cache.sqlite3"
    app = create_app(
        upstream_url="http://upstream",
        outbox_path=str(tmp_path / "outbox.sqlite3"),
        read_cache_path=str(cache_path),
    )

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://edge") as client:
        warm = await client.get("/state/summary")
    assert warm.status_code == 200
    assert warm.headers.get("x-statehub-edge-cache") == "hit"

    monkeypatch.setattr(relay.httpx, "AsyncClient", FailingAsyncClient)
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://edge") as client:
        stale = await client.get("/state/summary")

    assert stale.status_code == 200
    assert stale.headers.get("x-statehub-edge-cache") == "stale"
    assert stale.json()["ok"] is True
    assert int(stale.headers.get("x-statehub-edge-cache-age", "0")) >= 0


@pytest.mark.asyncio
async def test_relay_returns_503_for_uncached_get_when_upstream_unreachable(tmp_path, monkeypatch):
    from api.edge import relay

    monkeypatch.setattr(relay.httpx, "AsyncClient", FailingAsyncClient)
    app = create_app(
        upstream_url="http://upstream",
        outbox_path=str(tmp_path / "outbox.sqlite3"),
        read_cache_path=str(tmp_path / "read-cache.sqlite3"),
    )

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://edge") as client:
        response = await client.get("/state/summary")

    assert response.status_code == 503
    assert "not queueable" in response.json()["error"]