unkinben
99cc71f56c
## Summary - New `terraform` package type implementing the [Terraform Registry Protocol](https://developer.hashicorp.com/terraform/internals/provider-registry-protocol) - `construct_url` prepends `/v1/providers/` so paths like `hashicorp/vault/versions` map to `registry.terraform.io/v1/providers/hashicorp/vault/versions` - `resolve_content` rewrites `download_url`, `shasums_url`, and `shasums_signature_url` in per-version download info JSON to route through a companion `releases_remote` (generic remote proxying `releases.hashicorp.com`) - Built-in mutable pattern for `{namespace}/{type}/versions` — version lists expire and are re-fetched; per-version download info is immutable - Client configuration via `.terraformrc` / `.tofurc` host block — no changes to `.tf` provider source addresses needed ## Test plan - [x] 8 unit tests covering mutable detection, URL rewriting, binary pass-through, `construct_url` correctness, and cache miss behaviour - [x] End-to-end: OpenTofu 1.10.3 pulling `hashicorp/vault v4.5.0` through docker-compose stack — `tofu init` succeeded, provider installed and signed - [x] Verified `download_url` / `shasums_url` rewritten to `hashicorp-releases` proxy in cached response - [x] All 339 tests pass Reviewed-on: #45 Co-authored-by: Ben Vincent <ben@unkin.net> Co-committed-by: Ben Vincent <ben@unkin.net>
1529 lines
71 KiB
Python
1529 lines
71 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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Puppet Forge remote /api/v1/remote/puppet-test/...
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestPuppetRemote:
|
|
def test_module_metadata_is_mutable(self, client, patched_deps):
|
|
"""v3/modules/ paths are detected as mutable (package-type default)."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b'{"current_release":{"file_uri":"/v3/files/puppetlabs-stdlib-9.7.0.tar.gz"}}'
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
response = client.get("/api/v1/remote/puppet-test/v3/modules/puppetlabs-stdlib")
|
|
assert response.status_code == 200
|
|
deps["cache"].mark_index_cached.assert_not_called()
|
|
|
|
def test_releases_path_is_mutable(self, client, patched_deps):
|
|
"""v3/releases paths are detected as mutable (package-type default)."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b'{"file_uri":"/v3/files/puppetlabs-stdlib-9.7.0.tar.gz"}'
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
response = client.get("/api/v1/remote/puppet-test/v3/releases/puppetlabs-stdlib-9.7.0")
|
|
assert response.status_code == 200
|
|
|
|
def test_relative_file_uri_rewritten_to_absolute_proxy_url(self, client, patched_deps):
|
|
"""Relative /v3/files/ paths in JSON responses are rewritten to absolute proxy URLs."""
|
|
deps = patched_deps
|
|
meta = b'{"current_release":{"file_uri":"/v3/files/puppetlabs-stdlib-9.7.0.tar.gz","version":"9.7.0"}}'
|
|
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/puppet-test/v3/modules/puppetlabs-stdlib")
|
|
assert response.status_code == 200
|
|
assert b'"/v3/files/' not in response.content
|
|
assert b"/api/v1/remote/puppet-test/v3/files/puppetlabs-stdlib-9.7.0.tar.gz" in response.content
|
|
|
|
def test_absolute_forge_url_rewritten_to_proxy(self, client, patched_deps):
|
|
"""Absolute forgeapi.puppet.com URLs in JSON are rewritten to the proxy URL."""
|
|
deps = patched_deps
|
|
meta = b'{"uri":"https://forgeapi.puppet.com/v3/modules/puppetlabs-stdlib"}'
|
|
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/puppet-test/v3/modules/puppetlabs-stdlib")
|
|
assert response.status_code == 200
|
|
assert b"forgeapi.puppet.com" not in response.content
|
|
assert b"/api/v1/remote/puppet-test" 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'{"current_release":{}}'
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
response = client.get("/api/v1/remote/puppet-test/v3/modules/puppetlabs-concat")
|
|
assert response.status_code == 200
|
|
assert "application/json" in response.headers["content-type"]
|
|
|
|
def test_tarball_served_without_rewriting(self, client, patched_deps):
|
|
"""Module tarballs (v3/files/*.tar.gz) are served as binary without URL rewriting."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"\x1f\x8b tarball bytes"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
response = client.get("/api/v1/remote/puppet-test/v3/files/puppetlabs-stdlib-9.7.0.tar.gz")
|
|
assert response.status_code == 200
|
|
assert "application/gzip" in response.headers["content-type"]
|
|
assert response.headers["X-Artifact-Source"] == "cache"
|
|
|
|
def test_tarball_not_blocked_by_immutable_pattern(self, client, patched_deps):
|
|
"""v3/files/*.tar.gz matches the configured immutable_patterns and is allowed."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"\x1f\x8b tarball bytes"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
response = client.get("/api/v1/remote/puppet-test/v3/files/puppetlabs-inifile-6.2.0.tar.gz")
|
|
assert response.status_code == 200
|
|
|
|
def test_unknown_path_blocked(self, client, patched_deps):
|
|
"""Paths outside v3/modules, v3/releases, and v3/files are blocked."""
|
|
deps = patched_deps
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
response = client.get("/api/v1/remote/puppet-test/v3/users/puppetlabs")
|
|
assert response.status_code == 403
|
|
|
|
def test_metadata_cache_miss_fetches_upstream(self, client, patched_deps):
|
|
deps = patched_deps
|
|
meta = b'{"current_release":{"file_uri":"/v3/files/puppetlabs-stdlib-9.7.0.tar.gz"}}'
|
|
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/puppet-test/v3/modules/puppetlabs-stdlib")
|
|
|
|
mock_fetch.assert_called_once()
|
|
assert response.status_code == 200
|
|
assert b'"/v3/files/' not in response.content
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Terraform registry remote (terraform-registry-test)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTerraformRemote:
|
|
def test_versions_path_is_mutable(self, client, patched_deps):
|
|
"""Provider versions listing is detected as mutable."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b'{"versions":[]}'
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
response = client.get("/api/v1/remote/terraform-registry-test/hashicorp/vault/versions")
|
|
assert response.status_code == 200
|
|
deps["cache"].mark_index_cached.assert_not_called()
|
|
|
|
def test_versions_returns_json_content_type(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b'{"versions":[]}'
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
deps["cache"].is_index_valid.return_value = True
|
|
|
|
response = client.get("/api/v1/remote/terraform-registry-test/hashicorp/vault/versions")
|
|
assert response.status_code == 200
|
|
assert "application/json" in response.headers["content-type"]
|
|
|
|
def test_download_info_download_url_rewritten(self, client, patched_deps):
|
|
"""download_url in download-info JSON is rewritten to point to the releases proxy."""
|
|
deps = patched_deps
|
|
download_info = json.dumps(
|
|
{
|
|
"os": "linux",
|
|
"arch": "amd64",
|
|
"filename": "terraform-provider-vault_0.28.0_linux_amd64.zip",
|
|
"download_url": "https://releases.hashicorp.com/terraform-provider-vault/0.28.0/terraform-provider-vault_0.28.0_linux_amd64.zip",
|
|
"shasums_url": "https://releases.hashicorp.com/terraform-provider-vault/0.28.0/terraform-provider-vault_0.28.0_SHA256SUMS",
|
|
"shasums_signature_url": "https://releases.hashicorp.com/terraform-provider-vault/0.28.0/terraform-provider-vault_0.28.0_SHA256SUMS.sig",
|
|
}
|
|
).encode()
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = download_info
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
response = client.get("/api/v1/remote/terraform-registry-test/hashicorp/vault/0.28.0/download/linux/amd64")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "releases.hashicorp.com" not in data["download_url"]
|
|
assert "/api/v1/remote/hashicorp-releases-test/" in data["download_url"]
|
|
|
|
def test_download_info_shasums_url_rewritten(self, client, patched_deps):
|
|
"""shasums_url is also rewritten to the releases proxy."""
|
|
deps = patched_deps
|
|
download_info = json.dumps(
|
|
{
|
|
"os": "linux",
|
|
"arch": "amd64",
|
|
"filename": "terraform-provider-vault_0.28.0_linux_amd64.zip",
|
|
"download_url": "https://releases.hashicorp.com/terraform-provider-vault/0.28.0/terraform-provider-vault_0.28.0_linux_amd64.zip",
|
|
"shasums_url": "https://releases.hashicorp.com/terraform-provider-vault/0.28.0/terraform-provider-vault_0.28.0_SHA256SUMS",
|
|
"shasums_signature_url": "https://releases.hashicorp.com/terraform-provider-vault/0.28.0/terraform-provider-vault_0.28.0_SHA256SUMS.sig",
|
|
}
|
|
).encode()
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = download_info
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
response = client.get("/api/v1/remote/terraform-registry-test/hashicorp/vault/0.28.0/download/linux/amd64")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "/api/v1/remote/hashicorp-releases-test/" in data["shasums_url"]
|
|
assert "/api/v1/remote/hashicorp-releases-test/" in data["shasums_signature_url"]
|
|
assert "releases.hashicorp.com" not in data["shasums_url"]
|
|
assert "releases.hashicorp.com" not in data["shasums_signature_url"]
|
|
|
|
def test_download_info_path_preserved(self, client, patched_deps):
|
|
"""The path portion of the upstream URL is preserved when rewriting."""
|
|
deps = patched_deps
|
|
zip_path = "/terraform-provider-vault/0.28.0/terraform-provider-vault_0.28.0_linux_amd64.zip"
|
|
download_info = json.dumps(
|
|
{
|
|
"download_url": f"https://releases.hashicorp.com{zip_path}",
|
|
"shasums_url": "https://releases.hashicorp.com/terraform-provider-vault/0.28.0/terraform-provider-vault_0.28.0_SHA256SUMS",
|
|
"shasums_signature_url": "https://releases.hashicorp.com/terraform-provider-vault/0.28.0/terraform-provider-vault_0.28.0_SHA256SUMS.sig",
|
|
}
|
|
).encode()
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = download_info
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
response = client.get("/api/v1/remote/terraform-registry-test/hashicorp/vault/0.28.0/download/linux/amd64")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["download_url"].endswith(zip_path)
|
|
|
|
def test_zip_served_as_binary(self, client, patched_deps):
|
|
"""Provider zip files are served as binary without JSON rewriting."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = True
|
|
deps["storage"].download_object.return_value = b"PK\x03\x04 zip bytes"
|
|
deps["cache"].is_mutable_file.return_value = False
|
|
|
|
response = client.get(
|
|
"/api/v1/remote/hashicorp-releases-test/terraform-provider-vault/0.28.0/terraform-provider-vault_0.28.0_linux_amd64.zip"
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.headers["X-Artifact-Source"] == "cache"
|
|
|
|
def test_construct_url_prepends_v1_providers(self, client, patched_deps):
|
|
"""Upstream URL for the terraform package type prepends /v1/providers/."""
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = False
|
|
|
|
with patch(
|
|
"artifactapi.artifact.proxy.cache_single_artifact",
|
|
new_callable=AsyncMock,
|
|
return_value={"status": "cached"},
|
|
) as mock_fetch:
|
|
deps["storage"].download_object.return_value = b'{"versions":[]}'
|
|
deps["cache"].is_mutable_file.return_value = True
|
|
client.get("/api/v1/remote/terraform-registry-test/hashicorp/vault/versions")
|
|
|
|
called_url = mock_fetch.call_args[0][0]
|
|
assert called_url == "https://registry.terraform.io/v1/providers/hashicorp/vault/versions"
|
|
|
|
def test_versions_cache_miss_fetches_upstream(self, client, patched_deps):
|
|
deps = patched_deps
|
|
deps["storage"].exists.return_value = False
|
|
deps["storage"].download_object.return_value = b'{"versions":[]}'
|
|
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/terraform-registry-test/hashicorp/vault/versions")
|
|
|
|
mock_fetch.assert_called_once()
|
|
assert response.status_code == 200
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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
|