Frank Schwenk
c03f17488c
CI / lint-and-test (push) Failing after 1m18s
Replace Syncthing as the product upload path with a share-target app and POST /api/jobs. Web UI live-polls job progress. Co-authored-by: Cursor <cursoragent@cursor.com>
136 lines
4.3 KiB
Python
136 lines
4.3 KiB
Python
"""Tests for JSON API upload and job payloads."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from app import config, pipeline
|
|
from app.main import app
|
|
from fastapi.testclient import TestClient
|
|
from PIL import Image
|
|
|
|
|
|
@pytest.fixture
|
|
def client(data_dir: Path) -> TestClient:
|
|
return TestClient(app)
|
|
|
|
|
|
@pytest.fixture
|
|
def upload_token(monkeypatch: pytest.MonkeyPatch) -> str:
|
|
token = "test-upload-token"
|
|
monkeypatch.setattr(config, "UPLOAD_TOKEN", token)
|
|
return token
|
|
|
|
|
|
def _jpeg_bytes(size: tuple[int, int] = (80, 60)) -> bytes:
|
|
from io import BytesIO
|
|
|
|
buf = BytesIO()
|
|
Image.new("RGB", size, color=(12, 34, 56)).save(buf, format="JPEG", quality=85)
|
|
return buf.getvalue()
|
|
|
|
|
|
def test_health_reports_upload_disabled(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(config, "UPLOAD_TOKEN", "")
|
|
response = client.get("/api/health")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"ok": True, "upload": False}
|
|
|
|
|
|
def test_upload_disabled_without_token(client: TestClient, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(config, "UPLOAD_TOKEN", "")
|
|
response = client.post(
|
|
"/api/jobs",
|
|
files={"file": ("party.jpg", _jpeg_bytes(), "image/jpeg")},
|
|
)
|
|
assert response.status_code == 503
|
|
|
|
|
|
def test_upload_rejects_bad_token(client: TestClient, upload_token: str) -> None:
|
|
response = client.post(
|
|
"/api/jobs",
|
|
files={"file": ("party.jpg", _jpeg_bytes(), "image/jpeg")},
|
|
headers={"Authorization": "Bearer nope"},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_upload_accepts_bearer_and_queues(client: TestClient, upload_token: str, data_dir: Path) -> None:
|
|
response = client.post(
|
|
"/api/jobs",
|
|
files={"file": ("party.jpg", _jpeg_bytes(), "image/jpeg")},
|
|
headers={"Authorization": f"Bearer {upload_token}"},
|
|
)
|
|
assert response.status_code == 202
|
|
body = response.json()
|
|
assert body["status"] == "queued"
|
|
job_id = body["job_id"]
|
|
assert job_id.endswith("_party")
|
|
inbox = list(config.INBOX_DIR.glob("*_party.jpg"))
|
|
assert len(inbox) == 1
|
|
assert not inbox[0].name.startswith(".")
|
|
meta = pipeline.read_meta(job_id)
|
|
assert meta is not None
|
|
assert meta["status"] == "queued"
|
|
|
|
|
|
def test_upload_accepts_x_upload_token_header(client: TestClient, upload_token: str) -> None:
|
|
response = client.post(
|
|
"/api/jobs",
|
|
files={"file": ("cam.png", _jpeg_bytes(), "image/png")},
|
|
headers={"X-Upload-Token": upload_token},
|
|
)
|
|
assert response.status_code == 202
|
|
assert response.json()["status"] == "queued"
|
|
|
|
|
|
def test_upload_rejects_empty_file(client: TestClient, upload_token: str) -> None:
|
|
response = client.post(
|
|
"/api/jobs",
|
|
files={"file": ("empty.jpg", b"", "image/jpeg")},
|
|
headers={"Authorization": f"Bearer {upload_token}"},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_api_job_detail_and_list(client: TestClient, data_dir: Path) -> None:
|
|
stem = "photo01"
|
|
paths = pipeline.job_paths(stem)
|
|
paths.variants.mkdir(parents=True, exist_ok=True)
|
|
png = paths.variants / f"{stem}_v1.png"
|
|
Image.new("RGB", (32, 32), color=(1, 2, 3)).save(png, format="PNG")
|
|
pipeline.write_status(
|
|
paths,
|
|
"processing",
|
|
stage="variant",
|
|
progress={"current": 2, "total": 3},
|
|
created_at="2026-01-01T00:00:00",
|
|
)
|
|
pipeline.write_manifest(
|
|
paths,
|
|
{
|
|
"variants": [{"id": "v1", "file": png.name}],
|
|
"created_at": "2026-01-01T00:00:00",
|
|
},
|
|
)
|
|
|
|
listed = client.get("/api/jobs")
|
|
assert listed.status_code == 200
|
|
payload = listed.json()
|
|
assert payload["active"] is True
|
|
assert payload["jobs"][0]["job_id"] == stem
|
|
assert payload["jobs"][0]["status_label"] == "Läuft"
|
|
assert payload["jobs"][0]["stage_label"] == "Variante 2/3"
|
|
|
|
detail = client.get(f"/api/jobs/{stem}")
|
|
assert detail.status_code == 200
|
|
body = detail.json()
|
|
assert body["has_rembg"] is False
|
|
assert body["variants"][0]["url"].endswith(f"/jobs/{stem}/files/{png.name}")
|
|
assert body["progress"] == {"current": 2, "total": 3}
|
|
|
|
|
|
def test_api_job_missing(client: TestClient) -> None:
|
|
assert client.get("/api/jobs/does-not-exist").status_code == 404
|