File size: 3,352 Bytes
6cde1b1 7f0ad46 6cde1b1 7f0ad46 6cde1b1 7f0ad46 6cde1b1 7f0ad46 6cde1b1 7f0ad46 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 | 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,
FailingProcessRemoteImage,
InvalidProcessRemoteImage,
make_container,
)
TEST_IMAGE_URL = "https://httpbin.org/image/jpeg"
class ApiProcessUrlIntegrationTests(unittest.TestCase):
def test_process_url_returns_job_id_on_success(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
root = Path(temp_dir)
process_remote_image = FakeProcessRemoteImage()
app = create_app()
app.dependency_overrides[get_container] = lambda: make_container(
root,
process_remote_image=process_remote_image,
process_uploaded_image=FakeProcessUploadedImage(),
)
client = TestClient(app)
try:
response = client.post("/process/url", json={"sourceUrl": TEST_IMAGE_URL})
self.assertEqual(status.HTTP_200_OK, response.status_code)
self.assertEqual("test-remote-job", response.json()["jobId"])
self.assertEqual(1, len(process_remote_image.calls))
self.assertEqual(TEST_IMAGE_URL, process_remote_image.calls[0].source_url)
finally:
client.close()
app.dependency_overrides.clear()
def test_process_url_returns_500_for_runtime_errors(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=FailingProcessRemoteImage(),
process_uploaded_image=FakeProcessUploadedImage(),
)
client = TestClient(app)
try:
response = client.post("/process/url", json={"sourceUrl": TEST_IMAGE_URL})
self.assertEqual(status.HTTP_500_INTERNAL_SERVER_ERROR, response.status_code)
self.assertEqual("Internal processing error.", response.json()["detail"])
finally:
client.close()
app.dependency_overrides.clear()
def test_process_url_returns_400_for_value_errors(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=InvalidProcessRemoteImage(),
process_uploaded_image=FakeProcessUploadedImage(),
)
client = TestClient(app)
try:
response = client.post("/process/url", json={"sourceUrl": TEST_IMAGE_URL})
self.assertEqual(status.HTTP_400_BAD_REQUEST, response.status_code)
self.assertEqual("invalid source", response.json()["detail"])
finally:
client.close()
app.dependency_overrides.clear()
if __name__ == "__main__":
unittest.main()
|