File size: 6,924 Bytes
6cde1b1 9be3b9f 6cde1b1 4d0e069 6cde1b1 4d0e069 6cde1b1 4d0e069 6cde1b1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from fastapi import status
from fastapi.testclient import TestClient
from app.api.dependencies import get_container
from app.main import create_app
from tests.integration.api.support import FakeProcessRemoteImage, FakeProcessUploadedImage, make_container
class ApiProcessUploadIntegrationTests(unittest.TestCase):
def test_process_upload_accepts_real_demo_image_file(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
process_uploaded_image = FakeProcessUploadedImage()
app = create_app()
app.dependency_overrides[get_container] = lambda: make_container(
root,
process_remote_image=FakeProcessRemoteImage(),
process_uploaded_image=process_uploaded_image,
)
client = TestClient(app)
demo_path = Path(__file__).resolve().parents[2] / "demo.jpg"
expected_bytes = demo_path.read_bytes()
try:
with demo_path.open("rb") as image_file:
response = client.post(
"/process/upload",
files={"file": ("demo.jpg", image_file, "image/jpeg")},
data={"options": '{"formats":["jpg"]}'},
)
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertEqual("test-upload-job", response.json()["jobId"])
self.assertEqual(1, len(process_uploaded_image.calls))
self.assertEqual("demo.jpg", process_uploaded_image.calls[0][0])
self.assertEqual(expected_bytes, process_uploaded_image.calls[0][1])
self.assertEqual(["jpg"], process_uploaded_image.calls[0][2].formats)
finally:
client.close()
app.dependency_overrides.clear()
def test_process_upload_returns_job_id_on_success(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
process_uploaded_image = FakeProcessUploadedImage()
app = create_app()
app.dependency_overrides[get_container] = lambda: make_container(
root,
process_remote_image=FakeProcessRemoteImage(),
process_uploaded_image=process_uploaded_image,
)
client = TestClient(app)
try:
response = client.post(
"/process/upload",
files={"file": ("test.jpg", b"binary-image", "image/jpeg")},
data={"options": '{"formats":["jpg"],"upload":{"uploadToR2":false}}'},
)
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertEqual("test-upload-job", response.json()["jobId"])
self.assertEqual(1, len(process_uploaded_image.calls))
self.assertEqual("test.jpg", process_uploaded_image.calls[0][0])
self.assertEqual(b"binary-image", process_uploaded_image.calls[0][1])
self.assertEqual(["jpg"], process_uploaded_image.calls[0][2].formats)
finally:
client.close()
app.dependency_overrides.clear()
def test_process_upload_returns_400_for_invalid_options(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
app = create_app()
app.dependency_overrides[get_container] = lambda: make_container(
root,
process_remote_image=FakeProcessRemoteImage(),
process_uploaded_image=FakeProcessUploadedImage(),
)
client = TestClient(app)
try:
response = client.post(
"/process/upload",
files={"file": ("test.jpg", b"binary-image", "image/jpeg")},
data={"options": "{not-json}"},
)
self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
self.assertIn("Invalid options:", response.json()["detail"])
finally:
client.close()
app.dependency_overrides.clear()
def test_process_upload_returns_400_when_source_fields_are_provided(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
app = create_app()
app.dependency_overrides[get_container] = lambda: make_container(
root,
process_remote_image=FakeProcessRemoteImage(),
process_uploaded_image=FakeProcessUploadedImage(),
)
client = TestClient(app)
try:
response = client.post(
"/process/upload",
files={"file": ("test.jpg", b"binary-image", "image/jpeg")},
data={"options": '{"sourceUrl":"https://example.com/image.jpg"}'},
)
self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
self.assertIn("must not include source fields", response.json()["detail"])
finally:
client.close()
app.dependency_overrides.clear()
def test_process_upload_accepts_r2_prefix_override_with_nested_path(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
process_uploaded_image = FakeProcessUploadedImage()
app = create_app()
app.dependency_overrides[get_container] = lambda: make_container(
root,
process_remote_image=FakeProcessRemoteImage(),
process_uploaded_image=process_uploaded_image,
)
client = TestClient(app)
try:
response = client.post(
"/process/upload",
files={"file": ("test.jpg", b"binary-image", "image/jpeg")},
data={"options": '{"formats":["jpg"],"upload":{"uploadToR2":true,"r2Prefix":"nested/listing-123/gallery"}}'},
)
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertEqual("test-upload-job", response.json()["jobId"])
self.assertEqual(1, len(process_uploaded_image.calls))
request = process_uploaded_image.calls[0][2]
self.assertIsNotNone(request.upload)
self.assertTrue(request.upload.upload_to_r2)
self.assertEqual("nested/listing-123/gallery", request.upload.r2_prefix)
finally:
client.close()
app.dependency_overrides.clear()
if __name__ == "__main__":
unittest.main()
|