unkinben
af28631a41
Adds two per-remote config keys for docker remotes:
ban_tags_enabled: false # opt-in, default off
ban_tags:
- latest
- edge
When ban_tags_enabled is true and a manifest request arrives for a named
tag in ban_tags, the proxy returns 403. sha256-addressed pulls are never
blocked, so images already pulled can still be referenced by digest.
Blob requests are unaffected.
1141 lines
51 KiB
Python
1141 lines
51 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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Docker ban_tags feature
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDockerBanTags:
|
|
def test_banned_tag_returns_403(self, client, patched_deps):
|
|
response = client.get("/v2/docker-bantags-test/library/nginx/manifests/latest")
|
|
assert response.status_code == 403
|
|
assert "latest" in response.json()["detail"]
|
|
|
|
def test_second_banned_tag_returns_403(self, client, patched_deps):
|
|
response = client.get("/v2/docker-bantags-test/library/nginx/manifests/edge")
|
|
assert response.status_code == 403
|
|
assert "edge" in response.json()["detail"]
|
|
|
|
def test_allowed_tag_proceeds(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-bantags-test/library/nginx/manifests/1.25.3")
|
|
assert response.status_code == 200
|
|
|
|
def test_digest_pull_bypasses_ban(self, client, patched_deps):
|
|
# sha256-addressed pulls must never be blocked by the tag ban list
|
|
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
|
|
|
|
digest = "sha256:" + "a" * 64
|
|
with patch("artifactapi.artifact.proxy._fetch_last_modified", new_callable=AsyncMock, return_value=None):
|
|
response = client.get(f"/v2/docker-bantags-test/library/nginx/manifests/{digest}")
|
|
assert response.status_code == 200
|
|
|
|
def test_ban_tags_disabled_by_default(self, client, patched_deps):
|
|
# docker-test has no ban_tags_enabled — "latest" must pass through
|
|
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")
|
|
assert response.status_code == 200
|
|
|
|
def test_ban_tags_enabled_but_empty_list_allows_all(self, client, patched_deps):
|
|
# If ban_tags_enabled is true but ban_tags is empty nothing should be blocked.
|
|
# docker-test doesn't have ban_tags_enabled, but we can verify via the
|
|
# docker-bantags-test remote with an unlisted tag.
|
|
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-bantags-test/library/nginx/manifests/stable")
|
|
assert response.status_code == 200
|
|
|
|
def test_ban_check_does_not_apply_to_blobs(self, client, patched_deps):
|
|
# Blob paths don't contain /manifests/ — the ban check must not interfere
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"\x00" * 100
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
with patch("artifactapi.artifact.proxy._fetch_last_modified", new_callable=AsyncMock, return_value=None):
|
|
response = client.get("/v2/docker-bantags-test/library/nginx/blobs/sha256:" + "b" * 64)
|
|
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
|