cc5d67abd6
- Remove Tag/TagHistory model tests from unit tests - Update CacheSettings tests to remove allow_public_internet field - Replace tag= with version= in upload_test_file calls - Update test assertions to use versions instead of tags - Remove tests for tag: prefix downloads (now uses version:) - Update dependency tests for version-only schema
323 lines
12 KiB
Python
323 lines
12 KiB
Python
"""
|
|
Integration tests for error handling in upload and download operations.
|
|
|
|
Tests cover:
|
|
- Timeout handling
|
|
- Invalid request handling
|
|
- Resource cleanup on failures
|
|
- Graceful error responses
|
|
"""
|
|
|
|
import pytest
|
|
import io
|
|
import time
|
|
from tests.factories import (
|
|
compute_sha256,
|
|
upload_test_file,
|
|
generate_content_with_hash,
|
|
)
|
|
|
|
|
|
class TestUploadErrorHandling:
|
|
"""Tests for upload error handling."""
|
|
|
|
@pytest.mark.integration
|
|
def test_upload_to_nonexistent_project_returns_404(
|
|
self, integration_client, unique_test_id
|
|
):
|
|
"""Test upload to nonexistent project returns 404."""
|
|
content = b"test content for nonexistent project"
|
|
|
|
files = {"file": ("test.bin", io.BytesIO(content), "application/octet-stream")}
|
|
response = integration_client.post(
|
|
f"/api/v1/project/nonexistent-project-{unique_test_id}/nonexistent-pkg/upload",
|
|
files=files,
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.integration
|
|
def test_upload_to_nonexistent_package_returns_404(
|
|
self, integration_client, test_project, unique_test_id
|
|
):
|
|
"""Test upload to nonexistent package returns 404."""
|
|
content = b"test content for nonexistent package"
|
|
|
|
files = {"file": ("test.bin", io.BytesIO(content), "application/octet-stream")}
|
|
response = integration_client.post(
|
|
f"/api/v1/project/{test_project}/nonexistent-package-{unique_test_id}/upload",
|
|
files=files,
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.integration
|
|
def test_upload_empty_file_rejected(self, integration_client, test_package):
|
|
"""Test empty file upload is rejected."""
|
|
project, package = test_package
|
|
|
|
files = {"file": ("empty.bin", io.BytesIO(b""), "application/octet-stream")}
|
|
response = integration_client.post(
|
|
f"/api/v1/project/{project}/{package}/upload",
|
|
files=files,
|
|
)
|
|
assert response.status_code in [400, 422]
|
|
|
|
@pytest.mark.integration
|
|
def test_upload_missing_file_returns_422(self, integration_client, test_package):
|
|
"""Test upload without file field returns 422."""
|
|
project, package = test_package
|
|
|
|
response = integration_client.post(
|
|
f"/api/v1/project/{project}/{package}/upload",
|
|
data={"version": "no-file-provided"},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
@pytest.mark.integration
|
|
def test_upload_invalid_checksum_format_returns_400(
|
|
self, integration_client, test_package
|
|
):
|
|
"""Test upload with invalid checksum format returns 400."""
|
|
project, package = test_package
|
|
content = b"checksum format test"
|
|
|
|
files = {"file": ("test.bin", io.BytesIO(content), "application/octet-stream")}
|
|
response = integration_client.post(
|
|
f"/api/v1/project/{project}/{package}/upload",
|
|
files=files,
|
|
headers={"X-Checksum-SHA256": "invalid-hash-format"},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
@pytest.mark.integration
|
|
def test_upload_checksum_mismatch_returns_422(
|
|
self, integration_client, test_package
|
|
):
|
|
"""Test upload with mismatched checksum returns 422."""
|
|
project, package = test_package
|
|
content = b"checksum mismatch test"
|
|
wrong_hash = "0" * 64 # Valid format but wrong hash
|
|
|
|
files = {"file": ("test.bin", io.BytesIO(content), "application/octet-stream")}
|
|
response = integration_client.post(
|
|
f"/api/v1/project/{project}/{package}/upload",
|
|
files=files,
|
|
headers={"X-Checksum-SHA256": wrong_hash},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
@pytest.mark.integration
|
|
def test_upload_with_correct_checksum_succeeds(
|
|
self, integration_client, test_package
|
|
):
|
|
"""Test upload with correct checksum succeeds."""
|
|
project, package = test_package
|
|
content = b"correct checksum test"
|
|
correct_hash = compute_sha256(content)
|
|
|
|
files = {"file": ("test.bin", io.BytesIO(content), "application/octet-stream")}
|
|
response = integration_client.post(
|
|
f"/api/v1/project/{project}/{package}/upload",
|
|
files=files,
|
|
headers={"X-Checksum-SHA256": correct_hash},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["artifact_id"] == correct_hash
|
|
|
|
|
|
class TestDownloadErrorHandling:
|
|
"""Tests for download error handling."""
|
|
|
|
@pytest.mark.integration
|
|
def test_download_nonexistent_tag_returns_404(
|
|
self, integration_client, test_package
|
|
):
|
|
"""Test download of nonexistent tag returns 404."""
|
|
project, package = test_package
|
|
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project}/{package}/+/nonexistent-tag-xyz"
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.integration
|
|
def test_download_nonexistent_artifact_returns_404(
|
|
self, integration_client, test_package
|
|
):
|
|
"""Test download of nonexistent artifact ID returns 404."""
|
|
project, package = test_package
|
|
fake_hash = "a" * 64
|
|
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project}/{package}/+/artifact:{fake_hash}"
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.integration
|
|
def test_download_invalid_artifact_id_format(
|
|
self, integration_client, test_package
|
|
):
|
|
"""Test download with invalid artifact ID format."""
|
|
project, package = test_package
|
|
|
|
# Too short
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project}/{package}/+/artifact:abc123"
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.integration
|
|
def test_download_from_nonexistent_project_returns_404(
|
|
self, integration_client, unique_test_id
|
|
):
|
|
"""Test download from nonexistent project returns 404."""
|
|
response = integration_client.get(
|
|
f"/api/v1/project/nonexistent-{unique_test_id}/pkg/+/tag"
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.integration
|
|
def test_download_from_nonexistent_package_returns_404(
|
|
self, integration_client, test_project, unique_test_id
|
|
):
|
|
"""Test download from nonexistent package returns 404."""
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{test_project}/nonexistent-{unique_test_id}/+/tag"
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
class TestTimeoutBehavior:
|
|
"""Tests for timeout behavior (integration level)."""
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.slow
|
|
def test_large_upload_completes_within_reasonable_time(
|
|
self, integration_client, test_package, sized_content
|
|
):
|
|
"""Test that a 10MB upload completes within reasonable time."""
|
|
project, package = test_package
|
|
content, expected_hash = sized_content(10 * 1024 * 1024, seed=999) # 10MB
|
|
|
|
start_time = time.time()
|
|
result = upload_test_file(
|
|
integration_client, project, package, content, version="timeout-test"
|
|
)
|
|
elapsed = time.time() - start_time
|
|
|
|
assert result["artifact_id"] == expected_hash
|
|
# Should complete within 60 seconds for 10MB on local docker
|
|
assert elapsed < 60, f"Upload took too long: {elapsed:.2f}s"
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.slow
|
|
def test_large_download_completes_within_reasonable_time(
|
|
self, integration_client, test_package, sized_content
|
|
):
|
|
"""Test that a 10MB download completes within reasonable time."""
|
|
project, package = test_package
|
|
content, expected_hash = sized_content(10 * 1024 * 1024, seed=998) # 10MB
|
|
|
|
# First upload
|
|
upload_test_file(
|
|
integration_client, project, package, content, version="download-timeout-test"
|
|
)
|
|
|
|
# Then download and time it
|
|
start_time = time.time()
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project}/{package}/+/download-timeout-test",
|
|
params={"mode": "proxy"},
|
|
)
|
|
elapsed = time.time() - start_time
|
|
|
|
assert response.status_code == 200
|
|
assert len(response.content) == len(content)
|
|
# Should complete within 60 seconds for 10MB on local docker
|
|
assert elapsed < 60, f"Download took too long: {elapsed:.2f}s"
|
|
|
|
|
|
class TestResourceCleanup:
|
|
"""Tests for proper resource cleanup on failures.
|
|
|
|
Note: More comprehensive cleanup tests are in test_upload_download_api.py
|
|
(TestUploadFailureCleanup class) including S3 object cleanup verification.
|
|
"""
|
|
|
|
@pytest.mark.integration
|
|
def test_checksum_mismatch_no_orphaned_artifact(
|
|
self, integration_client, test_package, unique_test_id
|
|
):
|
|
"""Test checksum mismatch doesn't leave orphaned artifact."""
|
|
project, package = test_package
|
|
# Use unique content to ensure artifact doesn't exist from prior tests
|
|
content = f"checksum mismatch orphan test {unique_test_id}".encode()
|
|
wrong_hash = "0" * 64
|
|
actual_hash = compute_sha256(content)
|
|
|
|
# Verify artifact doesn't exist before test
|
|
pre_check = integration_client.get(f"/api/v1/artifact/{actual_hash}")
|
|
assert pre_check.status_code == 404, "Artifact should not exist before test"
|
|
|
|
files = {"file": ("test.bin", io.BytesIO(content), "application/octet-stream")}
|
|
response = integration_client.post(
|
|
f"/api/v1/project/{project}/{package}/upload",
|
|
files=files,
|
|
headers={"X-Checksum-SHA256": wrong_hash},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
# Verify no artifact was created with either hash
|
|
response1 = integration_client.get(f"/api/v1/artifact/{wrong_hash}")
|
|
response2 = integration_client.get(f"/api/v1/artifact/{actual_hash}")
|
|
assert response1.status_code == 404
|
|
assert response2.status_code == 404
|
|
|
|
|
|
class TestGracefulErrorResponses:
|
|
"""Tests for graceful and informative error responses."""
|
|
|
|
@pytest.mark.integration
|
|
def test_404_response_has_detail_message(
|
|
self, integration_client, test_package
|
|
):
|
|
"""Test 404 responses include a detail message."""
|
|
project, package = test_package
|
|
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project}/{package}/+/nonexistent-tag"
|
|
)
|
|
assert response.status_code == 404
|
|
data = response.json()
|
|
assert "detail" in data
|
|
assert len(data["detail"]) > 0
|
|
|
|
@pytest.mark.integration
|
|
def test_422_response_has_detail_message(self, integration_client, test_package):
|
|
"""Test 422 responses include a detail message."""
|
|
project, package = test_package
|
|
|
|
# Upload with mismatched checksum
|
|
content = b"detail message test"
|
|
wrong_hash = "0" * 64
|
|
|
|
files = {"file": ("test.bin", io.BytesIO(content), "application/octet-stream")}
|
|
response = integration_client.post(
|
|
f"/api/v1/project/{project}/{package}/upload",
|
|
files=files,
|
|
headers={"X-Checksum-SHA256": wrong_hash},
|
|
)
|
|
assert response.status_code == 422
|
|
data = response.json()
|
|
assert "detail" in data
|
|
|
|
@pytest.mark.integration
|
|
def test_error_response_is_json(self, integration_client, unique_test_id):
|
|
"""Test error responses are valid JSON."""
|
|
response = integration_client.get(
|
|
f"/api/v1/project/nonexistent-{unique_test_id}/pkg/+/tag"
|
|
)
|
|
assert response.status_code == 404
|
|
# Should not raise exception - valid JSON
|
|
data = response.json()
|
|
assert isinstance(data, dict)
|