66 lines
2.6 KiB
Python
66 lines
2.6 KiB
Python
from pathlib import Path
|
|
|
|
from fastapi import FastAPI
|
|
from starlette.responses import FileResponse, JSONResponse, Response
|
|
from starlette.testclient import TestClient
|
|
|
|
from k1link.web.response_compression import ResponseCompressionMiddleware
|
|
from k1link.web.session_api import _ReleasingFileResponse
|
|
|
|
|
|
def test_rrd_preserves_bytes_length_etag_and_ranges_with_gzip_client(tmp_path: Path) -> None:
|
|
payload = b"RRD compressed fixture" * 1024
|
|
path = tmp_path / "scene.rrd"
|
|
path.write_bytes(payload)
|
|
app = FastAPI()
|
|
app.add_middleware(ResponseCompressionMiddleware)
|
|
releases = []
|
|
|
|
@app.get("/scene.rrd")
|
|
def recording() -> FileResponse:
|
|
return _ReleasingFileResponse(
|
|
path,
|
|
headers={"ETag": '"fixture-generation"'},
|
|
release=lambda: releases.append(True),
|
|
)
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get("/scene.rrd?generation=fixture", headers={"Accept-Encoding": "gzip"})
|
|
assert response.content == payload
|
|
assert "content-encoding" not in response.headers
|
|
assert int(response.headers["content-length"]) == len(payload)
|
|
assert response.headers["etag"] == '"fixture-generation"'
|
|
partial = client.get(
|
|
"/scene.rrd", headers={"Accept-Encoding": "gzip", "Range": "bytes=100-2199"}
|
|
)
|
|
assert partial.status_code == 206
|
|
assert partial.content == payload[100:2200]
|
|
assert "content-encoding" not in partial.headers
|
|
assert partial.headers["content-range"] == f"bytes 100-2199/{len(payload)}"
|
|
assert partial.headers["content-length"] == "2100"
|
|
assert releases == [True, True]
|
|
assert _ReleasingFileResponse.chunk_size == 1024 * 1024
|
|
|
|
|
|
def test_blueprint_and_color_posts_bypass_gzip_but_json_keeps_it() -> None:
|
|
app = FastAPI()
|
|
app.add_middleware(ResponseCompressionMiddleware)
|
|
payload = b"native RRD" * 1024
|
|
|
|
@app.post("/{kind}.rrd")
|
|
def overlay(kind: str) -> Response:
|
|
return Response(payload, media_type="application/octet-stream")
|
|
|
|
@app.get("/metadata")
|
|
def metadata() -> JSONResponse:
|
|
return JSONResponse({"description": "a" * 8192})
|
|
|
|
with TestClient(app) as client:
|
|
for kind in ("blueprint", "point-colors"):
|
|
response = client.post(f"/{kind}.rrd", headers={"Accept-Encoding": "gzip"})
|
|
assert response.content == payload
|
|
assert "content-encoding" not in response.headers
|
|
response = client.get("/metadata", headers={"Accept-Encoding": "gzip"})
|
|
assert response.headers["content-encoding"] == "gzip"
|
|
assert response.json() == {"description": "a" * 8192}
|