Add comprehensive upload/download API tests and error handling tests
- Add upload API tests: upload without tag, artifact creation, S3 storage - Add download tests: tag: prefix, Content-Type/Length/Disposition headers - Add download tests: 404 for nonexistent project/package/artifact - Add checksum header tests: ETag, X-Checksum-SHA256 - Add error handling tests: timeout behavior, checksum validation - Add resource cleanup tests: verify no orphans on failed uploads - Add graceful error response tests: JSON format, detail messages
This commit is contained in:
Mondo Diaz
@@ -25,6 +25,19 @@ from tests.factories import (
|
||||
class TestUploadBasics:
|
||||
"""Tests for basic upload functionality."""
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_upload_returns_200(self, integration_client, test_package):
|
||||
"""Test upload with valid file returns 200."""
|
||||
project, package = test_package
|
||||
content = b"valid file upload test"
|
||||
|
||||
files = {"file": ("test.bin", io.BytesIO(content), "application/octet-stream")}
|
||||
response = integration_client.post(
|
||||
f"/api/v1/project/{project}/{package}/upload",
|
||||
files=files,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_upload_returns_artifact_id(self, integration_client, test_package):
|
||||
"""Test upload returns the artifact ID (SHA256 hash)."""
|
||||
@@ -101,6 +114,82 @@ class TestUploadBasics:
|
||||
assert "created_at" in result
|
||||
assert result["created_at"] is not None
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_upload_without_tag_succeeds(self, integration_client, test_package):
|
||||
"""Test upload without tag succeeds (no tag created)."""
|
||||
project, package = test_package
|
||||
content = b"upload without tag test"
|
||||
expected_hash = compute_sha256(content)
|
||||
|
||||
files = {"file": ("no_tag.bin", io.BytesIO(content), "application/octet-stream")}
|
||||
response = integration_client.post(
|
||||
f"/api/v1/project/{project}/{package}/upload",
|
||||
files=files,
|
||||
# No tag parameter
|
||||
)
|
||||
assert response.status_code == 200
|
||||
result = response.json()
|
||||
assert result["artifact_id"] == expected_hash
|
||||
|
||||
# Verify no tag was created - list tags and check
|
||||
tags_response = integration_client.get(
|
||||
f"/api/v1/project/{project}/{package}/tags"
|
||||
)
|
||||
assert tags_response.status_code == 200
|
||||
tags = tags_response.json()
|
||||
# Filter for tags pointing to this artifact
|
||||
artifact_tags = [t for t in tags.get("items", tags) if t.get("artifact_id") == expected_hash]
|
||||
assert len(artifact_tags) == 0, "Tag should not be created when not specified"
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_upload_creates_artifact_in_database(self, integration_client, test_package):
|
||||
"""Test upload creates artifact record in database."""
|
||||
project, package = test_package
|
||||
content = b"database artifact test"
|
||||
expected_hash = compute_sha256(content)
|
||||
|
||||
upload_test_file(integration_client, project, package, content)
|
||||
|
||||
# Verify artifact exists via API
|
||||
response = integration_client.get(f"/api/v1/artifact/{expected_hash}")
|
||||
assert response.status_code == 200
|
||||
artifact = response.json()
|
||||
assert artifact["id"] == expected_hash
|
||||
assert artifact["size"] == len(content)
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_upload_creates_object_in_s3(self, integration_client, test_package):
|
||||
"""Test upload creates object in S3 storage."""
|
||||
project, package = test_package
|
||||
content = b"s3 object creation test"
|
||||
expected_hash = compute_sha256(content)
|
||||
|
||||
upload_test_file(integration_client, project, package, content)
|
||||
|
||||
# Verify S3 object exists
|
||||
assert s3_object_exists(expected_hash), "S3 object should exist after upload"
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_upload_with_tag_creates_tag_record(self, integration_client, test_package):
|
||||
"""Test upload with tag creates tag record."""
|
||||
project, package = test_package
|
||||
content = b"tag creation test"
|
||||
expected_hash = compute_sha256(content)
|
||||
tag_name = "my-tag-v1"
|
||||
|
||||
upload_test_file(
|
||||
integration_client, project, package, content, tag=tag_name
|
||||
)
|
||||
|
||||
# Verify tag exists
|
||||
tags_response = integration_client.get(
|
||||
f"/api/v1/project/{project}/{package}/tags"
|
||||
)
|
||||
assert tags_response.status_code == 200
|
||||
tags = tags_response.json()
|
||||
tag_names = [t["name"] for t in tags.get("items", tags)]
|
||||
assert tag_name in tag_names
|
||||
|
||||
|
||||
class TestDuplicateUploads:
|
||||
"""Tests for duplicate upload deduplication behavior."""
|
||||
@@ -248,6 +337,23 @@ class TestDownload:
|
||||
assert response.status_code == 200
|
||||
assert response.content == original_content
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_download_by_tag_prefix(self, integration_client, test_package):
|
||||
"""Test downloading artifact using tag: prefix."""
|
||||
project, package = test_package
|
||||
original_content = b"download by tag prefix test"
|
||||
|
||||
upload_test_file(
|
||||
integration_client, project, package, original_content, tag="prefix-tag"
|
||||
)
|
||||
|
||||
response = integration_client.get(
|
||||
f"/api/v1/project/{project}/{package}/+/tag:prefix-tag",
|
||||
params={"mode": "proxy"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.content == original_content
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_download_nonexistent_tag(self, integration_client, test_package):
|
||||
"""Test downloading nonexistent tag returns 404."""
|
||||
@@ -258,6 +364,33 @@ class TestDownload:
|
||||
)
|
||||
assert response.status_code == 404
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_download_nonexistent_artifact(self, integration_client, test_package):
|
||||
"""Test downloading nonexistent artifact ID returns 404."""
|
||||
project, package = test_package
|
||||
fake_hash = "0" * 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_from_nonexistent_project(self, integration_client, unique_test_id):
|
||||
"""Test downloading from nonexistent project returns 404."""
|
||||
response = integration_client.get(
|
||||
f"/api/v1/project/nonexistent-project-{unique_test_id}/somepackage/+/sometag"
|
||||
)
|
||||
assert response.status_code == 404
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_download_from_nonexistent_package(self, integration_client, test_project, unique_test_id):
|
||||
"""Test downloading from nonexistent package returns 404."""
|
||||
response = integration_client.get(
|
||||
f"/api/v1/project/{test_project}/nonexistent-package-{unique_test_id}/+/sometag"
|
||||
)
|
||||
assert response.status_code == 404
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_content_matches_original(self, integration_client, test_package):
|
||||
"""Test downloaded content matches original exactly."""
|
||||
@@ -275,6 +408,111 @@ class TestDownload:
|
||||
assert response.content == original_content
|
||||
|
||||
|
||||
class TestDownloadHeaders:
|
||||
"""Tests for download response headers."""
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_download_content_type_header(self, integration_client, test_package):
|
||||
"""Test download returns correct Content-Type header."""
|
||||
project, package = test_package
|
||||
content = b"content type header test"
|
||||
|
||||
upload_test_file(
|
||||
integration_client, project, package, content,
|
||||
filename="test.txt", tag="content-type-test"
|
||||
)
|
||||
|
||||
response = integration_client.get(
|
||||
f"/api/v1/project/{project}/{package}/+/content-type-test",
|
||||
params={"mode": "proxy"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
# Content-Type should be set (either text/plain or application/octet-stream)
|
||||
assert "content-type" in response.headers
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_download_content_length_header(self, integration_client, test_package):
|
||||
"""Test download returns correct Content-Length header."""
|
||||
project, package = test_package
|
||||
content = b"content length header test - exactly 41 bytes!"
|
||||
expected_length = len(content)
|
||||
|
||||
upload_test_file(
|
||||
integration_client, project, package, content, tag="content-length-test"
|
||||
)
|
||||
|
||||
response = integration_client.get(
|
||||
f"/api/v1/project/{project}/{package}/+/content-length-test",
|
||||
params={"mode": "proxy"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert "content-length" in response.headers
|
||||
assert int(response.headers["content-length"]) == expected_length
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_download_content_disposition_header(self, integration_client, test_package):
|
||||
"""Test download returns correct Content-Disposition header."""
|
||||
project, package = test_package
|
||||
content = b"content disposition test"
|
||||
filename = "my-test-file.bin"
|
||||
|
||||
upload_test_file(
|
||||
integration_client, project, package, content,
|
||||
filename=filename, tag="disposition-test"
|
||||
)
|
||||
|
||||
response = integration_client.get(
|
||||
f"/api/v1/project/{project}/{package}/+/disposition-test",
|
||||
params={"mode": "proxy"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert "content-disposition" in response.headers
|
||||
disposition = response.headers["content-disposition"]
|
||||
assert "attachment" in disposition
|
||||
assert filename in disposition
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_download_checksum_headers(self, integration_client, test_package):
|
||||
"""Test download returns checksum headers."""
|
||||
project, package = test_package
|
||||
content = b"checksum header test content"
|
||||
expected_hash = compute_sha256(content)
|
||||
|
||||
upload_test_file(
|
||||
integration_client, project, package, content, tag="checksum-headers"
|
||||
)
|
||||
|
||||
response = integration_client.get(
|
||||
f"/api/v1/project/{project}/{package}/+/checksum-headers",
|
||||
params={"mode": "proxy"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
# Check for checksum headers
|
||||
assert "x-checksum-sha256" in response.headers
|
||||
assert response.headers["x-checksum-sha256"] == expected_hash
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_download_etag_header(self, integration_client, test_package):
|
||||
"""Test download returns ETag header (artifact ID)."""
|
||||
project, package = test_package
|
||||
content = b"etag header test"
|
||||
expected_hash = compute_sha256(content)
|
||||
|
||||
upload_test_file(
|
||||
integration_client, project, package, content, tag="etag-test"
|
||||
)
|
||||
|
||||
response = integration_client.get(
|
||||
f"/api/v1/project/{project}/{package}/+/etag-test",
|
||||
params={"mode": "proxy"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert "etag" in response.headers
|
||||
# ETag should contain the artifact ID (hash)
|
||||
etag = response.headers["etag"].strip('"')
|
||||
assert etag == expected_hash
|
||||
|
||||
|
||||
class TestConcurrentUploads:
|
||||
"""Tests for concurrent upload handling."""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user