96 lines
3.2 KiB
Python
96 lines
3.2 KiB
Python
import asyncio
|
|
from io import BytesIO
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
from fastapi import HTTPException
|
|
from PIL import Image
|
|
|
|
from app.api.upload import build_object_key, is_owned_upload_key, normalize_image, validate_image_upload
|
|
from app.config import settings, validate_security_settings
|
|
from app.services.wecom import consume_bind_token, store_bind_token
|
|
|
|
|
|
USER_ID = "a2d4447f-7b67-4304-ad9a-953007e75ef2"
|
|
|
|
|
|
def test_build_object_key_is_scoped_to_the_authenticated_user():
|
|
object_key = build_object_key(USER_ID, "jpg")
|
|
|
|
assert is_owned_upload_key(object_key, USER_ID)
|
|
assert not is_owned_upload_key(object_key, "cf610b28-51a7-423d-a55a-730e193c9971")
|
|
|
|
|
|
def test_upload_rejects_non_image_content_type():
|
|
try:
|
|
validate_image_upload("payload.exe", "application/octet-stream")
|
|
except HTTPException as exc:
|
|
assert exc.status_code == 400
|
|
else:
|
|
raise AssertionError("non-image uploads must be rejected")
|
|
|
|
|
|
def test_upload_rejects_an_invalid_filename_extension():
|
|
try:
|
|
validate_image_upload("payload.txt", "image/jpeg")
|
|
except HTTPException as exc:
|
|
assert exc.status_code == 400
|
|
else:
|
|
raise AssertionError("invalid extensions must be rejected")
|
|
|
|
|
|
def test_image_content_is_normalized_to_jpeg():
|
|
source = BytesIO()
|
|
Image.new("RGBA", (8, 8), color=(255, 0, 0, 128)).save(source, format="PNG")
|
|
|
|
normalized = normalize_image(source.getvalue())
|
|
|
|
with Image.open(BytesIO(normalized)) as image:
|
|
assert image.format == "JPEG"
|
|
assert image.mode == "RGB"
|
|
|
|
|
|
def test_image_content_rejects_non_images():
|
|
try:
|
|
normalize_image(b"not an image")
|
|
except HTTPException as exc:
|
|
assert exc.status_code == 400
|
|
else:
|
|
raise AssertionError("non-image bytes must be rejected")
|
|
|
|
|
|
def test_wecom_binding_tokens_are_persisted_and_consumed_atomically():
|
|
session = MagicMock()
|
|
session.execute = AsyncMock()
|
|
session.commit = AsyncMock()
|
|
|
|
token = asyncio.run(store_bind_token(session, "wecom-user"))
|
|
|
|
assert len(token) == 32
|
|
assert session.add.call_args.args[0].wecom_userid == "wecom-user"
|
|
session.commit.assert_awaited_once()
|
|
|
|
result = MagicMock()
|
|
result.scalar_one_or_none.return_value = "wecom-user"
|
|
session.execute = AsyncMock(return_value=result)
|
|
session.commit = AsyncMock()
|
|
|
|
assert asyncio.run(consume_bind_token(session, token)) == "wecom-user"
|
|
session.commit.assert_awaited_once()
|
|
|
|
|
|
def test_production_rejects_placeholder_database_configuration(monkeypatch):
|
|
monkeypatch.setattr(settings, "ENVIRONMENT", "production")
|
|
monkeypatch.setattr(settings, "SECRET_KEY", "a" * 32)
|
|
monkeypatch.setattr(settings, "DATABASE_URL", "postgresql+asyncpg://qiji_app:invalid@localhost:5432/qiji")
|
|
monkeypatch.setattr(settings, "MINIO_ENDPOINT", "minio.example.test:9000")
|
|
monkeypatch.setattr(settings, "MINIO_ACCESS_KEY", "qiji-app")
|
|
monkeypatch.setattr(settings, "MINIO_SECRET_KEY", "a-secure-minio-secret")
|
|
monkeypatch.setattr(settings, "MINIO_SECURE", True)
|
|
|
|
try:
|
|
validate_security_settings()
|
|
except RuntimeError as exc:
|
|
assert "DATABASE_URL" in str(exc)
|
|
else:
|
|
raise AssertionError("production configuration must reject placeholders")
|