348 lines
13 KiB
Python
348 lines
13 KiB
Python
"""
|
|
Integration tests for package version API endpoints.
|
|
|
|
Tests cover:
|
|
- Version creation via upload
|
|
- Version auto-detection from filename
|
|
- Version listing and retrieval
|
|
- Download by version prefix
|
|
- Version deletion
|
|
"""
|
|
|
|
import pytest
|
|
import io
|
|
from tests.factories import (
|
|
compute_sha256,
|
|
upload_test_file,
|
|
)
|
|
|
|
|
|
class TestVersionCreation:
|
|
"""Tests for creating versions via upload."""
|
|
|
|
@pytest.mark.integration
|
|
def test_upload_with_explicit_version(self, integration_client, test_package):
|
|
"""Test upload with explicit version parameter creates version record."""
|
|
project, package = test_package
|
|
content = b"version creation test"
|
|
expected_hash = compute_sha256(content)
|
|
|
|
files = {"file": ("app.tar.gz", io.BytesIO(content), "application/octet-stream")}
|
|
response = integration_client.post(
|
|
f"/api/v1/project/{project}/{package}/upload",
|
|
files=files,
|
|
data={"version": "1.0.0"},
|
|
)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result["artifact_id"] == expected_hash
|
|
assert result.get("version") == "1.0.0"
|
|
assert result.get("version_source") == "explicit"
|
|
|
|
@pytest.mark.integration
|
|
def test_upload_with_version_and_tag(self, integration_client, test_package):
|
|
"""Test upload with both version and tag creates both records."""
|
|
project, package = test_package
|
|
content = b"version and tag test"
|
|
|
|
files = {"file": ("app.tar.gz", io.BytesIO(content), "application/octet-stream")}
|
|
response = integration_client.post(
|
|
f"/api/v1/project/{project}/{package}/upload",
|
|
files=files,
|
|
data={"version": "2.0.0", "tag": "latest"},
|
|
)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result.get("version") == "2.0.0"
|
|
|
|
# Verify tag was also created
|
|
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 "latest" in tag_names
|
|
|
|
@pytest.mark.integration
|
|
def test_duplicate_version_same_content_succeeds(self, integration_client, test_package):
|
|
"""Test uploading same version with same content succeeds (deduplication)."""
|
|
project, package = test_package
|
|
content = b"version dedup test"
|
|
|
|
# First upload with version
|
|
files1 = {"file": ("app1.tar.gz", io.BytesIO(content), "application/octet-stream")}
|
|
response1 = integration_client.post(
|
|
f"/api/v1/project/{project}/{package}/upload",
|
|
files=files1,
|
|
data={"version": "3.0.0"},
|
|
)
|
|
assert response1.status_code == 200
|
|
|
|
# Second upload with same version and same content succeeds
|
|
files2 = {"file": ("app2.tar.gz", io.BytesIO(content), "application/octet-stream")}
|
|
response2 = integration_client.post(
|
|
f"/api/v1/project/{project}/{package}/upload",
|
|
files=files2,
|
|
data={"version": "3.0.0"},
|
|
)
|
|
# This succeeds because it's the same artifact (deduplication)
|
|
assert response2.status_code == 200
|
|
|
|
|
|
class TestVersionAutoDetection:
|
|
"""Tests for automatic version detection from filename."""
|
|
|
|
@pytest.mark.integration
|
|
def test_version_detected_from_filename_tarball(self, integration_client, test_package):
|
|
"""Test version is auto-detected from tarball filename or metadata."""
|
|
project, package = test_package
|
|
content = b"auto detect version tarball"
|
|
|
|
files = {"file": ("myapp-1.2.3.tar.gz", io.BytesIO(content), "application/octet-stream")}
|
|
response = integration_client.post(
|
|
f"/api/v1/project/{project}/{package}/upload",
|
|
files=files,
|
|
)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result.get("version") == "1.2.3"
|
|
# Version source can be 'filename' or 'metadata' depending on detection order
|
|
assert result.get("version_source") in ["filename", "metadata"]
|
|
|
|
@pytest.mark.integration
|
|
def test_version_detected_from_filename_zip(self, integration_client, test_package):
|
|
"""Test version is auto-detected from zip filename."""
|
|
project, package = test_package
|
|
content = b"auto detect version zip"
|
|
|
|
files = {"file": ("package-2.0.0.zip", io.BytesIO(content), "application/octet-stream")}
|
|
response = integration_client.post(
|
|
f"/api/v1/project/{project}/{package}/upload",
|
|
files=files,
|
|
)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result.get("version") == "2.0.0"
|
|
assert result.get("version_source") == "filename"
|
|
|
|
@pytest.mark.integration
|
|
def test_explicit_version_overrides_filename(self, integration_client, test_package):
|
|
"""Test explicit version parameter overrides filename detection."""
|
|
project, package = test_package
|
|
content = b"explicit override test"
|
|
|
|
files = {"file": ("myapp-1.0.0.tar.gz", io.BytesIO(content), "application/octet-stream")}
|
|
response = integration_client.post(
|
|
f"/api/v1/project/{project}/{package}/upload",
|
|
files=files,
|
|
data={"version": "9.9.9"},
|
|
)
|
|
assert response.status_code == 200
|
|
result = response.json()
|
|
assert result.get("version") == "9.9.9"
|
|
assert result.get("version_source") == "explicit"
|
|
|
|
@pytest.mark.integration
|
|
def test_no_version_detected_from_plain_filename(self, integration_client, test_package):
|
|
"""Test no version is created for filenames without version pattern."""
|
|
project, package = test_package
|
|
content = b"no version in filename"
|
|
|
|
files = {"file": ("plain-file.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
|
|
result = response.json()
|
|
# Version should be None or not present
|
|
assert result.get("version") is None
|
|
|
|
|
|
class TestVersionListing:
|
|
"""Tests for listing and retrieving versions."""
|
|
|
|
@pytest.mark.integration
|
|
def test_list_versions(self, integration_client, test_package):
|
|
"""Test listing all versions for a package."""
|
|
project, package = test_package
|
|
|
|
# Create multiple versions
|
|
for ver in ["1.0.0", "1.1.0", "2.0.0"]:
|
|
content = f"version {ver} content".encode()
|
|
files = {"file": (f"app-{ver}.tar.gz", io.BytesIO(content), "application/octet-stream")}
|
|
response = integration_client.post(
|
|
f"/api/v1/project/{project}/{package}/upload",
|
|
files=files,
|
|
data={"version": ver},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
# List versions
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project}/{package}/versions"
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
versions = [v["version"] for v in data.get("items", data)]
|
|
assert "1.0.0" in versions
|
|
assert "1.1.0" in versions
|
|
assert "2.0.0" in versions
|
|
|
|
@pytest.mark.integration
|
|
def test_get_specific_version(self, integration_client, test_package):
|
|
"""Test getting details for a specific version."""
|
|
project, package = test_package
|
|
content = b"specific version test"
|
|
expected_hash = compute_sha256(content)
|
|
|
|
# Create version
|
|
files = {"file": ("app-4.0.0.tar.gz", io.BytesIO(content), "application/octet-stream")}
|
|
integration_client.post(
|
|
f"/api/v1/project/{project}/{package}/upload",
|
|
files=files,
|
|
data={"version": "4.0.0"},
|
|
)
|
|
|
|
# Get version details
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project}/{package}/versions/4.0.0"
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["version"] == "4.0.0"
|
|
assert data["artifact_id"] == expected_hash
|
|
|
|
@pytest.mark.integration
|
|
def test_get_nonexistent_version_returns_404(self, integration_client, test_package):
|
|
"""Test getting nonexistent version returns 404."""
|
|
project, package = test_package
|
|
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project}/{package}/versions/99.99.99"
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
class TestDownloadByVersion:
|
|
"""Tests for downloading artifacts by version."""
|
|
|
|
@pytest.mark.integration
|
|
def test_download_by_version_prefix(self, integration_client, test_package):
|
|
"""Test downloading artifact using version: prefix."""
|
|
project, package = test_package
|
|
content = b"download by version test"
|
|
expected_hash = compute_sha256(content)
|
|
|
|
# Upload with version
|
|
files = {"file": ("app.tar.gz", io.BytesIO(content), "application/octet-stream")}
|
|
integration_client.post(
|
|
f"/api/v1/project/{project}/{package}/upload",
|
|
files=files,
|
|
data={"version": "5.0.0"},
|
|
)
|
|
|
|
# Download by version prefix
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project}/{package}/+/version:5.0.0",
|
|
params={"mode": "proxy"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.content == content
|
|
|
|
@pytest.mark.integration
|
|
def test_download_nonexistent_version_returns_404(self, integration_client, test_package):
|
|
"""Test downloading nonexistent version returns 404."""
|
|
project, package = test_package
|
|
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project}/{package}/+/version:99.0.0"
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.integration
|
|
def test_version_resolution_priority(self, integration_client, test_package):
|
|
"""Test that version: prefix explicitly resolves to version, not tag."""
|
|
project, package = test_package
|
|
version_content = b"this is the version content"
|
|
tag_content = b"this is the tag content"
|
|
|
|
# Create a version 6.0.0
|
|
files1 = {"file": ("app-v.tar.gz", io.BytesIO(version_content), "application/octet-stream")}
|
|
integration_client.post(
|
|
f"/api/v1/project/{project}/{package}/upload",
|
|
files=files1,
|
|
data={"version": "6.0.0"},
|
|
)
|
|
|
|
# Create a tag named "6.0.0" pointing to different content
|
|
files2 = {"file": ("app-t.tar.gz", io.BytesIO(tag_content), "application/octet-stream")}
|
|
integration_client.post(
|
|
f"/api/v1/project/{project}/{package}/upload",
|
|
files=files2,
|
|
data={"tag": "6.0.0"},
|
|
)
|
|
|
|
# Download with version: prefix should get version content
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project}/{package}/+/version:6.0.0",
|
|
params={"mode": "proxy"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.content == version_content
|
|
|
|
# Download with tag: prefix should get tag content
|
|
response2 = integration_client.get(
|
|
f"/api/v1/project/{project}/{package}/+/tag:6.0.0",
|
|
params={"mode": "proxy"},
|
|
)
|
|
assert response2.status_code == 200
|
|
assert response2.content == tag_content
|
|
|
|
|
|
class TestVersionDeletion:
|
|
"""Tests for deleting versions."""
|
|
|
|
@pytest.mark.integration
|
|
def test_delete_version(self, integration_client, test_package):
|
|
"""Test deleting a version."""
|
|
project, package = test_package
|
|
content = b"delete version test"
|
|
|
|
# Create version
|
|
files = {"file": ("app.tar.gz", io.BytesIO(content), "application/octet-stream")}
|
|
integration_client.post(
|
|
f"/api/v1/project/{project}/{package}/upload",
|
|
files=files,
|
|
data={"version": "7.0.0"},
|
|
)
|
|
|
|
# Verify version exists
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project}/{package}/versions/7.0.0"
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
# Delete version - returns 204 No Content on success
|
|
delete_response = integration_client.delete(
|
|
f"/api/v1/project/{project}/{package}/versions/7.0.0"
|
|
)
|
|
assert delete_response.status_code == 204
|
|
|
|
# Verify version no longer exists
|
|
response2 = integration_client.get(
|
|
f"/api/v1/project/{project}/{package}/versions/7.0.0"
|
|
)
|
|
assert response2.status_code == 404
|
|
|
|
@pytest.mark.integration
|
|
def test_delete_nonexistent_version_returns_404(self, integration_client, test_package):
|
|
"""Test deleting nonexistent version returns 404."""
|
|
project, package = test_package
|
|
|
|
response = integration_client.delete(
|
|
f"/api/v1/project/{project}/{package}/versions/99.0.0"
|
|
)
|
|
assert response.status_code == 404
|