tegwick
1cf949bda4
The edge relay now persists successful GET responses and serves them with stale markers when upstream is unreachable. Extend Forgejo image workflow path filters so api changes trigger registry publishes.
114 lines
No EOL
3.6 KiB
Python
114 lines
No EOL
3.6 KiB
Python
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"] |