Spaces:
Running on Zero
Running on Zero
| """FastAPI contracts that do not download model weights.""" | |
| from pathlib import Path | |
| from urllib.parse import parse_qs, urlsplit | |
| from fastapi.testclient import TestClient | |
| from app import create_app | |
| from config import Settings | |
| from core.loader import ModelLoader | |
| def make_app(tmp_path: Path, *, max_upload_mb: int = 100, asset_signing_secret: str | None = None): | |
| ModelLoader.reset_instance() | |
| settings = Settings( | |
| output_folder=tmp_path / "output", | |
| tmp_folder=tmp_path / "tmp", | |
| static_folder=tmp_path / "static", | |
| device="cpu", | |
| max_upload_mb=max_upload_mb, | |
| asset_signing_secret=asset_signing_secret, | |
| ) | |
| return create_app(settings) | |
| def test_health_does_not_load_a_model(tmp_path: Path) -> None: | |
| application = make_app(tmp_path) | |
| with TestClient(application) as client: | |
| response = client.get("/health") | |
| assert response.status_code == 200 | |
| assert response.json()["status"] == "ok" | |
| assert response.json()["active_model"] is None | |
| def test_root_redirects_to_gradio_ui(tmp_path: Path) -> None: | |
| application = make_app(tmp_path) | |
| with TestClient(application) as client: | |
| response = client.get("/?__theme=dark", follow_redirects=False) | |
| direct_response = client.get("/ui", follow_redirects=False) | |
| assert response.status_code == 307 | |
| assert response.headers["location"] == "/ui/?__theme=dark" | |
| assert direct_response.status_code == 307 | |
| assert direct_response.headers["location"] == "/ui/" | |
| def test_mounted_gradio_ui_runtime_options(tmp_path: Path) -> None: | |
| application = make_app(tmp_path) | |
| mount = next( | |
| route | |
| for route in application.routes | |
| if route.path == "/ui" and hasattr(route.app, "get_blocks") | |
| ) | |
| blocks = mount.app.get_blocks() | |
| assert blocks.ssr_mode is False | |
| assert blocks.mcp_server is True | |
| def test_validation_errors_have_stable_shape(tmp_path: Path) -> None: | |
| application = make_app(tmp_path) | |
| with TestClient(application) as client: | |
| response = client.post( | |
| "/api/image", | |
| json={"prompt": "city", "width": 257, "height": 768, "steps": 4, "seed": 1}, | |
| ) | |
| assert response.status_code == 400 | |
| assert response.json()["success"] is False | |
| assert response.json()["code"] == "invalid_dimensions" | |
| assert response.headers["X-Request-ID"] | |
| def test_transcription_rejects_oversized_content_length_before_multipart_parse( | |
| tmp_path: Path, | |
| ) -> None: | |
| application = make_app(tmp_path, max_upload_mb=1) | |
| with TestClient(application) as client: | |
| response = client.post( | |
| "/transcribe", | |
| content=b"not-parsed", | |
| headers={ | |
| "content-type": "multipart/form-data; boundary=unused", | |
| "content-length": str(3 * 1024 * 1024), | |
| }, | |
| ) | |
| assert response.status_code == 413 | |
| assert response.json()["code"] == "upload_too_large" | |
| assert response.headers["X-Request-ID"] | |
| def test_signed_asset_is_publicly_fetchable_without_api_auth(tmp_path: Path) -> None: | |
| application = make_app(tmp_path, asset_signing_secret="test-secret") | |
| outputs = application.state.gateway.outputs | |
| target = outputs.allocate("images") | |
| target.write_bytes(b"asset-bytes") | |
| with TestClient(application) as client: | |
| url = outputs.signed_url(target) | |
| response = client.get(url) | |
| raw_output = client.get(outputs.public_path(target)) | |
| assert response.status_code == 200 | |
| assert response.content == b"asset-bytes" | |
| assert response.headers["content-type"].startswith("image/png") | |
| assert raw_output.status_code == 404 | |
| def test_signed_asset_rejects_tampered_signature(tmp_path: Path) -> None: | |
| application = make_app(tmp_path, asset_signing_secret="test-secret") | |
| outputs = application.state.gateway.outputs | |
| target = outputs.allocate("images") | |
| target.write_bytes(b"asset-bytes") | |
| url = outputs.signed_url(target).replace("sig=", "sig=x") | |
| with TestClient(application) as client: | |
| response = client.get(url) | |
| assert response.status_code == 403 | |
| assert response.json()["code"] == "asset_forbidden" | |
| def test_signed_asset_rejects_expired_url(tmp_path: Path) -> None: | |
| application = make_app(tmp_path, asset_signing_secret="test-secret") | |
| outputs = application.state.gateway.outputs | |
| target = outputs.allocate("images") | |
| target.write_bytes(b"asset-bytes") | |
| url = outputs.signed_url(target) | |
| parsed = urlsplit(url) | |
| query = parse_qs(parsed.query) | |
| expired = f"{parsed.path}?expires=1&sig={query['sig'][0]}" | |
| with TestClient(application) as client: | |
| response = client.get(expired) | |
| assert response.status_code == 403 | |
| assert response.json()["code"] == "asset_forbidden" | |
| def test_signed_asset_rejects_path_tampering(tmp_path: Path) -> None: | |
| application = make_app(tmp_path, asset_signing_secret="test-secret") | |
| outputs = application.state.gateway.outputs | |
| target = outputs.allocate("images") | |
| target.write_bytes(b"asset-bytes") | |
| url = outputs.signed_url(target) | |
| parsed = urlsplit(url) | |
| query = parse_qs(parsed.query) | |
| tampered = f"/asset/images/other.png?expires={query['expires'][0]}&sig={query['sig'][0]}" | |
| with TestClient(application) as client: | |
| response = client.get(tampered) | |
| assert response.status_code == 403 | |
| assert response.json()["code"] == "asset_forbidden" | |
| def test_signed_asset_rejects_invalid_category(tmp_path: Path) -> None: | |
| application = make_app(tmp_path, asset_signing_secret="test-secret") | |
| with TestClient(application) as client: | |
| response = client.get( | |
| "/asset/secrets/file.png?expires=9999999999&sig=invalid" | |
| ) | |
| assert response.status_code == 404 | |
| assert response.json()["code"] == "asset_not_found" | |