mirror of
https://github.com/azaion/gps-denied-onboard.git
synced 2026-04-22 07:06:38 +00:00
Denys Zaitsev
125 lines
5.8 KiB
Python
125 lines
5.8 KiB
Python
import pytest
|
|
import io
|
|
from fastapi.testclient import TestClient
|
|
from unittest.mock import Mock
|
|
from fastapi import FastAPI
|
|
|
|
from f01_flight_api import router, get_lifecycle_manager
|
|
from f02_1_flight_lifecycle_manager import FlightLifecycleManager
|
|
|
|
# --- Setup ---
|
|
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
mock_manager = Mock(spec=FlightLifecycleManager)
|
|
app.dependency_overrides[get_lifecycle_manager] = lambda: mock_manager
|
|
client = TestClient(app)
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_mocks():
|
|
mock_manager.reset_mock()
|
|
mock_manager.queue_images.return_value = True
|
|
|
|
def create_dummy_file(filename: str, content: bytes = b"dummy_image_data", content_type: str = "image/jpeg"):
|
|
return ("images", (filename, content, content_type))
|
|
|
|
# --- Unit Tests ---
|
|
|
|
class TestUnitImageUpload:
|
|
"""Unit tests defined in 01.02_feature_image_upload.md"""
|
|
|
|
def test_batch_size_validation(self):
|
|
flight_id = "flight_123"
|
|
|
|
# 1. Too few (9 images) -> 400
|
|
files_too_few = [create_dummy_file(f"AD{i:06d}.jpg") for i in range(1, 10)]
|
|
resp = client.post(f"/api/v1/flights/{flight_id}/images/batch", data={"start_sequence": 1, "end_sequence": 9, "batch_number": 1}, files=files_too_few)
|
|
assert resp.status_code == 400
|
|
assert "between 10 and 50" in resp.json()["detail"]
|
|
|
|
# 2. Too many (51 images) -> 400
|
|
files_too_many = [create_dummy_file(f"AD{i:06d}.jpg") for i in range(1, 52)]
|
|
resp = client.post(f"/api/v1/flights/{flight_id}/images/batch", data={"start_sequence": 1, "end_sequence": 51, "batch_number": 1}, files=files_too_many)
|
|
assert resp.status_code == 400
|
|
|
|
# 3. Exactly 10 (Accepted)
|
|
files_valid = [create_dummy_file(f"AD{i:06d}.jpg") for i in range(1, 11)]
|
|
resp = client.post(f"/api/v1/flights/{flight_id}/images/batch", data={"start_sequence": 1, "end_sequence": 10, "batch_number": 1}, files=files_valid)
|
|
assert resp.status_code == 202
|
|
assert resp.json()["accepted"] is True
|
|
assert len(resp.json()["sequences"]) == 10
|
|
|
|
def test_sequence_validation(self):
|
|
flight_id = "flight_123"
|
|
files = [create_dummy_file(f"AD{i:06d}.jpg") for i in range(100, 110)]
|
|
|
|
# 1. Valid sequence
|
|
resp = client.post(f"/api/v1/flights/{flight_id}/images/batch", data={"start_sequence": 100, "end_sequence": 109, "batch_number": 1}, files=files)
|
|
assert resp.status_code == 202
|
|
|
|
# 2. Invalid sequence (start > end)
|
|
resp = client.post(f"/api/v1/flights/{flight_id}/images/batch", data={"start_sequence": 110, "end_sequence": 100, "batch_number": 1}, files=files)
|
|
assert resp.status_code == 400
|
|
|
|
# 3. Gap in sequence (mismatch between count and sequence span)
|
|
resp = client.post(f"/api/v1/flights/{flight_id}/images/batch", data={"start_sequence": 100, "end_sequence": 115, "batch_number": 1}, files=files)
|
|
assert resp.status_code == 400
|
|
|
|
def test_image_format_validation(self):
|
|
flight_id = "flight_123"
|
|
|
|
# 1. Invalid format (Text file)
|
|
files = [create_dummy_file(f"AD{i:06d}.jpg") for i in range(1, 10)]
|
|
files.append(create_dummy_file("AD000010.txt", content_type="text/plain"))
|
|
|
|
resp = client.post(f"/api/v1/flights/{flight_id}/images/batch", data={"start_sequence": 1, "end_sequence": 10, "batch_number": 1}, files=files)
|
|
assert resp.status_code == 400
|
|
assert "Invalid image format" in resp.json()["detail"]
|
|
|
|
def test_size_limits(self):
|
|
flight_id = "flight_123"
|
|
|
|
# Simulate 500MB+ payload by mocking read
|
|
# FastAPI validates the actual bytes read. We will construct 10 files of 51MB each
|
|
large_content = b"0" * (51 * 1024 * 1024)
|
|
files = [create_dummy_file(f"AD{i:06d}.jpg", content=large_content) for i in range(1, 11)]
|
|
|
|
resp = client.post(f"/api/v1/flights/{flight_id}/images/batch", data={"start_sequence": 1, "end_sequence": 10, "batch_number": 1}, files=files)
|
|
assert resp.status_code == 413
|
|
assert "exceeds 500MB" in resp.json()["detail"]
|
|
|
|
# --- Integration Tests ---
|
|
|
|
class TestIntegrationImageUpload:
|
|
"""Integration tests defined in 01.02_feature_image_upload.md"""
|
|
|
|
def test_sequential_batch_uploads(self):
|
|
flight_id = "flight_123"
|
|
|
|
# Batch 1 (1-15)
|
|
files_1 = [create_dummy_file(f"AD{i:06d}.jpg") for i in range(1, 16)]
|
|
resp1 = client.post(f"/api/v1/flights/{flight_id}/images/batch", data={"start_sequence": 1, "end_sequence": 15, "batch_number": 1}, files=files_1)
|
|
|
|
assert resp1.status_code == 202
|
|
assert resp1.json()["next_expected"] == 16
|
|
|
|
# Batch 2 (16-30)
|
|
files_2 = [create_dummy_file(f"AD{i:06d}.jpg") for i in range(16, 31)]
|
|
resp2 = client.post(f"/api/v1/flights/{flight_id}/images/batch", data={"start_sequence": 16, "end_sequence": 30, "batch_number": 2}, files=files_2)
|
|
|
|
assert resp2.status_code == 202
|
|
assert resp2.json()["next_expected"] == 31
|
|
assert mock_manager.queue_images.call_count == 2
|
|
|
|
def test_concurrent_uploads_to_same_flight(self):
|
|
"""Simulates parallel handling (FastAPI inherently handles concurrent requests)."""
|
|
flight_id = "flight_123"
|
|
files_1 = [create_dummy_file(f"AD{i:06d}.jpg") for i in range(1, 11)]
|
|
files_2 = [create_dummy_file(f"AD{i:06d}.jpg") for i in range(11, 21)]
|
|
|
|
r1 = client.post(f"/api/v1/flights/{flight_id}/images/batch", data={"start_sequence": 1, "end_sequence": 10, "batch_number": 1}, files=files_1)
|
|
r2 = client.post(f"/api/v1/flights/{flight_id}/images/batch", data={"start_sequence": 11, "end_sequence": 20, "batch_number": 2}, files=files_2)
|
|
|
|
assert r1.status_code == 202
|
|
assert r2.status_code == 202
|
|
assert mock_manager.queue_images.call_count == 2 |