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
461 lines
16 KiB
Python
461 lines
16 KiB
Python
"""
|
|
Integration tests for download verification API endpoints.
|
|
|
|
These tests verify:
|
|
- Checksum headers in download responses
|
|
- Pre-verification mode
|
|
- Streaming verification mode
|
|
- HEAD request headers
|
|
- Verification failure handling
|
|
"""
|
|
|
|
import pytest
|
|
import hashlib
|
|
import base64
|
|
import io
|
|
|
|
|
|
# =============================================================================
|
|
# Test Fixtures
|
|
# =============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def upload_test_file(integration_client):
|
|
"""
|
|
Factory fixture to upload a test file and return its artifact ID.
|
|
|
|
Usage:
|
|
artifact_id = upload_test_file(project, package, content, version="v1.0")
|
|
"""
|
|
|
|
def _upload(project_name: str, package_name: str, content: bytes, tag: str = None):
|
|
files = {
|
|
"file": ("test-file.bin", io.BytesIO(content), "application/octet-stream")
|
|
}
|
|
data = {}
|
|
if tag:
|
|
data["tag"] = tag
|
|
|
|
response = integration_client.post(
|
|
f"/api/v1/project/{project_name}/{package_name}/upload",
|
|
files=files,
|
|
data=data,
|
|
)
|
|
assert response.status_code == 200, f"Upload failed: {response.text}"
|
|
return response.json()["artifact_id"]
|
|
|
|
return _upload
|
|
|
|
|
|
# =============================================================================
|
|
# Integration Tests - Download Headers
|
|
# =============================================================================
|
|
|
|
|
|
class TestDownloadChecksumHeaders:
|
|
"""Tests for checksum headers in download responses."""
|
|
|
|
@pytest.mark.integration
|
|
def test_download_includes_sha256_header(
|
|
self, integration_client, test_package, upload_test_file
|
|
):
|
|
"""Test download response includes X-Checksum-SHA256 header."""
|
|
project_name, package_name = test_package
|
|
content = b"Content for SHA256 header test"
|
|
|
|
# Upload file
|
|
artifact_id = upload_test_file(
|
|
project_name, package_name, content, version="sha256-header-test"
|
|
)
|
|
|
|
# Download with proxy mode
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project_name}/{package_name}/+/sha256-header-test",
|
|
params={"mode": "proxy"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "X-Checksum-SHA256" in response.headers
|
|
assert response.headers["X-Checksum-SHA256"] == artifact_id
|
|
|
|
@pytest.mark.integration
|
|
def test_download_includes_etag_header(
|
|
self, integration_client, test_package, upload_test_file
|
|
):
|
|
"""Test download response includes ETag header."""
|
|
project_name, package_name = test_package
|
|
content = b"Content for ETag header test"
|
|
|
|
artifact_id = upload_test_file(
|
|
project_name, package_name, content, version="etag-test"
|
|
)
|
|
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project_name}/{package_name}/+/etag-test",
|
|
params={"mode": "proxy"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "ETag" in response.headers
|
|
# ETag should be quoted artifact ID
|
|
assert response.headers["ETag"] == f'"{artifact_id}"'
|
|
|
|
@pytest.mark.integration
|
|
def test_download_includes_digest_header(
|
|
self, integration_client, test_package, upload_test_file
|
|
):
|
|
"""Test download response includes RFC 3230 Digest header."""
|
|
project_name, package_name = test_package
|
|
content = b"Content for Digest header test"
|
|
sha256 = hashlib.sha256(content).hexdigest()
|
|
|
|
upload_test_file(project_name, package_name, content, version="digest-test")
|
|
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project_name}/{package_name}/+/digest-test",
|
|
params={"mode": "proxy"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "Digest" in response.headers
|
|
|
|
# Verify Digest format: sha-256=<base64>
|
|
digest = response.headers["Digest"]
|
|
assert digest.startswith("sha-256=")
|
|
|
|
# Verify base64 content matches
|
|
b64_hash = digest.split("=", 1)[1]
|
|
decoded = base64.b64decode(b64_hash)
|
|
assert decoded == bytes.fromhex(sha256)
|
|
|
|
@pytest.mark.integration
|
|
def test_download_includes_content_length_header(
|
|
self, integration_client, test_package, upload_test_file
|
|
):
|
|
"""Test download response includes X-Content-Length header."""
|
|
project_name, package_name = test_package
|
|
content = b"Content for X-Content-Length test"
|
|
|
|
upload_test_file(project_name, package_name, content, version="content-length-test")
|
|
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project_name}/{package_name}/+/content-length-test",
|
|
params={"mode": "proxy"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "X-Content-Length" in response.headers
|
|
assert response.headers["X-Content-Length"] == str(len(content))
|
|
|
|
@pytest.mark.integration
|
|
def test_download_includes_verified_header_false(
|
|
self, integration_client, test_package, upload_test_file
|
|
):
|
|
"""Test download without verification has X-Verified: false."""
|
|
project_name, package_name = test_package
|
|
content = b"Content for X-Verified false test"
|
|
|
|
upload_test_file(project_name, package_name, content, version="verified-false-test")
|
|
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project_name}/{package_name}/+/verified-false-test",
|
|
params={"mode": "proxy", "verify": "false"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "X-Verified" in response.headers
|
|
assert response.headers["X-Verified"] == "false"
|
|
|
|
|
|
# =============================================================================
|
|
# Integration Tests - Pre-Verification Mode
|
|
# =============================================================================
|
|
|
|
|
|
class TestPreVerificationMode:
|
|
"""Tests for pre-verification download mode."""
|
|
|
|
@pytest.mark.integration
|
|
def test_pre_verify_success(
|
|
self, integration_client, test_package, upload_test_file
|
|
):
|
|
"""Test pre-verification mode succeeds for valid content."""
|
|
project_name, package_name = test_package
|
|
content = b"Content for pre-verification success test"
|
|
|
|
upload_test_file(project_name, package_name, content, version="pre-verify-success")
|
|
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project_name}/{package_name}/+/pre-verify-success",
|
|
params={"mode": "proxy", "verify": "true", "verify_mode": "pre"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.content == content
|
|
assert "X-Verified" in response.headers
|
|
assert response.headers["X-Verified"] == "true"
|
|
|
|
@pytest.mark.integration
|
|
def test_pre_verify_returns_complete_content(
|
|
self, integration_client, test_package, upload_test_file
|
|
):
|
|
"""Test pre-verification returns complete content."""
|
|
project_name, package_name = test_package
|
|
# Use binary content to verify no corruption
|
|
content = bytes(range(256)) * 10 # 2560 bytes of all byte values
|
|
|
|
upload_test_file(project_name, package_name, content, version="pre-verify-content")
|
|
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project_name}/{package_name}/+/pre-verify-content",
|
|
params={"mode": "proxy", "verify": "true", "verify_mode": "pre"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.content == content
|
|
|
|
|
|
# =============================================================================
|
|
# Integration Tests - Streaming Verification Mode
|
|
# =============================================================================
|
|
|
|
|
|
class TestStreamingVerificationMode:
|
|
"""Tests for streaming verification download mode."""
|
|
|
|
@pytest.mark.integration
|
|
def test_stream_verify_success(
|
|
self, integration_client, test_package, upload_test_file
|
|
):
|
|
"""Test streaming verification mode succeeds for valid content."""
|
|
project_name, package_name = test_package
|
|
content = b"Content for streaming verification success test"
|
|
|
|
upload_test_file(
|
|
project_name, package_name, content, version="stream-verify-success"
|
|
)
|
|
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project_name}/{package_name}/+/stream-verify-success",
|
|
params={"mode": "proxy", "verify": "true", "verify_mode": "stream"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.content == content
|
|
# X-Verified is "pending" for streaming mode (verified after transfer)
|
|
assert "X-Verified" in response.headers
|
|
|
|
@pytest.mark.integration
|
|
def test_stream_verify_large_content(
|
|
self, integration_client, test_package, upload_test_file
|
|
):
|
|
"""Test streaming verification with larger content."""
|
|
project_name, package_name = test_package
|
|
# 100KB of content
|
|
content = b"x" * (100 * 1024)
|
|
|
|
upload_test_file(project_name, package_name, content, version="stream-verify-large")
|
|
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project_name}/{package_name}/+/stream-verify-large",
|
|
params={"mode": "proxy", "verify": "true", "verify_mode": "stream"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.content == content
|
|
|
|
|
|
# =============================================================================
|
|
# Integration Tests - HEAD Request Headers
|
|
# =============================================================================
|
|
|
|
|
|
class TestHeadRequestHeaders:
|
|
"""Tests for HEAD request checksum headers."""
|
|
|
|
@pytest.mark.integration
|
|
def test_head_includes_sha256_header(
|
|
self, integration_client, test_package, upload_test_file
|
|
):
|
|
"""Test HEAD request includes X-Checksum-SHA256 header."""
|
|
project_name, package_name = test_package
|
|
content = b"Content for HEAD SHA256 test"
|
|
|
|
artifact_id = upload_test_file(
|
|
project_name, package_name, content, version="head-sha256-test"
|
|
)
|
|
|
|
response = integration_client.head(
|
|
f"/api/v1/project/{project_name}/{package_name}/+/head-sha256-test"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "X-Checksum-SHA256" in response.headers
|
|
assert response.headers["X-Checksum-SHA256"] == artifact_id
|
|
|
|
@pytest.mark.integration
|
|
def test_head_includes_etag(
|
|
self, integration_client, test_package, upload_test_file
|
|
):
|
|
"""Test HEAD request includes ETag header."""
|
|
project_name, package_name = test_package
|
|
content = b"Content for HEAD ETag test"
|
|
|
|
artifact_id = upload_test_file(
|
|
project_name, package_name, content, version="head-etag-test"
|
|
)
|
|
|
|
response = integration_client.head(
|
|
f"/api/v1/project/{project_name}/{package_name}/+/head-etag-test"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "ETag" in response.headers
|
|
assert response.headers["ETag"] == f'"{artifact_id}"'
|
|
|
|
@pytest.mark.integration
|
|
def test_head_includes_digest(
|
|
self, integration_client, test_package, upload_test_file
|
|
):
|
|
"""Test HEAD request includes Digest header."""
|
|
project_name, package_name = test_package
|
|
content = b"Content for HEAD Digest test"
|
|
|
|
upload_test_file(project_name, package_name, content, version="head-digest-test")
|
|
|
|
response = integration_client.head(
|
|
f"/api/v1/project/{project_name}/{package_name}/+/head-digest-test"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "Digest" in response.headers
|
|
assert response.headers["Digest"].startswith("sha-256=")
|
|
|
|
@pytest.mark.integration
|
|
def test_head_includes_content_length(
|
|
self, integration_client, test_package, upload_test_file
|
|
):
|
|
"""Test HEAD request includes X-Content-Length header."""
|
|
project_name, package_name = test_package
|
|
content = b"Content for HEAD Content-Length test"
|
|
|
|
upload_test_file(project_name, package_name, content, version="head-length-test")
|
|
|
|
response = integration_client.head(
|
|
f"/api/v1/project/{project_name}/{package_name}/+/head-length-test"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert "X-Content-Length" in response.headers
|
|
assert response.headers["X-Content-Length"] == str(len(content))
|
|
|
|
@pytest.mark.integration
|
|
def test_head_no_body(self, integration_client, test_package, upload_test_file):
|
|
"""Test HEAD request returns no body."""
|
|
project_name, package_name = test_package
|
|
content = b"Content for HEAD no-body test"
|
|
|
|
upload_test_file(project_name, package_name, content, version="head-no-body-test")
|
|
|
|
response = integration_client.head(
|
|
f"/api/v1/project/{project_name}/{package_name}/+/head-no-body-test"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.content == b""
|
|
|
|
|
|
# =============================================================================
|
|
# Integration Tests - Range Requests
|
|
# =============================================================================
|
|
|
|
|
|
class TestRangeRequestHeaders:
|
|
"""Tests for range request handling with checksum headers."""
|
|
|
|
@pytest.mark.integration
|
|
def test_range_request_includes_checksum_headers(
|
|
self, integration_client, test_package, upload_test_file
|
|
):
|
|
"""Test range request includes checksum headers."""
|
|
project_name, package_name = test_package
|
|
content = b"Content for range request checksum header test"
|
|
|
|
upload_test_file(project_name, package_name, content, version="range-checksum-test")
|
|
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project_name}/{package_name}/+/range-checksum-test",
|
|
headers={"Range": "bytes=0-9"},
|
|
params={"mode": "proxy"},
|
|
)
|
|
|
|
assert response.status_code == 206
|
|
assert "X-Checksum-SHA256" in response.headers
|
|
# Checksum is for the FULL file, not the range
|
|
assert len(response.headers["X-Checksum-SHA256"]) == 64
|
|
|
|
|
|
# =============================================================================
|
|
# Integration Tests - Client-Side Verification
|
|
# =============================================================================
|
|
|
|
|
|
class TestClientSideVerification:
|
|
"""Tests demonstrating client-side verification using headers."""
|
|
|
|
@pytest.mark.integration
|
|
def test_client_can_verify_downloaded_content(
|
|
self, integration_client, test_package, upload_test_file
|
|
):
|
|
"""Test client can verify downloaded content using header."""
|
|
project_name, package_name = test_package
|
|
content = b"Content for client-side verification test"
|
|
|
|
upload_test_file(project_name, package_name, content, version="client-verify-test")
|
|
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project_name}/{package_name}/+/client-verify-test",
|
|
params={"mode": "proxy"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
# Get expected hash from header
|
|
expected_hash = response.headers["X-Checksum-SHA256"]
|
|
|
|
# Compute actual hash of downloaded content
|
|
actual_hash = hashlib.sha256(response.content).hexdigest()
|
|
|
|
# Verify match
|
|
assert actual_hash == expected_hash
|
|
|
|
@pytest.mark.integration
|
|
def test_client_can_verify_using_digest_header(
|
|
self, integration_client, test_package, upload_test_file
|
|
):
|
|
"""Test client can verify using RFC 3230 Digest header."""
|
|
project_name, package_name = test_package
|
|
content = b"Content for Digest header verification"
|
|
|
|
upload_test_file(project_name, package_name, content, version="digest-verify-test")
|
|
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project_name}/{package_name}/+/digest-verify-test",
|
|
params={"mode": "proxy"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
# Parse Digest header
|
|
digest_header = response.headers["Digest"]
|
|
assert digest_header.startswith("sha-256=")
|
|
b64_hash = digest_header.split("=", 1)[1]
|
|
expected_hash_bytes = base64.b64decode(b64_hash)
|
|
|
|
# Compute actual hash of downloaded content
|
|
actual_hash_bytes = hashlib.sha256(response.content).digest()
|
|
|
|
# Verify match
|
|
assert actual_hash_bytes == expected_hash_bytes
|