unkin/artifactapi
2
0
Fork 0
Code Issues 5 Pull Requests 2 Actions Packages Projects Releases 1 Wiki Activity
Files
9287cf7cf2a77b855e4922942dc661b2d8a2e868
artifactapi/tests/test_routes.py
T
unkinben 9287cf7cf2 feat: add Puppet Forge remote type (#44) 
## Summary

- Adds \`package: puppet\` for proxying Puppet Forge (forgeapi.puppet.com)
- \`remote/puppet.py\` rewrites JSON responses: absolute forge URLs → proxy URLs, and relative \`/v3/files/\` \`file_uri\` paths → absolute proxy URLs. g10k uses Go's \`url.ResolveReference\`, so an absolute \`file_uri\` overrides the base URL entirely — tarballs are fetched directly from the proxy without a second hop
- Built-in mutable patterns: \`^v3/modules/\` and \`^v3/releases\` (module metadata); tarballs at \`v3/files/\` are configured as immutable via \`immutable_patterns\`
- 9 new tests covering mutable detection, URL rewriting (relative \`file_uri\` and absolute forge URLs), content-type, tarball pass-through, and pattern blocking

## Client configuration

**g10k config file** (\`forge_base_url\` at root level):
\`\`\`yaml
cachedir: /tmp/g10k
forge_base_url: https://artifacts.example.com/api/v1/remote/puppet-forge
sources:
  control:
    remote: git@git.example.com:puppet/control.git
    basedir: /etc/puppetlabs/code/environments
\`\`\`

**Puppetfile** (\`forge.baseUrl\` directive, works with \`-puppetfile\` mode):
\`\`\`ruby
forge.baseUrl https://artifacts.example.com/api/v1/remote/puppet-forge

mod 'puppetlabs-stdlib', '9.7.0'
\`\`\`

## Test plan

- [x] 331 unit tests pass (\`make test\`)
- [x] End-to-end: g10k 0.9.10 on AlmaLinux 9 via \`forge_base_url\` — stdlib 9.7.0, inifile 6.2.0, concat 9.1.0 installed; proxy logs confirm cache MISS → fetch → ADD for metadata and tarballs
- [x] End-to-end: \`forge.baseUrl\` Puppetfile directive with \`-puppetfile\` mode — same result

Reviewed-on: #44
2026-05-17 10:56:50 +10:00

1385 lines
63 KiB
Python
Raw Blame History

"""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
# ---------------------------------------------------------------------------
# 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
Reference in New Issue View Git Blame Copy Permalink