build(security): add migration and CI security gates
This commit is contained in:
@@ -0,0 +1,59 @@
|
||||
from io import BytesIO
|
||||
from unittest.mock import patch
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
from PIL import Image
|
||||
|
||||
from app.api import system_config, upload
|
||||
|
||||
|
||||
def _png_bytes() -> bytes:
|
||||
buffer = BytesIO()
|
||||
Image.new("RGB", (8, 8), color=(10, 20, 30)).save(buffer, format="PNG")
|
||||
return buffer.getvalue()
|
||||
|
||||
|
||||
def _upload_client(authenticated: bool) -> TestClient:
|
||||
app = FastAPI()
|
||||
app.include_router(upload.router)
|
||||
if authenticated:
|
||||
app.dependency_overrides[upload.require_any_role] = lambda: {
|
||||
"user_id": "a2d4447f-7b67-4304-ad9a-953007e75ef2",
|
||||
"role": "manager",
|
||||
}
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
def test_upload_endpoint_requires_bearer_authentication():
|
||||
response = _upload_client(authenticated=False).post(
|
||||
"/upload/image",
|
||||
files={"file": ("photo.png", _png_bytes(), "image/png")},
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
def test_upload_endpoint_validates_and_normalizes_before_storage():
|
||||
with patch("app.api.upload.put_validated_image") as put_image:
|
||||
response = _upload_client(authenticated=True).post(
|
||||
"/upload/image",
|
||||
files={"file": ("photo.png", _png_bytes(), "image/png")},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
object_key = response.json()["object_key"]
|
||||
assert object_key.startswith("uploads/a2d4447f-7b67-4304-ad9a-953007e75ef2/")
|
||||
assert object_key.endswith(".jpg")
|
||||
stored_bytes = put_image.call_args.args[1]
|
||||
with Image.open(BytesIO(stored_bytes)) as image:
|
||||
assert image.format == "JPEG"
|
||||
|
||||
|
||||
def test_system_configuration_cannot_be_read_without_director_authentication():
|
||||
app = FastAPI()
|
||||
app.include_router(system_config.router)
|
||||
|
||||
response = TestClient(app).get("/system-config")
|
||||
|
||||
assert response.status_code == 401
|
||||
Reference in New Issue
Block a user