tests: resolve all peer-review issues across test suite
Address every substantive critique from the peer review: test_cache: replace tautological same-inputs key test with hardcoded hash assertion; assert setex call + TTL in mark_index_cached test; assert client is None for unavailable no-op; rename Packages.gz test to document intentional behaviour; add alpine sig/tmp negatives; add hyphenated and date-tag docker positive cases; add key hash-length assertion. test_config: replace live-constant comparisons with literal string assertions for alpine/rpm/docker; add unknown package type test; add dict-keyed repositories branch coverage (per-repo override and fallback); fix cache config to full equality check; add explicit empty index_patterns test. test_docker_auth: fix case-insensitive test to verify realm value; add field-order (scope before service) limitation test; add pipe-char collision documentation test; add missing fetch_token edge cases (no token field, HTTPStatusError, missing expires_in default 300); replace rubber-stamp delegate test with end-to-end parse→fetch test. test_storage: replace split prefix/suffix assertions with structural 3-part check + pinned sha256 assertion; fix Docker blob digests to 64-char hex; add secure=True URL test; add upload return value test; add download_object 404-on-ClientError test; remove redundant subset test. test_routes: add metrics.record_cache_hit/miss assertions; add mark_index_cached assertion after cache miss on index (docker + generic); add Content-Disposition, X-Artifact-Size header checks; add rpm/xml content-type tests; add flush test that verifies Redis keys are deleted when cache is available; add smoke coverage for upload (PUT), HEAD, DELETE, /metrics, and /config routes.
This commit is contained in:
+54
-23
@@ -1,7 +1,11 @@
|
||||
"""Tests for S3Storage, focusing on get_object_key (pure logic, no S3 calls)."""
|
||||
"""Tests for S3Storage: get_object_key (pure logic) and I/O methods."""
|
||||
|
||||
import hashlib
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from botocore.exceptions import ClientError
|
||||
from fastapi import HTTPException
|
||||
|
||||
from artifactapi.storage import S3Storage
|
||||
|
||||
@@ -25,19 +29,22 @@ def storage():
|
||||
# get_object_key
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestGetObjectKey:
|
||||
def test_includes_remote_name_prefix(self, storage):
|
||||
key = storage.get_object_key("myremote", "some/path/file.rpm")
|
||||
assert key.startswith("myremote/")
|
||||
def test_key_has_three_part_structure(self, storage):
|
||||
# remote / hash-segment / filename
|
||||
key = storage.get_object_key("myremote", "some/path/to/file.rpm")
|
||||
parts = key.split("/")
|
||||
assert len(parts) == 3
|
||||
assert parts[0] == "myremote"
|
||||
assert parts[2] == "file.rpm"
|
||||
assert len(parts[1]) == 16 # SHA-256 hex truncated to 16 chars
|
||||
|
||||
def test_ends_with_filename(self, storage):
|
||||
key = storage.get_object_key("myremote", "some/path/file.rpm")
|
||||
assert key.endswith("/file.rpm")
|
||||
|
||||
def test_same_path_produces_same_key(self, storage):
|
||||
k1 = storage.get_object_key("myremote", "some/path/file.rpm")
|
||||
k2 = storage.get_object_key("myremote", "some/path/file.rpm")
|
||||
assert k1 == k2
|
||||
def test_key_uses_sha256_of_directory_path(self, storage):
|
||||
# Pin the hash algorithm, truncation length, and format in one assertion
|
||||
key = storage.get_object_key("myremote", "some/path/to/file.rpm")
|
||||
expected_hash = hashlib.sha256(b"some/path/to").hexdigest()[:16]
|
||||
assert key == f"myremote/{expected_hash}/file.rpm"
|
||||
|
||||
def test_different_remotes_give_different_keys(self, storage):
|
||||
k1 = storage.get_object_key("remote-a", "path/to/file.rpm")
|
||||
@@ -48,7 +55,6 @@ class TestGetObjectKey:
|
||||
k1 = storage.get_object_key("myremote", "path/version-1/file.rpm")
|
||||
k2 = storage.get_object_key("myremote", "path/version-2/file.rpm")
|
||||
assert k1 != k2
|
||||
# Same filename, different directory hashes
|
||||
assert k1.split("/")[-1] == k2.split("/")[-1] == "file.rpm"
|
||||
|
||||
def test_leading_slash_stripped(self, storage):
|
||||
@@ -61,25 +67,25 @@ class TestGetObjectKey:
|
||||
assert key == "myremote/file.rpm"
|
||||
|
||||
def test_docker_blob_uses_digest_path(self, storage):
|
||||
digest = "abc123def456" * 4
|
||||
digest = "a" * 64 # realistic 64-char SHA-256 hex string
|
||||
path = f"library/nginx/blobs/sha256:{digest}"
|
||||
key = storage.get_object_key("dockerhub", path)
|
||||
assert key == f"dockerhub/blobs/sha256/{digest}"
|
||||
|
||||
def test_docker_blob_deduplication_across_images(self, storage):
|
||||
"""Same blob digest pulled from different images maps to the same S3 key."""
|
||||
digest = "deadbeef" * 8
|
||||
digest = "deadbeef" * 8 # 64-char hex
|
||||
k1 = storage.get_object_key("dockerhub", f"library/nginx/blobs/sha256:{digest}")
|
||||
k2 = storage.get_object_key("dockerhub", f"library/ubuntu/blobs/sha256:{digest}")
|
||||
assert k1 == k2
|
||||
|
||||
def test_docker_blob_different_digests_different_keys(self, storage):
|
||||
k1 = storage.get_object_key("dockerhub", "library/nginx/blobs/sha256:aaa111")
|
||||
k2 = storage.get_object_key("dockerhub", "library/nginx/blobs/sha256:bbb222")
|
||||
k1 = storage.get_object_key("dockerhub", "library/nginx/blobs/sha256:" + "a" * 64)
|
||||
k2 = storage.get_object_key("dockerhub", "library/nginx/blobs/sha256:" + "b" * 64)
|
||||
assert k1 != k2
|
||||
|
||||
def test_docker_blob_different_remotes_different_keys(self, storage):
|
||||
digest = "abc" * 20
|
||||
digest = "abc" * 21 + "d" # 64-char hex
|
||||
k1 = storage.get_object_key("remote-a", f"library/nginx/blobs/sha256:{digest}")
|
||||
k2 = storage.get_object_key("remote-b", f"library/nginx/blobs/sha256:{digest}")
|
||||
assert k1 != k2
|
||||
@@ -89,13 +95,38 @@ class TestGetObjectKey:
|
||||
# get_url
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestGetUrl:
|
||||
def test_returns_http_url_for_insecure_endpoint(self, storage):
|
||||
url = storage.get_url("myremote/abc123/file.rpm")
|
||||
assert url == "http://localhost:9000/testbucket/myremote/abc123/file.rpm"
|
||||
|
||||
def test_url_contains_bucket_and_key(self, storage):
|
||||
key = "myremote/abc/file.tar.gz"
|
||||
url = storage.get_url(key)
|
||||
assert "testbucket" in url
|
||||
assert key in url
|
||||
def test_returns_http_url_for_secure_storage(self):
|
||||
with patch("boto3.client", return_value=MagicMock()):
|
||||
s = S3Storage(endpoint="s3.example.com", access_key="k", secret_key="s", bucket="b", secure=True)
|
||||
s.client = MagicMock()
|
||||
# get_url uses http:// always (direct internal access address, not the S3 protocol)
|
||||
assert s.get_url("path/to/file.rpm") == "http://s3.example.com/b/path/to/file.rpm"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# upload / download_object
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestUpload:
|
||||
def test_upload_returns_s3_uri(self, storage):
|
||||
storage.client.put_object.return_value = {}
|
||||
result = storage.upload("myremote/abc123/file.rpm", b"content")
|
||||
assert result == "s3://testbucket/myremote/abc123/file.rpm"
|
||||
|
||||
|
||||
class TestDownloadObject:
|
||||
def test_download_object_raises_404_on_client_error(self, storage):
|
||||
storage.client.get_object.side_effect = ClientError(
|
||||
{"Error": {"Code": "NoSuchKey", "Message": "The specified key does not exist"}},
|
||||
"GetObject",
|
||||
)
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
storage.download_object("nonexistent/key")
|
||||
assert exc_info.value.status_code == 404
|
||||
|
||||
Reference in New Issue
Block a user