unkinben
a115904bbc
Tag manifests (e.g. library/nginx/manifests/latest) and their sha256-addressed counterparts were stored at separate S3 keys with no cross-reference, so a sha256 manifest request always missed cache even when the identical content had just been stored under the tag key. After serving any mutable (tag) manifest, compute the sha256 of the response body and write it under the digest key (manifests/sha256:<hex>) if absent. The next sha256-addressed pull hits cache immediately. Also adds a short-lived Redis distributed lock (SET NX EX 30) around upstream fetches so that concurrent pods racing for the same cold key poll storage for up to 5 s before issuing a duplicate upstream request, eliminating the thundering herd on deploy events. Includes unit tests for both the lock primitives (acquire/release, fail-open when Redis is unavailable) and the docker proxy behaviour (cross-link written on tag hit, not written for sha256 requests, lock acquired/released, poll path serves from cache without upstream fetch, fallback fetch when poll times out). Reviewed-on: #42
1190 lines
53 KiB
Python
1190 lines
53 KiB
Python
"""FastAPI route tests using TestClient with mocked service dependencies."""
|
|
|
|
import hashlib
|
|
import json
|
|
from datetime import UTC
|
|
from unittest.mock import ANY, AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Per-test service mocks (replace module-level globals in main.py)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_storage():
|
|
m = MagicMock()
|
|
m.get_object_key.return_value = "test-remote/abc123/file.ext"
|
|
m.exists.return_value = False
|
|
m.download_object.return_value = b"fake content"
|
|
m.bucket = "testbucket"
|
|
m.client = MagicMock()
|
|
return m
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_cache():
|
|
m = MagicMock()
|
|
m.is_mutable_file.return_value = False
|
|
m.is_index_valid.return_value = True
|
|
m.available = False
|
|
m.client = None
|
|
return m
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_database():
|
|
m = MagicMock()
|
|
m.available = False
|
|
return m
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_metrics():
|
|
return MagicMock()
|
|
|
|
|
|
@pytest.fixture
|
|
def patched_deps(mock_storage, mock_cache, mock_database, mock_metrics):
|
|
"""Swap the module-level service instances in main.py for the duration of a test."""
|
|
import artifactapi.main as main_mod
|
|
|
|
with (
|
|
patch.object(main_mod, "storage", mock_storage),
|
|
patch.object(main_mod, "cache", mock_cache),
|
|
patch.object(main_mod, "database", mock_database),
|
|
patch.object(main_mod, "metrics", mock_metrics),
|
|
):
|
|
yield {
|
|
"storage": mock_storage,
|
|
"cache": mock_cache,
|
|
"database": mock_database,
|
|
"metrics": mock_metrics,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Basic / health endpoints
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestBasicEndpoints:
|
|
def test_root_returns_remote_list(self, client):
|
|
response = client.get("/")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "remotes" in data
|
|
assert isinstance(data["remotes"], list)
|
|
assert len(data["remotes"]) > 0
|
|
|
|
def test_root_contains_version(self, client):
|
|
response = client.get("/")
|
|
assert "version" in response.json()
|
|
|
|
def test_health_check(self, client):
|
|
response = client.get("/health")
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "healthy"
|
|
|
|
def test_docker_v2_ping(self, client):
|
|
response = client.get("/v2/")
|
|
assert response.status_code == 200
|
|
assert response.headers.get("Docker-Distribution-Api-Version") == "registry/2.0"
|
|
assert response.json() == {}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Docker proxy /v2/{remote}/{path}
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDockerProxy:
|
|
def test_unknown_remote_returns_404(self, client, patched_deps):
|
|
response = client.get("/v2/no-such-remote/library/nginx/manifests/latest")
|
|
assert response.status_code == 404
|
|
|
|
def test_non_docker_package_returns_400(self, client, patched_deps):
|
|
# alpine-test is package: alpine, not docker
|
|
response = client.get("/v2/alpine-test/library/nginx/manifests/latest")
|
|
assert response.status_code == 400
|
|
|
|
def test_pattern_blocked_returns_403(self, client, patched_deps):
|
|
# docker-restricted allows only "library/nginx"
|
|
response = client.get("/v2/docker-restricted/library/ubuntu/manifests/latest")
|
|
assert response.status_code == 403
|
|
|
|
def test_allowed_pattern_proceeds_to_cache(self, client, patched_deps):
|
|
deps = patched_deps
|
|
manifest = json.dumps(
|
|
{
|
|
"mediaType": "application/vnd.oci.image.manifest.v1+json",
|
|
"layers": [],
|
|
}
|
|
).encode()
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = manifest
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
response = client.get("/v2/docker-restricted/library/nginx/manifests/latest")
|
|
assert response.status_code == 200
|
|
|
|
def test_cache_hit_manifest_returns_correct_content_type(self, client, patched_deps):
|
|
deps = patched_deps
|
|
manifest = json.dumps(
|
|
{
|
|
"mediaType": "application/vnd.docker.distribution.manifest.v2+json",
|
|
"schemaVersion": 2,
|
|
"layers": [],
|
|
}
|
|
).encode()
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = manifest
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
response = client.get("/v2/docker-test/library/nginx/manifests/latest")
|
|
assert response.status_code == 200
|
|
ct = response.headers["content-type"]
|
|
assert ct.startswith("application/vnd.docker.distribution.manifest.v2+json")
|
|
|
|
def test_cache_hit_sets_docker_content_digest_header(self, client, patched_deps):
|
|
deps = patched_deps
|
|
manifest = json.dumps(
|
|
{
|
|
"mediaType": "application/vnd.oci.image.manifest.v1+json",
|
|
"layers": [],
|
|
}
|
|
).encode()
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = manifest
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
response = client.get("/v2/docker-test/library/nginx/manifests/latest")
|
|
expected = f"sha256:{hashlib.sha256(manifest).hexdigest()}"
|
|
assert response.headers["Docker-Content-Digest"] == expected
|
|
|
|
def test_cache_hit_records_metrics(self, client, patched_deps):
|
|
deps = patched_deps
|
|
manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = manifest
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
client.get("/v2/docker-test/library/nginx/manifests/latest")
|
|
deps["metrics"].record_cache_hit.assert_called_once_with("docker-test", ANY)
|
|
|
|
def test_head_request_returns_no_body(self, client, patched_deps):
|
|
deps = patched_deps
|
|
manifest = json.dumps(
|
|
{
|
|
"mediaType": "application/vnd.oci.image.manifest.v1+json",
|
|
"layers": [],
|
|
}
|
|
).encode()
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = manifest
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
response = client.head("/v2/docker-test/library/nginx/manifests/latest")
|
|
assert response.status_code == 200
|
|
assert response.content == b""
|
|
|
|
def test_cache_miss_calls_upstream_fetch(self, client, patched_deps):
|
|
deps = patched_deps
|
|
manifest = json.dumps(
|
|
{
|
|
"mediaType": "application/vnd.oci.image.manifest.v1+json",
|
|
"layers": [],
|
|
}
|
|
).encode()
|
|
deps["storage"].exists.return_value = False
|
|
deps["storage"].download_object.return_value = manifest
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
|
|
with patch(
|
|
"artifactapi.artifact.proxy.cache_single_artifact",
|
|
new_callable=AsyncMock,
|
|
return_value={"status": "cached"},
|
|
) as mock_fetch:
|
|
response = client.get("/v2/docker-test/library/nginx/manifests/latest")
|
|
|
|
mock_fetch.assert_called_once()
|
|
assert response.status_code == 200
|
|
|
|
def test_cache_miss_on_index_marks_index_cached(self, client, patched_deps):
|
|
deps = patched_deps
|
|
manifest = json.dumps(
|
|
{
|
|
"mediaType": "application/vnd.oci.image.manifest.v1+json",
|
|
"layers": [],
|
|
}
|
|
).encode()
|
|
deps["storage"].exists.return_value = False
|
|
deps["storage"].download_object.return_value = manifest
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
|
|
with patch(
|
|
"artifactapi.artifact.proxy.cache_single_artifact",
|
|
new_callable=AsyncMock,
|
|
return_value={"status": "cached"},
|
|
):
|
|
response = client.get("/v2/docker-test/library/nginx/manifests/latest")
|
|
|
|
assert response.status_code == 200
|
|
deps["cache"].mark_index_cached.assert_called_once()
|
|
|
|
def test_index_expired_triggers_refetch(self, client, patched_deps):
|
|
deps = patched_deps
|
|
manifest = json.dumps(
|
|
{
|
|
"mediaType": "application/vnd.oci.image.manifest.v1+json",
|
|
"layers": [],
|
|
}
|
|
).encode()
|
|
deps["storage"].exists.return_value = True # cached in S3
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = False # but TTL expired
|
|
deps["storage"].download_object.return_value = manifest
|
|
|
|
with patch("artifactapi.artifact.proxy._upstream_reachable", new_callable=AsyncMock, return_value=True):
|
|
with patch(
|
|
"artifactapi.artifact.proxy.cache_single_artifact",
|
|
new_callable=AsyncMock,
|
|
return_value={"status": "cached"},
|
|
) as mock_fetch:
|
|
response = client.get("/v2/docker-test/library/nginx/manifests/latest")
|
|
|
|
mock_fetch.assert_called_once()
|
|
assert response.status_code == 200
|
|
|
|
# --- Issue 1: sha256 digest cross-linking ---
|
|
|
|
def test_tag_manifest_is_stored_under_digest_key_on_cache_hit(self, client, patched_deps):
|
|
# When serving a cached tag manifest the handler must also write the content
|
|
# under the sha256 digest key so subsequent sha256-addressed pulls hit cache.
|
|
deps = patched_deps
|
|
manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
|
|
# First exists call (tag manifest): hit. Second (digest key): miss → triggers upload.
|
|
deps["storage"].exists.side_effect = [True, False]
|
|
deps["storage"].download_object.return_value = manifest
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
response = client.get("/v2/docker-test/library/nginx/manifests/v1.25.3")
|
|
|
|
assert response.status_code == 200
|
|
deps["storage"].upload.assert_called_once_with(deps["storage"].get_object_key.return_value, manifest)
|
|
|
|
def test_tag_manifest_digest_key_not_written_when_already_exists(self, client, patched_deps):
|
|
# When the digest key already exists in storage upload must not be called.
|
|
deps = patched_deps
|
|
manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
|
|
# Both the tag key and the digest key already present.
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = manifest
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
client.get("/v2/docker-test/library/nginx/manifests/v1.25.3")
|
|
|
|
deps["storage"].upload.assert_not_called()
|
|
|
|
def test_sha256_manifest_request_is_not_cross_linked(self, client, patched_deps):
|
|
# sha256-addressed manifests are immutable — the cross-link logic must not apply.
|
|
deps = patched_deps
|
|
manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = manifest
|
|
deps["cache"].is_mutable_file.return_value = False # sha256 manifest is immutable
|
|
|
|
with patch("artifactapi.artifact.proxy._fetch_last_modified", new_callable=AsyncMock, return_value=None):
|
|
client.get("/v2/docker-test/library/nginx/manifests/sha256:" + "a" * 64)
|
|
|
|
deps["storage"].upload.assert_not_called()
|
|
|
|
# --- Issue 2: thundering herd distributed lock ---
|
|
|
|
def test_lock_acquired_and_released_on_upstream_fetch(self, client, patched_deps):
|
|
deps = patched_deps
|
|
manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
|
|
deps["storage"].exists.side_effect = [False, False] # initial miss; digest key also absent
|
|
deps["storage"].download_object.return_value = manifest
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].acquire_fetch_lock.return_value = True
|
|
|
|
with patch(
|
|
"artifactapi.artifact.proxy.cache_single_artifact",
|
|
new_callable=AsyncMock,
|
|
return_value={"status": "cached"},
|
|
):
|
|
response = client.get("/v2/docker-test/library/nginx/manifests/latest")
|
|
|
|
deps["cache"].acquire_fetch_lock.assert_called_once()
|
|
deps["cache"].release_fetch_lock.assert_called_once()
|
|
assert response.status_code == 200
|
|
|
|
def test_lock_released_even_when_fetch_returns_error(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = False
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].acquire_fetch_lock.return_value = True
|
|
|
|
with patch(
|
|
"artifactapi.artifact.proxy.cache_single_artifact",
|
|
new_callable=AsyncMock,
|
|
return_value={"status": "error", "error": "upstream down"},
|
|
):
|
|
response = client.get("/v2/docker-test/library/nginx/manifests/latest")
|
|
|
|
deps["cache"].release_fetch_lock.assert_called_once()
|
|
assert response.status_code == 502
|
|
|
|
def test_thundering_herd_polls_storage_when_lock_not_acquired(self, client, patched_deps):
|
|
# When the lock is held by another pod the handler must poll storage and serve
|
|
# from cache once the competing fetch completes, without issuing its own upstream request.
|
|
deps = patched_deps
|
|
manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
|
|
# Initial cache check: miss. First poll iteration: another pod has written it.
|
|
# Third call is for the digest cross-link check (is_mutable=True path); digest key exists.
|
|
deps["storage"].exists.side_effect = [False, True, True]
|
|
deps["storage"].download_object.return_value = manifest
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
deps["cache"].acquire_fetch_lock.return_value = False # lock held by peer
|
|
|
|
with patch("artifactapi.artifact.docker.asyncio.sleep", new_callable=AsyncMock):
|
|
with patch(
|
|
"artifactapi.artifact.proxy.cache_single_artifact",
|
|
new_callable=AsyncMock,
|
|
) as mock_fetch:
|
|
response = client.get("/v2/docker-test/library/nginx/manifests/latest")
|
|
|
|
mock_fetch.assert_not_called()
|
|
assert response.status_code == 200
|
|
|
|
def test_thundering_herd_falls_through_to_fetch_if_poll_times_out(self, client, patched_deps):
|
|
# If the item never appears in storage during the poll window the handler must
|
|
# still issue its own upstream fetch as a fallback.
|
|
deps = patched_deps
|
|
manifest = json.dumps({"mediaType": "application/vnd.oci.image.manifest.v1+json", "layers": []}).encode()
|
|
# All exists calls return False — item never appears during polling.
|
|
deps["storage"].exists.return_value = False
|
|
deps["storage"].download_object.return_value = manifest
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].acquire_fetch_lock.return_value = False # lock held by peer
|
|
|
|
with patch("artifactapi.artifact.docker.asyncio.sleep", new_callable=AsyncMock):
|
|
with patch(
|
|
"artifactapi.artifact.proxy.cache_single_artifact",
|
|
new_callable=AsyncMock,
|
|
return_value={"status": "cached"},
|
|
) as mock_fetch:
|
|
response = client.get("/v2/docker-test/library/nginx/manifests/latest")
|
|
|
|
mock_fetch.assert_called_once()
|
|
assert response.status_code == 200
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Generic artifact route /api/v1/remote/{remote}/{path}
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGenericArtifactRoute:
|
|
def test_unknown_remote_returns_404(self, client, patched_deps):
|
|
response = client.get("/api/v1/remote/nonexistent/path/to/file.tar.gz")
|
|
assert response.status_code == 404
|
|
|
|
def test_pattern_blocked_returns_403(self, client, patched_deps):
|
|
# generic-test only allows .tar.gz
|
|
response = client.get("/api/v1/remote/generic-test/some/path/file.rpm")
|
|
assert response.status_code == 403
|
|
|
|
def test_cache_hit_returns_200_with_source_header(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"tar content"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
response = client.get("/api/v1/remote/generic-test/some/path/archive.tar.gz")
|
|
assert response.status_code == 200
|
|
assert response.headers["X-Artifact-Source"] == "cache"
|
|
assert response.content == b"tar content"
|
|
|
|
def test_cache_hit_sets_content_disposition(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"content"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
response = client.get("/api/v1/remote/generic-test/some/path/archive.tar.gz")
|
|
disposition = response.headers["content-disposition"]
|
|
assert "attachment" in disposition
|
|
assert "archive.tar.gz" in disposition
|
|
|
|
def test_cache_hit_sets_artifact_size_header(self, client, patched_deps):
|
|
deps = patched_deps
|
|
content = b"some artifact content bytes"
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = content
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
response = client.get("/api/v1/remote/generic-test/some/path/archive.tar.gz")
|
|
assert response.headers["X-Artifact-Size"] == str(len(content))
|
|
|
|
def test_cache_hit_records_metrics(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"content"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
client.get("/api/v1/remote/generic-test/some/path/archive.tar.gz")
|
|
deps["metrics"].record_cache_hit.assert_called_once_with("generic-test", ANY)
|
|
|
|
def test_cache_hit_records_artifact_mapping(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"content"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
client.get("/api/v1/remote/generic-test/some/path/archive.tar.gz")
|
|
deps["database"].record_artifact_mapping.assert_called_once()
|
|
|
|
def test_cache_hit_rpm_returns_correct_content_type(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"rpm bytes"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
response = client.get("/api/v1/remote/rpm-test/almalinux/9/x86_64/bash-5.1.8.x86_64.rpm")
|
|
assert response.status_code == 200
|
|
assert "application/x-rpm" in response.headers["content-type"]
|
|
|
|
def test_cache_hit_xml_returns_correct_content_type(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"<?xml version='1.0'?>"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
response = client.get("/api/v1/remote/rpm-test/repo/repodata/primary.xml")
|
|
assert response.status_code == 200
|
|
assert "application/xml" in response.headers["content-type"]
|
|
|
|
def test_cache_miss_fetches_upstream_and_returns_200(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = False
|
|
deps["storage"].download_object.return_value = b"fresh content"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
with patch(
|
|
"artifactapi.artifact.proxy.cache_single_artifact",
|
|
new_callable=AsyncMock,
|
|
return_value={"status": "cached"},
|
|
) as mock_fetch:
|
|
response = client.get("/api/v1/remote/generic-test/some/path/archive.tar.gz")
|
|
|
|
mock_fetch.assert_called_once()
|
|
assert response.status_code == 200
|
|
assert response.headers["X-Artifact-Source"] == "remote"
|
|
|
|
def test_cache_miss_records_metrics(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = False
|
|
deps["storage"].download_object.return_value = b"fresh content"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
with patch(
|
|
"artifactapi.artifact.proxy.cache_single_artifact",
|
|
new_callable=AsyncMock,
|
|
return_value={"status": "cached"},
|
|
):
|
|
client.get("/api/v1/remote/generic-test/some/path/archive.tar.gz")
|
|
|
|
deps["metrics"].record_cache_miss.assert_called_once_with("generic-test", ANY)
|
|
|
|
def test_cache_miss_on_index_marks_index_cached(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = False
|
|
deps["storage"].download_object.return_value = b"APKINDEX content"
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
|
|
with patch(
|
|
"artifactapi.artifact.proxy.cache_single_artifact",
|
|
new_callable=AsyncMock,
|
|
return_value={"status": "cached"},
|
|
):
|
|
response = client.get("/api/v1/remote/alpine-test/alpine/v3.18/x86_64/APKINDEX.tar.gz")
|
|
|
|
assert response.status_code == 200
|
|
deps["cache"].mark_index_cached.assert_called_once()
|
|
|
|
def test_upstream_error_returns_502(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = False
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
with patch(
|
|
"artifactapi.artifact.proxy.cache_single_artifact",
|
|
new_callable=AsyncMock,
|
|
return_value={"status": "error", "error": "upstream unreachable"},
|
|
):
|
|
response = client.get("/api/v1/remote/generic-test/some/path/archive.tar.gz")
|
|
|
|
assert response.status_code == 502
|
|
|
|
def test_mutable_file_bypasses_immutable_patterns(self, client, patched_deps):
|
|
"""Mutable files must be served even when they don't match immutable_patterns."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"APKINDEX content"
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
# APKINDEX.tar.gz does not match alpine-test's immutable_patterns (.*.apk$),
|
|
# but since is_mutable_file returns True it must be allowed through.
|
|
response = client.get("/api/v1/remote/alpine-test/alpine/v3.18/x86_64/APKINDEX.tar.gz")
|
|
assert response.status_code == 200
|
|
|
|
def test_mutable_unchanged_refreshes_ttl_without_redownload(self, client, patched_deps):
|
|
"""When check_mutable_updates=True and upstream says 304, TTL is refreshed in place."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"metadata content"
|
|
# File is mutable and its TTL has expired
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = False
|
|
deps["cache"].get_mutable_meta.return_value = {"etag": '"abc"'}
|
|
|
|
with patch("artifactapi.artifact.proxy.check_upstream_changed", new_callable=AsyncMock, return_value=False):
|
|
response = client.get("/api/v1/remote/check-mutable-test/metadata.json")
|
|
|
|
assert response.status_code == 200
|
|
deps["cache"].mark_index_cached.assert_called()
|
|
# S3 object must NOT have been deleted (no re-download)
|
|
deps["storage"].client.delete_object.assert_not_called()
|
|
|
|
def test_mutable_changed_triggers_redownload(self, client, patched_deps):
|
|
"""When check_mutable_updates=True and upstream says 200, cache is invalidated."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = False
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = False
|
|
deps["cache"].get_mutable_meta.return_value = {"etag": '"abc"'}
|
|
|
|
with patch("artifactapi.artifact.proxy.check_upstream_changed", new_callable=AsyncMock, return_value=True):
|
|
with patch("artifactapi.artifact.proxy.cache_single_artifact", new_callable=AsyncMock) as mock_cache:
|
|
mock_cache.return_value = {"status": "error", "error": "upstream gone"}
|
|
response = client.get("/api/v1/remote/check-mutable-test/metadata.json")
|
|
|
|
assert response.status_code == 502
|
|
|
|
def test_mutable_changed_redownloads_successfully(self, client, patched_deps):
|
|
"""When check_mutable_updates=True and upstream says 200, fresh copy is fetched and served."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"fresh metadata"
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = False
|
|
deps["cache"].get_mutable_meta.return_value = {"etag": '"abc"'}
|
|
|
|
with patch("artifactapi.artifact.proxy.check_upstream_changed", new_callable=AsyncMock, return_value=True):
|
|
with patch("artifactapi.artifact.proxy.cache_single_artifact", new_callable=AsyncMock) as mock_cache:
|
|
mock_cache.return_value = {"status": "cached", "etag": '"def"', "last_modified": None}
|
|
response = client.get("/api/v1/remote/check-mutable-test/metadata.json")
|
|
|
|
assert response.status_code == 200
|
|
mock_cache.assert_called_once()
|
|
|
|
def test_mutable_backend_unreachable_on_check_updates_keeps_stale(self, client, patched_deps):
|
|
"""When check_mutable_updates=True and backend is unreachable, stale copy is kept and TTL refreshed."""
|
|
from artifactapi.artifact.proxy import UpstreamUnreachable
|
|
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"stale metadata"
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = False
|
|
deps["cache"].get_mutable_meta.return_value = {"etag": '"abc"'}
|
|
|
|
with patch("artifactapi.artifact.proxy.check_upstream_changed", side_effect=UpstreamUnreachable("connection refused")):
|
|
response = client.get("/api/v1/remote/check-mutable-test/metadata.json")
|
|
|
|
assert response.status_code == 200
|
|
deps["cache"].mark_index_cached.assert_called()
|
|
deps["storage"].client.delete_object.assert_not_called()
|
|
|
|
def test_mutable_backend_unreachable_on_expiry_keeps_stale(self, client, patched_deps):
|
|
"""When a regular mutable file expires and backend is unreachable, stale copy is kept and TTL refreshed."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"stale APKINDEX"
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = False
|
|
|
|
with patch("artifactapi.artifact.proxy._upstream_reachable", new_callable=AsyncMock, return_value=False):
|
|
response = client.get("/api/v1/remote/alpine-test/alpine/v3.18/x86_64/APKINDEX.tar.gz")
|
|
|
|
assert response.status_code == 200
|
|
deps["cache"].mark_index_cached.assert_called()
|
|
deps["storage"].client.delete_object.assert_not_called()
|
|
|
|
def test_mutable_flag_off_skips_conditional_check(self, client, patched_deps):
|
|
"""When check_mutable_updates is not set, expired mutable files are always re-fetched."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = False
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = False
|
|
|
|
with patch("artifactapi.artifact.proxy.check_upstream_changed", new_callable=AsyncMock) as mock_check:
|
|
with patch("artifactapi.artifact.proxy.cache_single_artifact", new_callable=AsyncMock) as mock_cache:
|
|
mock_cache.return_value = {"status": "error", "error": "upstream gone"}
|
|
client.get("/api/v1/remote/custom-index-test/metadata.json")
|
|
|
|
mock_check.assert_not_called()
|
|
|
|
def test_local_repo_file_not_found_returns_404(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["database"].get_local_file_metadata.return_value = None
|
|
deps["database"].available = True
|
|
|
|
response = client.get("/api/v1/local/local-test/path/to/nonexistent.bin")
|
|
assert response.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Upload route PUT /api/v1/local/{local}/{path}
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestUploadRoute:
|
|
def test_unknown_local_returns_404(self, client, patched_deps):
|
|
response = client.put(
|
|
"/api/v1/local/nonexistent/path/to/file.tar.gz",
|
|
files={"file": ("file.tar.gz", b"content", "application/octet-stream")},
|
|
)
|
|
assert response.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# HEAD route HEAD /api/v1/local/{local}/{path}
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestHeadRoute:
|
|
def test_local_repo_file_not_found_returns_404(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["database"].get_local_file_metadata.return_value = None
|
|
deps["database"].available = True
|
|
|
|
response = client.head("/api/v1/local/local-test/path/to/nonexistent.bin")
|
|
assert response.status_code == 404
|
|
|
|
def test_unknown_local_returns_404(self, client, patched_deps):
|
|
response = client.head("/api/v1/local/nonexistent/path/to/file.bin")
|
|
assert response.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DELETE route DELETE /api/v1/local/{local}/{path}
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDeleteRoute:
|
|
def test_unknown_local_returns_404(self, client, patched_deps):
|
|
response = client.delete("/api/v1/local/nonexistent/path/to/file.tar.gz")
|
|
assert response.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cache flush PUT /cache/flush
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCacheFlushEndpoint:
|
|
def test_flush_all_returns_flushed_structure(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["cache"].available = False
|
|
deps["storage"].client.list_objects_v2.return_value = {}
|
|
|
|
response = client.put("/cache/flush")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "flushed" in data
|
|
assert "redis_keys" in data["flushed"]
|
|
assert "s3_objects" in data["flushed"]
|
|
|
|
def test_flush_specific_remote_echoes_remote(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["cache"].available = False
|
|
deps["storage"].client.list_objects_v2.return_value = {}
|
|
|
|
response = client.put("/cache/flush?remote=alpine-test")
|
|
assert response.status_code == 200
|
|
assert response.json()["remote"] == "alpine-test"
|
|
|
|
def test_flush_all_deletes_redis_keys_when_cache_available(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["cache"].available = True
|
|
redis_mock = MagicMock()
|
|
deps["cache"].client = redis_mock
|
|
# index:* returns keys; mutable:meta:* and metrics:* return nothing
|
|
redis_mock.keys.side_effect = [["index:test:abc", "index:test:def"], [], []]
|
|
deps["storage"].client.list_objects_v2.return_value = {}
|
|
|
|
response = client.put("/cache/flush")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["flushed"]["redis_keys"] == 2
|
|
redis_mock.delete.assert_called_once_with("index:test:abc", "index:test:def")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Metrics endpoint GET /metrics
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMetricsEndpoint:
|
|
def test_returns_prometheus_text_by_default(self, client, patched_deps):
|
|
response = client.get("/metrics")
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"].startswith("text/plain")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config endpoint GET /config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestConfigEndpoint:
|
|
def test_returns_config_with_remotes(self, client):
|
|
response = client.get("/config")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "remotes" in data
|
|
assert "alpine-test" in data["remotes"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PyPI remote /api/v1/remote/pypi-test/...
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestPyPIRemote:
|
|
def test_simple_index_is_mutable(self, client, patched_deps):
|
|
"""simple/ paths are detected as mutable (package-type default)."""
|
|
deps = patched_deps
|
|
html = b"<html><body><a href='https://files.pythonhosted.org/packages/requests-2.31.0.tar.gz'>...</a></body></html>"
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = html
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
response = client.get("/api/v1/remote/pypi-test/simple/requests/")
|
|
assert response.status_code == 200
|
|
deps["cache"].mark_index_cached.assert_not_called()
|
|
|
|
def test_simple_index_urls_rewritten_to_proxy(self, client, patched_deps):
|
|
"""files.pythonhosted.org URLs in a cached simple index are rewritten to our proxy."""
|
|
deps = patched_deps
|
|
html = b"<html><body><a href='https://files.pythonhosted.org/packages/requests-2.31.0.tar.gz'>...</a></body></html>"
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = html
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
response = client.get("/api/v1/remote/pypi-test/simple/requests/")
|
|
assert response.status_code == 200
|
|
assert b"files.pythonhosted.org" not in response.content
|
|
assert b"/api/v1/remote/pypi-test/packages/requests-2.31.0.tar.gz" in response.content
|
|
|
|
def test_simple_index_content_type_is_html(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"<html></html>"
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
response = client.get("/api/v1/remote/pypi-test/simple/requests/")
|
|
assert response.status_code == 200
|
|
assert "text/html" in response.headers["content-type"]
|
|
|
|
def test_simple_index_cache_miss_fetches_upstream(self, client, patched_deps):
|
|
deps = patched_deps
|
|
html = b"<html><body><a href='https://files.pythonhosted.org/packages/p-1.0.whl'>...</a></body></html>"
|
|
deps["storage"].exists.return_value = False
|
|
deps["storage"].download_object.return_value = html
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
|
|
with patch(
|
|
"artifactapi.artifact.proxy.cache_single_artifact",
|
|
new_callable=AsyncMock,
|
|
return_value={"status": "cached"},
|
|
) as mock_fetch:
|
|
response = client.get("/api/v1/remote/pypi-test/simple/requests/")
|
|
|
|
mock_fetch.assert_called_once()
|
|
assert response.status_code == 200
|
|
assert b"files.pythonhosted.org" not in response.content
|
|
|
|
def test_wheel_file_immutable_returns_correct_content_type(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"PK wheel bytes"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
response = client.get("/api/v1/remote/pypi-test/packages/requests-2.31.0-py3-none-any.whl")
|
|
assert response.status_code == 200
|
|
assert "application/zip" in response.headers["content-type"]
|
|
assert response.headers["X-Artifact-Source"] == "cache"
|
|
|
|
def test_sdist_immutable_returns_correct_content_type(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"tar bytes"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
response = client.get("/api/v1/remote/pypi-test/packages/requests-2.31.0.tar.gz")
|
|
assert response.status_code == 200
|
|
assert "application/gzip" in response.headers["content-type"]
|
|
|
|
def test_unknown_extension_on_pypi_remote_returns_403(self, client, patched_deps):
|
|
"""Paths that don't match immutable_patterns and aren't mutable are blocked."""
|
|
response = client.get("/api/v1/remote/pypi-test/packages/requests.unknown")
|
|
assert response.status_code == 403
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# npm remote /api/v1/remote/npm-test/...
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestNpmRemote:
|
|
def test_package_metadata_is_mutable(self, client, patched_deps):
|
|
"""Top-level package metadata paths are detected as mutable."""
|
|
deps = patched_deps
|
|
meta = b'{"name":"express","versions":{}}'
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = meta
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
response = client.get("/api/v1/remote/npm-test/express")
|
|
assert response.status_code == 200
|
|
deps["cache"].mark_index_cached.assert_not_called()
|
|
|
|
def test_metadata_tarball_urls_rewritten_to_proxy(self, client, patched_deps):
|
|
"""registry.npmjs.org tarball URLs in metadata JSON are rewritten to our proxy."""
|
|
deps = patched_deps
|
|
meta = b'{"dist":{"tarball":"https://registry.npmjs.org/express/-/express-4.18.2.tgz"}}'
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = meta
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
response = client.get("/api/v1/remote/npm-test/express")
|
|
assert response.status_code == 200
|
|
assert b"registry.npmjs.org" not in response.content
|
|
assert b"/api/v1/remote/npm-test/express/-/express-4.18.2.tgz" in response.content
|
|
|
|
def test_metadata_content_type_is_json(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b'{"name":"express"}'
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
response = client.get("/api/v1/remote/npm-test/express")
|
|
assert response.status_code == 200
|
|
assert "application/json" in response.headers["content-type"]
|
|
|
|
def test_scoped_package_metadata_rewritten(self, client, patched_deps):
|
|
"""@scope/package metadata URLs are also rewritten back to the same npm-test remote."""
|
|
deps = patched_deps
|
|
meta = b'{"dist":{"tarball":"https://registry.npmjs.org/@babel/core/-/core-7.21.0.tgz"}}'
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = meta
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
response = client.get("/api/v1/remote/npm-test/@babel/core")
|
|
assert response.status_code == 200
|
|
assert b"registry.npmjs.org" not in response.content
|
|
assert b"/api/v1/remote/npm-test/@babel/core/-/core-7.21.0.tgz" in response.content
|
|
|
|
def test_tarball_not_rewritten(self, client, patched_deps):
|
|
"""Tarball requests (.tgz) bypass URL rewriting and return binary."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"\x1f\x8b tgz bytes"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
response = client.get("/api/v1/remote/npm-test/express/-/express-4.18.2.tgz")
|
|
assert response.status_code == 200
|
|
assert "application/gzip" in response.headers["content-type"]
|
|
assert response.headers["X-Artifact-Source"] == "cache"
|
|
|
|
def test_metadata_cache_miss_fetches_upstream(self, client, patched_deps):
|
|
deps = patched_deps
|
|
meta = b'{"dist":{"tarball":"https://registry.npmjs.org/lodash/-/lodash-4.17.21.tgz"}}'
|
|
deps["storage"].exists.return_value = False
|
|
deps["storage"].download_object.return_value = meta
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
|
|
with patch(
|
|
"artifactapi.artifact.proxy.cache_single_artifact",
|
|
new_callable=AsyncMock,
|
|
return_value={"status": "cached"},
|
|
) as mock_fetch:
|
|
response = client.get("/api/v1/remote/npm-test/lodash")
|
|
|
|
mock_fetch.assert_called_once()
|
|
assert response.status_code == 200
|
|
assert b"registry.npmjs.org" not in response.content
|
|
|
|
def test_tarball_immutable_allowed_on_npm_remote(self, client, patched_deps):
|
|
"""Tarballs (.tgz) match immutable_patterns and are served without rewriting."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"tgz bytes"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
response = client.get("/api/v1/remote/npm-test/express/-/express-4.18.2.tgz")
|
|
assert response.status_code == 200
|
|
assert "application/gzip" in response.headers["content-type"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helm remote /api/v1/remote/helm-test/...
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestHelmRemote:
|
|
def test_index_yaml_is_mutable(self, client, patched_deps):
|
|
"""index.yaml is detected as mutable (package-type default)."""
|
|
deps = patched_deps
|
|
index = b"apiVersion: v1\nentries:\n vault:\n - urls:\n - https://helm.releases.hashicorp.com/vault-0.29.1.tgz\n"
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = index
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
response = client.get("/api/v1/remote/helm-test/index.yaml")
|
|
assert response.status_code == 200
|
|
deps["cache"].mark_index_cached.assert_not_called()
|
|
|
|
def test_index_yaml_urls_rewritten_to_proxy(self, client, patched_deps):
|
|
"""base_url chart URLs in a cached index.yaml are rewritten to our proxy."""
|
|
deps = patched_deps
|
|
index = b"apiVersion: v1\nentries:\n vault:\n - urls:\n - https://helm.releases.hashicorp.com/vault-0.29.1.tgz\n"
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = index
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
response = client.get("/api/v1/remote/helm-test/index.yaml")
|
|
assert response.status_code == 200
|
|
assert b"helm.releases.hashicorp.com" not in response.content
|
|
assert b"/api/v1/remote/helm-test/vault-0.29.1.tgz" in response.content
|
|
|
|
def test_index_yaml_content_type_is_yaml(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"apiVersion: v1\nentries: {}\n"
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
response = client.get("/api/v1/remote/helm-test/index.yaml")
|
|
assert response.status_code == 200
|
|
assert "text/yaml" in response.headers["content-type"]
|
|
|
|
def test_chart_tarball_immutable_returns_gzip_content_type(self, client, patched_deps):
|
|
"""Versioned chart tarballs match immutable_patterns and are served as binary."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"\x1f\x8b chart bytes"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
response = client.get("/api/v1/remote/helm-test/vault-0.29.1.tgz")
|
|
assert response.status_code == 200
|
|
assert "application/gzip" in response.headers["content-type"]
|
|
assert response.headers["X-Artifact-Source"] == "cache"
|
|
|
|
def test_index_yaml_cache_miss_fetches_upstream(self, client, patched_deps):
|
|
deps = patched_deps
|
|
index = b"apiVersion: v1\nentries:\n vault:\n - urls:\n - https://helm.releases.hashicorp.com/vault-0.29.1.tgz\n"
|
|
deps["storage"].exists.return_value = False
|
|
deps["storage"].download_object.return_value = index
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
|
|
with patch(
|
|
"artifactapi.artifact.proxy.cache_single_artifact",
|
|
new_callable=AsyncMock,
|
|
return_value={"status": "cached"},
|
|
) as mock_fetch:
|
|
response = client.get("/api/v1/remote/helm-test/index.yaml")
|
|
|
|
mock_fetch.assert_called_once()
|
|
assert response.status_code == 200
|
|
assert b"helm.releases.hashicorp.com" not in response.content
|
|
|
|
def test_non_tgz_non_yaml_path_blocked_by_pattern(self, client, patched_deps):
|
|
"""Paths that don't match immutable_patterns and aren't mutable are blocked."""
|
|
deps = patched_deps
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
response = client.get("/api/v1/remote/helm-test/vault.zip")
|
|
assert response.status_code == 403
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Quarantine (quarantine-test remote: quarantine_new=True, quarantine_days=3)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestQuarantine:
|
|
def _recent_date(self, days_ago=1):
|
|
"""Return an HTTP-format date string N days in the past (within quarantine window)."""
|
|
from datetime import datetime, timedelta
|
|
from email.utils import format_datetime
|
|
|
|
dt = datetime.now(UTC) - timedelta(days=days_ago)
|
|
return format_datetime(dt, usegmt=True)
|
|
|
|
def _old_date(self, days_ago=10):
|
|
"""Return an HTTP-format date string N days in the past (outside quarantine window)."""
|
|
from datetime import datetime, timedelta
|
|
from email.utils import format_datetime
|
|
|
|
dt = datetime.now(UTC) - timedelta(days=days_ago)
|
|
return format_datetime(dt, usegmt=True)
|
|
|
|
def test_cache_miss_recent_artifact_quarantined(self, client, patched_deps):
|
|
"""Cache miss: artifact published within quarantine window → 404."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = False
|
|
deps["storage"].download_object.return_value = b"content"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
with patch(
|
|
"artifactapi.artifact.proxy.cache_single_artifact",
|
|
new_callable=AsyncMock,
|
|
return_value={"status": "cached", "last_modified": self._recent_date()},
|
|
):
|
|
response = client.get("/api/v1/remote/quarantine-test/some/path/package-1.0.tar.gz")
|
|
|
|
assert response.status_code == 404
|
|
assert "quarantined" in response.json()["detail"].lower()
|
|
|
|
def test_cache_miss_old_artifact_allowed(self, client, patched_deps):
|
|
"""Cache miss: artifact published outside quarantine window → 200."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = False
|
|
deps["storage"].download_object.return_value = b"content"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
with patch(
|
|
"artifactapi.artifact.proxy.cache_single_artifact",
|
|
new_callable=AsyncMock,
|
|
return_value={"status": "cached", "last_modified": self._old_date()},
|
|
):
|
|
response = client.get("/api/v1/remote/quarantine-test/some/path/package-1.0.tar.gz")
|
|
|
|
assert response.status_code == 200
|
|
|
|
def test_cache_miss_no_last_modified_fails_open(self, client, patched_deps):
|
|
"""Cache miss: no Last-Modified header → fail open (200, not quarantined)."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = False
|
|
deps["storage"].download_object.return_value = b"content"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
with patch(
|
|
"artifactapi.artifact.proxy.cache_single_artifact",
|
|
new_callable=AsyncMock,
|
|
return_value={"status": "cached", "last_modified": None},
|
|
):
|
|
response = client.get("/api/v1/remote/quarantine-test/some/path/package-1.0.tar.gz")
|
|
|
|
assert response.status_code == 200
|
|
|
|
def test_cache_hit_recent_artifact_quarantined(self, client, patched_deps):
|
|
"""Cache hit: stored publish date within quarantine window → 404."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"content"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
deps["cache"].get_artifact_published.return_value = self._recent_date()
|
|
|
|
response = client.get("/api/v1/remote/quarantine-test/some/path/package-1.0.tar.gz")
|
|
|
|
assert response.status_code == 404
|
|
assert "quarantined" in response.json()["detail"].lower()
|
|
|
|
def test_cache_hit_old_artifact_allowed(self, client, patched_deps):
|
|
"""Cache hit: stored publish date outside quarantine window → 200."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"content"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
deps["cache"].get_artifact_published.return_value = self._old_date()
|
|
|
|
response = client.get("/api/v1/remote/quarantine-test/some/path/package-1.0.tar.gz")
|
|
|
|
assert response.status_code == 200
|
|
|
|
def test_cache_hit_no_stored_date_fetches_upstream(self, client, patched_deps):
|
|
"""Cache hit: no stored date → HEAD upstream to get Last-Modified."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"content"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
deps["cache"].get_artifact_published.return_value = None
|
|
|
|
with patch(
|
|
"artifactapi.artifact.proxy._fetch_last_modified",
|
|
new_callable=AsyncMock,
|
|
return_value=self._old_date(),
|
|
) as mock_fetch:
|
|
response = client.get("/api/v1/remote/quarantine-test/some/path/package-1.0.tar.gz")
|
|
|
|
mock_fetch.assert_called_once()
|
|
assert response.status_code == 200
|
|
|
|
def test_quarantine_disabled_allows_recent_artifact(self, client, patched_deps):
|
|
"""quarantine_new=False: recent artifacts are not blocked."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = False
|
|
deps["storage"].download_object.return_value = b"content"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
with patch(
|
|
"artifactapi.artifact.proxy.cache_single_artifact",
|
|
new_callable=AsyncMock,
|
|
return_value={"status": "cached", "last_modified": self._recent_date()},
|
|
):
|
|
response = client.get("/api/v1/remote/quarantine-disabled/some/path/package-1.0.tar.gz")
|
|
|
|
assert response.status_code == 200
|
|
|
|
def test_quarantine_detail_includes_available_date(self, client, patched_deps):
|
|
"""The 404 detail should include the date when the artifact becomes available."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = False
|
|
deps["storage"].download_object.return_value = b"content"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
with patch(
|
|
"artifactapi.artifact.proxy.cache_single_artifact",
|
|
new_callable=AsyncMock,
|
|
return_value={"status": "cached", "last_modified": self._recent_date()},
|
|
):
|
|
response = client.get("/api/v1/remote/quarantine-test/some/path/package-1.0.tar.gz")
|
|
|
|
assert response.status_code == 404
|
|
detail = response.json()["detail"]
|
|
assert "available after" in detail
|
|
assert "3-day" in detail
|