Complete metadata query EPIC: add upload endpoints, enhance response fields, standardize audit actions (#18, #19, #20, #22)
- Add GET /api/v1/uploads global endpoint with project/package/user/date filters
- Add GET /api/v1/project/{project}/uploads project-level uploads endpoint
- Add has_more field to PaginationMeta for pagination UI
- Add upload_id, content_type, original_name, created_at to UploadResponse
- Standardize audit action names: project.delete, package.delete, tag.delete, artifact.upload
- Add 13 new integration tests for upload query endpoints and response fields
- 130 tests passing
This commit is contained in:
@@ -307,3 +307,216 @@ class TestArtifactProvenance:
|
||||
assert "project_name" in tag
|
||||
assert "package_name" in tag
|
||||
assert "tag_name" in tag
|
||||
|
||||
|
||||
class TestGlobalUploadsEndpoint:
|
||||
"""Tests for /api/v1/uploads endpoint (global admin)."""
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_global_uploads_returns_200(self, integration_client):
|
||||
"""Test that global uploads endpoint returns 200."""
|
||||
response = integration_client.get("/api/v1/uploads")
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.json()
|
||||
assert "items" in data
|
||||
assert "pagination" in data
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_global_uploads_pagination(self, integration_client):
|
||||
"""Test that global uploads endpoint respects pagination."""
|
||||
response = integration_client.get("/api/v1/uploads?limit=5&page=1")
|
||||
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_global_uploads_filter_by_project(self, integration_client, test_package):
|
||||
"""Test filtering global uploads by project name."""
|
||||
project_name, package_name = test_package
|
||||
|
||||
# Upload a file
|
||||
upload_test_file(
|
||||
integration_client,
|
||||
project_name,
|
||||
package_name,
|
||||
b"global filter test",
|
||||
"global.txt",
|
||||
)
|
||||
|
||||
response = integration_client.get(f"/api/v1/uploads?project={project_name}")
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.json()
|
||||
for item in data["items"]:
|
||||
assert item["project_name"] == project_name
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_global_uploads_filter_by_uploader(self, integration_client, test_package):
|
||||
"""Test filtering global uploads by uploaded_by."""
|
||||
project_name, package_name = test_package
|
||||
|
||||
# Upload a file
|
||||
upload_test_file(
|
||||
integration_client,
|
||||
project_name,
|
||||
package_name,
|
||||
b"uploader filter test",
|
||||
"uploader.txt",
|
||||
)
|
||||
|
||||
# Filter by anonymous (default user)
|
||||
response = integration_client.get("/api/v1/uploads?uploaded_by=anonymous")
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.json()
|
||||
for item in data["items"]:
|
||||
assert item["uploaded_by"] == "anonymous"
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_global_uploads_has_more_field(self, integration_client):
|
||||
"""Test that pagination includes has_more field."""
|
||||
response = integration_client.get("/api/v1/uploads?limit=1")
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.json()
|
||||
assert "has_more" in data["pagination"]
|
||||
assert isinstance(data["pagination"]["has_more"], bool)
|
||||
|
||||
|
||||
class TestProjectUploadsEndpoint:
|
||||
"""Tests for /api/v1/project/{project}/uploads endpoint."""
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_project_uploads_returns_200(self, integration_client, test_project):
|
||||
"""Test that project uploads endpoint returns 200."""
|
||||
response = integration_client.get(f"/api/v1/project/{test_project}/uploads")
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.json()
|
||||
assert "items" in data
|
||||
assert "pagination" in data
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_project_uploads_after_upload(self, integration_client, test_package):
|
||||
"""Test that uploads are recorded in project uploads."""
|
||||
project_name, package_name = test_package
|
||||
|
||||
# Upload a file
|
||||
upload_test_file(
|
||||
integration_client,
|
||||
project_name,
|
||||
package_name,
|
||||
b"project uploads test",
|
||||
"project.txt",
|
||||
)
|
||||
|
||||
response = integration_client.get(f"/api/v1/project/{project_name}/uploads")
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.json()
|
||||
assert len(data["items"]) >= 1
|
||||
|
||||
# Verify project name matches
|
||||
for item in data["items"]:
|
||||
assert item["project_name"] == project_name
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_project_uploads_filter_by_package(self, integration_client, test_package):
|
||||
"""Test filtering project uploads by package name."""
|
||||
project_name, package_name = test_package
|
||||
|
||||
# Upload a file
|
||||
upload_test_file(
|
||||
integration_client,
|
||||
project_name,
|
||||
package_name,
|
||||
b"package filter test",
|
||||
"pkgfilter.txt",
|
||||
)
|
||||
|
||||
response = integration_client.get(
|
||||
f"/api/v1/project/{project_name}/uploads?package={package_name}"
|
||||
)
|
||||
assert response.status_code == 200
|
||||
|
||||
data = response.json()
|
||||
for item in data["items"]:
|
||||
assert item["package_name"] == package_name
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_project_uploads_not_found(self, integration_client):
|
||||
"""Test that non-existent project returns 404."""
|
||||
response = integration_client.get("/api/v1/project/nonexistent/uploads")
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
class TestUploadResponseFields:
|
||||
"""Tests for enhanced UploadResponse fields (Issue #19)."""
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_upload_response_has_upload_id(self, integration_client, test_package):
|
||||
"""Test that upload response includes upload_id."""
|
||||
project_name, package_name = test_package
|
||||
|
||||
upload_result = upload_test_file(
|
||||
integration_client,
|
||||
project_name,
|
||||
package_name,
|
||||
b"upload id test",
|
||||
"uploadid.txt",
|
||||
)
|
||||
|
||||
# upload_id should be present
|
||||
assert "upload_id" in upload_result
|
||||
assert upload_result["upload_id"] is not None
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_upload_response_has_content_type(self, integration_client, test_package):
|
||||
"""Test that upload response includes content_type."""
|
||||
project_name, package_name = test_package
|
||||
|
||||
upload_result = upload_test_file(
|
||||
integration_client,
|
||||
project_name,
|
||||
package_name,
|
||||
b"content type test",
|
||||
"content.txt",
|
||||
)
|
||||
|
||||
assert "content_type" in upload_result
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_upload_response_has_original_name(self, integration_client, test_package):
|
||||
"""Test that upload response includes original_name."""
|
||||
project_name, package_name = test_package
|
||||
|
||||
upload_result = upload_test_file(
|
||||
integration_client,
|
||||
project_name,
|
||||
package_name,
|
||||
b"original name test",
|
||||
"originalname.txt",
|
||||
)
|
||||
|
||||
assert "original_name" in upload_result
|
||||
assert upload_result["original_name"] == "originalname.txt"
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_upload_response_has_created_at(self, integration_client, test_package):
|
||||
"""Test that upload response includes created_at."""
|
||||
project_name, package_name = test_package
|
||||
|
||||
upload_result = upload_test_file(
|
||||
integration_client,
|
||||
project_name,
|
||||
package_name,
|
||||
b"created at test",
|
||||
"createdat.txt",
|
||||
)
|
||||
|
||||
assert "created_at" in upload_result
|
||||
assert upload_result["created_at"] is not None
|
||||
|
||||
Reference in New Issue
Block a user