344 lines
12 KiB
Python
344 lines
12 KiB
Python
"""
|
|
Integration tests for package API endpoints.
|
|
|
|
Tests cover:
|
|
- Package CRUD operations
|
|
- Package listing with pagination, search, filtering
|
|
- Package stats endpoint
|
|
- Package-level audit logs
|
|
- Cascade delete behavior
|
|
"""
|
|
|
|
import pytest
|
|
from tests.factories import compute_sha256, upload_test_file
|
|
|
|
|
|
class TestPackageCRUD:
|
|
"""Tests for package create, read, update, delete operations."""
|
|
|
|
@pytest.mark.integration
|
|
def test_create_package(self, integration_client, test_project, unique_test_id):
|
|
"""Test creating a new package."""
|
|
package_name = f"test-create-pkg-{unique_test_id}"
|
|
|
|
response = integration_client.post(
|
|
f"/api/v1/project/{test_project}/packages",
|
|
json={
|
|
"name": package_name,
|
|
"description": "Test package",
|
|
"format": "npm",
|
|
"platform": "linux",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert data["name"] == package_name
|
|
assert data["description"] == "Test package"
|
|
assert data["format"] == "npm"
|
|
assert data["platform"] == "linux"
|
|
|
|
@pytest.mark.integration
|
|
def test_get_package(self, integration_client, test_package):
|
|
"""Test getting a package by name."""
|
|
project_name, package_name = test_package
|
|
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project_name}/packages/{package_name}"
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert data["name"] == package_name
|
|
|
|
@pytest.mark.integration
|
|
def test_get_nonexistent_package(self, integration_client, test_project):
|
|
"""Test getting a non-existent package returns 404."""
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{test_project}/packages/nonexistent-pkg"
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.integration
|
|
def test_list_packages(self, integration_client, test_package):
|
|
"""Test listing packages includes created package."""
|
|
project_name, package_name = test_package
|
|
|
|
response = integration_client.get(f"/api/v1/project/{project_name}/packages")
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert "items" in data
|
|
assert "pagination" in data
|
|
|
|
package_names = [p["name"] for p in data["items"]]
|
|
assert package_name in package_names
|
|
|
|
@pytest.mark.integration
|
|
def test_delete_package(self, integration_client, test_project, unique_test_id):
|
|
"""Test deleting a package."""
|
|
package_name = f"test-delete-pkg-{unique_test_id}"
|
|
|
|
# Create package
|
|
integration_client.post(
|
|
f"/api/v1/project/{test_project}/packages",
|
|
json={"name": package_name, "description": "To be deleted"},
|
|
)
|
|
|
|
# Delete package
|
|
response = integration_client.delete(
|
|
f"/api/v1/project/{test_project}/packages/{package_name}"
|
|
)
|
|
assert response.status_code == 204
|
|
|
|
# Verify deleted
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{test_project}/packages/{package_name}"
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
class TestPackageListingFilters:
|
|
"""Tests for package listing with filters and pagination."""
|
|
|
|
@pytest.mark.integration
|
|
def test_packages_pagination(self, integration_client, test_project):
|
|
"""Test package listing respects pagination parameters."""
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{test_project}/packages?page=1&limit=5"
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert len(data["items"]) <= 5
|
|
assert data["pagination"]["limit"] == 5
|
|
assert data["pagination"]["page"] == 1
|
|
|
|
@pytest.mark.integration
|
|
def test_packages_filter_by_format(
|
|
self, integration_client, test_project, unique_test_id
|
|
):
|
|
"""Test package filtering by format."""
|
|
# Create a package with specific format
|
|
package_name = f"npm-pkg-{unique_test_id}"
|
|
integration_client.post(
|
|
f"/api/v1/project/{test_project}/packages",
|
|
json={"name": package_name, "format": "npm"},
|
|
)
|
|
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{test_project}/packages?format=npm"
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
for pkg in data["items"]:
|
|
assert pkg["format"] == "npm"
|
|
|
|
@pytest.mark.integration
|
|
def test_packages_filter_by_platform(
|
|
self, integration_client, test_project, unique_test_id
|
|
):
|
|
"""Test package filtering by platform."""
|
|
# Create a package with specific platform
|
|
package_name = f"linux-pkg-{unique_test_id}"
|
|
integration_client.post(
|
|
f"/api/v1/project/{test_project}/packages",
|
|
json={"name": package_name, "platform": "linux"},
|
|
)
|
|
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{test_project}/packages?platform=linux"
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
for pkg in data["items"]:
|
|
assert pkg["platform"] == "linux"
|
|
|
|
|
|
class TestPackageStats:
|
|
"""Tests for package statistics endpoint."""
|
|
|
|
@pytest.mark.integration
|
|
def test_package_stats_returns_valid_response(
|
|
self, integration_client, test_package
|
|
):
|
|
"""Test package stats endpoint returns expected fields."""
|
|
project, package = test_package
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project}/packages/{package}/stats"
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert "package_id" in data
|
|
assert "package_name" in data
|
|
assert "project_name" in data
|
|
assert "version_count" in data
|
|
assert "artifact_count" in data
|
|
assert "total_size_bytes" in data
|
|
assert "upload_count" in data
|
|
assert "deduplicated_uploads" in data
|
|
assert "storage_saved_bytes" in data
|
|
assert "deduplication_ratio" in data
|
|
|
|
@pytest.mark.integration
|
|
def test_package_stats_not_found(self, integration_client, test_project):
|
|
"""Test package stats returns 404 for non-existent package."""
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{test_project}/packages/nonexistent-package/stats"
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
class TestPackageAuditLogs:
|
|
"""Tests for package-level audit logs endpoint."""
|
|
|
|
@pytest.mark.integration
|
|
def test_package_audit_logs_returns_200(self, integration_client, test_package):
|
|
"""Test package audit logs endpoint returns 200."""
|
|
project_name, package_name = test_package
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project_name}/{package_name}/audit-logs"
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert "items" in data
|
|
assert "pagination" in data
|
|
|
|
@pytest.mark.integration
|
|
def test_package_audit_logs_project_not_found(self, integration_client):
|
|
"""Test non-existent project returns 404."""
|
|
response = integration_client.get(
|
|
"/api/v1/project/nonexistent/nonexistent/audit-logs"
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.integration
|
|
def test_package_audit_logs_package_not_found(
|
|
self, integration_client, test_project
|
|
):
|
|
"""Test non-existent package returns 404."""
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{test_project}/nonexistent-package/audit-logs"
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
class TestPackageCascadeDelete:
|
|
"""Tests for cascade delete behavior when deleting packages."""
|
|
|
|
@pytest.mark.integration
|
|
def test_ref_count_decrements_on_package_delete(
|
|
self, integration_client, unique_test_id
|
|
):
|
|
"""Test ref_count decrements when package is deleted.
|
|
|
|
Each package can only have one version per artifact (same content = same version).
|
|
This test verifies that deleting a package decrements the artifact's ref_count.
|
|
"""
|
|
project_name = f"cascade-pkg-{unique_test_id}"
|
|
package_name = f"test-pkg-{unique_test_id}"
|
|
|
|
# Create project
|
|
response = integration_client.post(
|
|
"/api/v1/projects",
|
|
json={
|
|
"name": project_name,
|
|
"description": "Test project",
|
|
"is_public": True,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
# Create package
|
|
response = integration_client.post(
|
|
f"/api/v1/project/{project_name}/packages",
|
|
json={"name": package_name, "description": "Test package"},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
# Upload content with version
|
|
content = f"cascade delete test {unique_test_id}".encode()
|
|
expected_hash = compute_sha256(content)
|
|
|
|
upload_test_file(
|
|
integration_client, project_name, package_name, content, version="1.0.0"
|
|
)
|
|
|
|
# Verify ref_count is 1
|
|
response = integration_client.get(f"/api/v1/artifact/{expected_hash}")
|
|
assert response.json()["ref_count"] == 1
|
|
|
|
# Delete the package
|
|
delete_response = integration_client.delete(
|
|
f"/api/v1/project/{project_name}/packages/{package_name}"
|
|
)
|
|
assert delete_response.status_code == 204
|
|
|
|
# Verify ref_count is 0
|
|
response = integration_client.get(f"/api/v1/artifact/{expected_hash}")
|
|
assert response.json()["ref_count"] == 0
|
|
|
|
# Cleanup
|
|
integration_client.delete(f"/api/v1/projects/{project_name}")
|
|
|
|
|
|
class TestPackageUploads:
|
|
"""Tests for package-level uploads endpoint."""
|
|
|
|
@pytest.mark.integration
|
|
def test_package_uploads_returns_200(self, integration_client, test_package):
|
|
"""Test package uploads endpoint returns 200."""
|
|
project_name, package_name = test_package
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project_name}/{package_name}/uploads"
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert "items" in data
|
|
assert "pagination" in data
|
|
|
|
@pytest.mark.integration
|
|
def test_package_uploads_after_upload(self, integration_client, test_package):
|
|
"""Test uploads are recorded after file upload."""
|
|
project_name, package_name = test_package
|
|
|
|
# Upload a file
|
|
upload_result = upload_test_file(
|
|
integration_client,
|
|
project_name,
|
|
package_name,
|
|
b"test upload content",
|
|
"test.txt",
|
|
)
|
|
assert upload_result["artifact_id"]
|
|
|
|
# Check uploads endpoint
|
|
response = integration_client.get(
|
|
f"/api/v1/project/{project_name}/{package_name}/uploads"
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert len(data["items"]) >= 1
|
|
|
|
# Verify upload record fields
|
|
upload = data["items"][0]
|
|
assert "artifact_id" in upload
|
|
assert "package_name" in upload
|
|
assert "project_name" in upload
|
|
assert "uploaded_at" in upload
|
|
assert "uploaded_by" in upload
|
|
|
|
@pytest.mark.integration
|
|
def test_package_uploads_project_not_found(self, integration_client):
|
|
"""Test non-existent project returns 404."""
|
|
response = integration_client.get(
|
|
"/api/v1/project/nonexistent/nonexistent/uploads"
|
|
)
|
|
assert response.status_code == 404
|