unkinben
1656664dfa
ci/woodpecker/tag/docker Pipeline was successful
Repository types now live under dedicated top-level keys instead of a shared remotes: block distinguished by a type field: remotes: caching proxy remotes (no type field needed) virtuals: virtual merged-index repositories locals: local upload repositories Routes for local repos move from /api/v1/remote/ to /api/v1/local/. config.py gains get_virtual_config() and get_local_config() lookups. Root endpoint now reports all three sections. Drop root conf.d/ (was an exact duplicate of examples/conf.d-method/). Reviewed-on: #31
1063 lines
47 KiB
Python
1063 lines
47 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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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
|