File size: 12,987 Bytes
44a2550 e7bf1e6 44a2550 e7bf1e6 44a2550 e7bf1e6 44a2550 e7bf1e6 44a2550 e7bf1e6 44a2550 |
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 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 |
"""Integration tests for FastAPI endpoints."""
import pytest
from unittest.mock import patch, MagicMock
import json
class TestRootEndpoint:
"""Test root endpoint."""
def test_root(self, test_client):
"""Test root endpoint returns API info."""
response = test_client.get("/")
assert response.status_code == 200
data = response.json()
assert data["name"] == "Rescored API"
assert data["version"] == "1.0.0"
assert data["docs"] == "/docs"
class TestHealthCheck:
"""Test health check endpoint."""
def test_health_check_healthy(self, test_client, mock_redis):
"""Test health check when all services are healthy."""
mock_redis.ping.return_value = True
response = test_client.get("/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "healthy"
assert data["redis"] == "healthy"
def test_health_check_redis_down(self, test_client, mock_redis):
"""Test health check when Redis is down."""
mock_redis.ping.side_effect = Exception("Connection failed")
response = test_client.get("/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "degraded"
assert data["redis"] == "unhealthy"
class TestTranscribeEndpoint:
"""Test transcription submission endpoint."""
@patch('main.process_transcription_task')
@patch('app_utils.check_video_availability')
@patch('main.validate_youtube_url')
def test_submit_valid_transcription(
self,
mock_validate,
mock_check_availability,
mock_task,
test_client,
mock_redis
):
"""Test submitting valid transcription request."""
mock_validate.return_value = (True, "dQw4w9WgXcQ")
mock_check_availability.return_value = {
'available': True,
'info': {'duration': 180}
}
mock_task.delay.return_value = MagicMock(id="task-id")
response = test_client.post(
"/api/v1/transcribe",
json={"youtube_url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ"}
)
assert response.status_code == 201
data = response.json()
assert "job_id" in data
assert data["status"] == "queued"
assert "websocket_url" in data
assert data["estimated_duration_seconds"] == 120
# Verify Redis was called to store job
assert mock_redis.hset.called
# Verify Celery task was queued
assert mock_task.delay.called
@patch('main.validate_youtube_url')
def test_submit_invalid_url(self, mock_validate, test_client):
"""Test submitting invalid YouTube URL."""
mock_validate.return_value = (False, "Invalid YouTube URL format")
response = test_client.post(
"/api/v1/transcribe",
json={"youtube_url": "https://invalid.com/video"}
)
assert response.status_code == 400
assert "Invalid YouTube URL format" in response.json()["detail"]
@patch('main.validate_youtube_url')
@patch('main.check_video_availability')
def test_submit_unavailable_video(
self,
mock_check_availability,
mock_validate,
test_client
):
"""Test submitting unavailable video."""
mock_validate.return_value = (True, "dQw4w9WgXcQ")
mock_check_availability.return_value = {
'available': False,
'reason': 'Video too long (max 15 minutes)'
}
response = test_client.post(
"/api/v1/transcribe",
json={"youtube_url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ"}
)
assert response.status_code == 422
assert "too long" in response.json()["detail"]
@patch('main.validate_youtube_url')
@patch('main.check_video_availability')
def test_submit_with_options(
self,
mock_check_availability,
mock_validate,
test_client,
mock_redis
):
"""Test submitting transcription with custom options."""
mock_validate.return_value = (True, "dQw4w9WgXcQ")
mock_check_availability.return_value = {'available': True, 'info': {}}
with patch('main.process_transcription_task') as mock_task:
response = test_client.post(
"/api/v1/transcribe",
json={
"youtube_url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
"options": {"instruments": ["piano", "guitar"]}
}
)
assert response.status_code == 201
class TestRateLimiting:
"""Test rate limiting middleware."""
@patch('main.validate_youtube_url')
@patch('main.check_video_availability')
@patch('main.process_transcription_task')
def test_rate_limit_enforced(
self,
mock_task,
mock_check_availability,
mock_validate,
test_client,
mock_redis
):
"""Test that rate limit is enforced after 10 requests."""
mock_validate.return_value = (True, "dQw4w9WgXcQ")
mock_check_availability.return_value = {'available': True, 'info': {}}
mock_task.delay.return_value = MagicMock(id="task-id")
# Mock Redis counter for rate limiting
mock_redis.get.return_value = "10" # Already at limit
response = test_client.post(
"/api/v1/transcribe",
json={"youtube_url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ"}
)
assert response.status_code == 429
assert "Rate limit exceeded" in response.json()["detail"]
@patch('main.validate_youtube_url')
@patch('main.check_video_availability')
@patch('main.process_transcription_task')
def test_rate_limit_under_limit(
self,
mock_task,
mock_check_availability,
mock_validate,
test_client,
mock_redis
):
"""Test that requests under limit succeed."""
mock_validate.return_value = (True, "dQw4w9WgXcQ")
mock_check_availability.return_value = {'available': True, 'info': {}}
mock_task.delay.return_value = MagicMock(id="task-id")
# Mock Redis counter for rate limiting (under limit)
mock_redis.get.return_value = "5" # 5 out of 10
response = test_client.post(
"/api/v1/transcribe",
json={"youtube_url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ"}
)
assert response.status_code == 201 # Request succeeds
assert mock_redis.pipeline.called # Counter incremented
class TestJobStatusEndpoint:
"""Test job status endpoint."""
def test_get_existing_job_status(self, test_client, mock_redis, sample_job_data):
"""Test getting status of existing job."""
mock_redis.hgetall.return_value = sample_job_data
response = test_client.get(f"/api/v1/jobs/{sample_job_data['job_id']}")
assert response.status_code == 200
data = response.json()
assert data["job_id"] == sample_job_data["job_id"]
assert data["status"] == "queued"
assert data["progress"] == 0
assert data["current_stage"] == "queued"
def test_get_nonexistent_job(self, test_client, mock_redis):
"""Test getting status of nonexistent job."""
mock_redis.hgetall.return_value = {}
response = test_client.get("/api/v1/jobs/nonexistent-id")
assert response.status_code == 404
assert "not found" in response.json()["detail"]
def test_get_completed_job_status(self, test_client, mock_redis, sample_job_data):
"""Test getting status of completed job."""
completed_job = {**sample_job_data, "status": "completed", "progress": 100}
mock_redis.hgetall.return_value = completed_job
response = test_client.get(f"/api/v1/jobs/{sample_job_data['job_id']}")
assert response.status_code == 200
data = response.json()
assert data["status"] == "completed"
assert data["progress"] == 100
assert data["result_url"] is not None
def test_get_failed_job_status(self, test_client, mock_redis, sample_job_data):
"""Test getting status of failed job."""
error_data = {"message": "Transcription failed", "stage": "audio_download"}
failed_job = {
**sample_job_data,
"status": "failed",
"error": json.dumps(error_data)
}
mock_redis.hgetall.return_value = failed_job
response = test_client.get(f"/api/v1/jobs/{sample_job_data['job_id']}")
assert response.status_code == 200
data = response.json()
assert data["status"] == "failed"
assert data["error"] is not None
assert data["error"]["message"] == "Transcription failed"
class TestScoreDownloadEndpoint:
"""Test score download endpoint."""
def test_download_completed_score(
self,
test_client,
mock_redis,
sample_job_data,
temp_storage_dir,
sample_musicxml_content
):
"""Test downloading a completed score."""
# Create a real MusicXML file
score_path = temp_storage_dir / "score.musicxml"
score_path.write_text(sample_musicxml_content)
completed_job = {
**sample_job_data,
"status": "completed",
"output_path": str(score_path)
}
mock_redis.hgetall.return_value = completed_job
response = test_client.get(f"/api/v1/scores/{sample_job_data['job_id']}")
assert response.status_code == 200
assert response.headers["content-type"] == "application/vnd.recordare.musicxml+xml"
assert "score-partwise" in response.text
def test_download_nonexistent_job(self, test_client, mock_redis):
"""Test downloading score for nonexistent job."""
mock_redis.hgetall.return_value = {}
response = test_client.get("/api/v1/scores/nonexistent-id")
assert response.status_code == 404
def test_download_incomplete_job(self, test_client, mock_redis, sample_job_data):
"""Test downloading score for incomplete job."""
mock_redis.hgetall.return_value = sample_job_data # Still queued
response = test_client.get(f"/api/v1/scores/{sample_job_data['job_id']}")
assert response.status_code == 404
assert "not available" in response.json()["detail"]
def test_download_missing_file(self, test_client, mock_redis, sample_job_data):
"""Test downloading score when file is missing."""
completed_job = {
**sample_job_data,
"status": "completed",
"output_path": "/nonexistent/path/score.musicxml"
}
mock_redis.hgetall.return_value = completed_job
response = test_client.get(f"/api/v1/scores/{sample_job_data['job_id']}")
assert response.status_code == 404
assert "not found" in response.json()["detail"]
class TestMIDIDownloadEndpoint:
"""Test MIDI download endpoint."""
def test_download_completed_midi(self, test_client, sample_job_id, tmp_path, mock_redis):
"""Test downloading MIDI from completed job."""
# Create a dummy MIDI file
midi_file = tmp_path / "test.mid"
midi_file.write_bytes(b"MIDI_DATA")
# Set job as completed with MIDI path
mock_redis.hgetall.return_value = {
"status": "completed",
"midi_path": str(midi_file)
}
response = test_client.get(f"/api/v1/scores/{sample_job_id}/midi")
assert response.status_code == 200
assert response.headers["content-type"] == "audio/midi"
assert response.content == b"MIDI_DATA"
def test_download_nonexistent_job_midi(self, test_client, mock_redis):
"""Test downloading MIDI from nonexistent job."""
mock_redis.hgetall.return_value = {}
response = test_client.get("/api/v1/scores/nonexistent/midi")
assert response.status_code == 404
assert "not available" in response.json()["detail"]
def test_download_incomplete_job_midi(self, test_client, sample_job_id, mock_redis):
"""Test downloading MIDI from incomplete job."""
mock_redis.hgetall.return_value = {"status": "processing"}
response = test_client.get(f"/api/v1/scores/{sample_job_id}/midi")
assert response.status_code == 404
def test_download_missing_midi_file(self, test_client, sample_job_id, mock_redis):
"""Test downloading when MIDI file doesn't exist."""
mock_redis.hgetall.return_value = {
"status": "completed",
"midi_path": "/nonexistent/path.mid"
}
response = test_client.get(f"/api/v1/scores/{sample_job_id}/midi")
assert response.status_code == 404
assert "file not found" in response.json()["detail"].lower()
|