feat: rename include/index patterns to immutable/mutable with per-remote TTL

Replace the include_patterns/index_patterns split with a clearer
immutable_patterns/mutable_patterns model:

- immutable_patterns: artifacts cached indefinitely (no TTL)
- mutable_patterns: artifacts that expire and are re-fetched after
  cache.mutable_ttl seconds (replaces cache.index_ttl)

_PACKAGE_INDEX_PATTERNS renamed to _PACKAGE_MUTABLE_PATTERNS; all
built-in package-type index patterns (APKINDEX, repomd, manifests, etc.)
default to the remote's mutable_ttl (default 1 hour).

cache.file_ttl renamed to cache.immutable_ttl for consistency.
Adds github-archive remote to remotes.yaml as a worked example showing
tag archives as immutable and branch archives as mutable (1-day TTL).

docker-compose.yml: fix VERSION=dev → 2.2.2.dev0 (valid PEP 440),
add :z SELinux label to volume mounts.
This commit is contained in:
2026-04-27 00:40:13 +10:00
parent 4619ae18d8
commit ce01a94141
8 changed files with 173 additions and 183 deletions
+3 -3
View File
@@ -6,12 +6,12 @@ services:
context: .
dockerfile: Dockerfile
args:
- VERSION=dev
- VERSION=2.2.2.dev0
ports:
- "8000:8000"
volumes:
- ./remotes.yaml:/app/remotes.yaml:ro
- ./ca-bundle.pem:/app/ca-bundle.pem:ro
- ./remotes.yaml:/app/remotes.yaml:ro,z
- ./ca-bundle.pem:/app/ca-bundle.pem:ro,z
environment:
- CONFIG_PATH=/app/remotes.yaml
- DBHOST=postgres
+2 -2
View File
@@ -19,8 +19,8 @@ class RedisCache:
self.client = None
self.available = False
def is_index_file(self, file_path: str, patterns: list[str] | None = None) -> bool:
"""Return True if file_path matches any of the index patterns."""
def is_mutable_file(self, file_path: str, patterns: list[str] | None = None) -> bool:
"""Return True if file_path matches any of the mutable patterns."""
if patterns is None:
patterns = []
return any(re.search(p, file_path) for p in patterns)
+10 -15
View File
@@ -3,7 +3,7 @@ import os
import yaml
_PACKAGE_INDEX_PATTERNS: dict[str, list[str]] = {
_PACKAGE_MUTABLE_PATTERNS: dict[str, list[str]] = {
"alpine": [
r"APKINDEX\.tar\.gz$",
],
@@ -55,26 +55,21 @@ class ConfigManager:
self._check_reload()
return self.config.get("remotes", {}).get(remote_name)
def get_repository_patterns(self, remote_name: str, repo_path: str) -> list:
def get_immutable_patterns(self, remote_name: str, repo_path: str = "") -> list[str]:
remote_config = self.get_remote_config(remote_name)
if not remote_config:
return []
repositories = remote_config.get("repositories", {})
# Handle both dict (GitHub style) and list (Alpine style) repositories
if isinstance(repositories, dict):
repo_config = repositories.get(repo_path)
if repo_config:
patterns = repo_config.get("include_patterns", [])
patterns = repo_config.get("immutable_patterns", [])
else:
patterns = remote_config.get("include_patterns", [])
elif isinstance(repositories, list):
# For Alpine, repositories is just a list of allowed repo names
# Pattern matching is handled by the main include_patterns
patterns = remote_config.get("include_patterns", [])
patterns = remote_config.get("immutable_patterns", [])
else:
patterns = remote_config.get("include_patterns", [])
patterns = remote_config.get("immutable_patterns", [])
return patterns
@@ -129,18 +124,18 @@ class ConfigManager:
db_url = f"postgresql://{db_user}:{db_pass}@{db_host}:{db_port}/{db_name}"
return {"url": db_url}
def get_index_patterns(self, remote_name: str) -> list[str]:
"""Return index-file patterns for a remote.
def get_mutable_patterns(self, remote_name: str) -> list[str]:
"""Return mutable-file patterns for a remote (TTL is configured per-remote in cache.index_ttl).
Merges the package-level defaults with any extra patterns listed under
``index_patterns`` in the remote's config.
``mutable_patterns`` in the remote's config.
"""
remote_config = self.get_remote_config(remote_name)
if not remote_config:
return []
package = remote_config.get("package", "generic")
defaults = _PACKAGE_INDEX_PATTERNS.get(package, [])
extra = remote_config.get("index_patterns", [])
defaults = _PACKAGE_MUTABLE_PATTERNS.get(package, [])
extra = remote_config.get("mutable_patterns", [])
return defaults + [p for p in extra if p not in defaults]
def get_cache_config(self, remote_name: str) -> dict:
+23 -27
View File
@@ -163,13 +163,13 @@ async def construct_remote_url(remote_name: str, path: str) -> str:
async def check_artifact_patterns(remote_name: str, repo_path: str, file_path: str, full_path: str) -> bool:
# First check if this is an index file - always allow index files
index_patterns = config.get_index_patterns(remote_name)
if cache.is_index_file(file_path, index_patterns) or cache.is_index_file(full_path, index_patterns):
# Mutable files (index files) are always allowed through
mutable_patterns = config.get_mutable_patterns(remote_name)
if cache.is_mutable_file(file_path, mutable_patterns) or cache.is_mutable_file(full_path, mutable_patterns):
return True
# Then check basic include patterns
patterns = config.get_repository_patterns(remote_name, repo_path)
# Check immutable include patterns
patterns = config.get_immutable_patterns(remote_name, repo_path)
if not patterns:
return True # Allow all if no patterns configured
@@ -183,7 +183,6 @@ async def check_artifact_patterns(remote_name: str, repo_path: str, file_path: s
if not pattern_matched:
return False
# All remotes now use pattern-based filtering only - no additional checks needed
return True
@@ -297,15 +296,13 @@ async def get_artifact(remote_name: str, path: str):
if not storage.exists(cached_key):
cached_key = None
# For index files, check Redis TTL validity
# For mutable files, check Redis TTL validity
filename = os.path.basename(path)
is_index = cache.is_index_file(path, config.get_index_patterns(remote_name))
is_mutable = cache.is_mutable_file(path, config.get_mutable_patterns(remote_name))
if cached_key and is_index:
# Index file exists, but check if it's still valid
if cached_key and is_mutable:
if not cache.is_index_valid(remote_name, path):
# Index has expired, remove it from S3
logger.info(f"Index EXPIRED: {remote_name}/{path} - removing from cache")
logger.info(f"Mutable file EXPIRED: {remote_name}/{path} - removing from cache")
cache.cleanup_expired_index(storage, remote_name, path)
cached_key = None # Force re-download
@@ -359,13 +356,12 @@ async def get_artifact(remote_name: str, path: str):
logger.error(f"Cache ADD FAILED: {remote_name}/{path} - {result['error']}")
raise HTTPException(status_code=502, detail=f"Failed to fetch artifact: {result['error']}")
# Mark index files as cached in Redis if this was a new download
if result["status"] == "cached" and is_index:
# Get TTL from remote config
# Mark mutable files as cached in Redis with TTL
if result["status"] == "cached" and is_mutable:
cache_config = config.get_cache_config(remote_name)
index_ttl = cache_config.get("index_ttl", 300) # Default 5 minutes
cache.mark_index_cached(remote_name, path, index_ttl)
logger.info(f"Index file cached with TTL: {remote_name}/{path} (ttl: {index_ttl}s)")
mutable_ttl = cache_config.get("mutable_ttl", 3600)
cache.mark_index_cached(remote_name, path, mutable_ttl)
logger.info(f"Mutable file cached with TTL: {remote_name}/{path} (ttl: {mutable_ttl}s)")
# Now return the cached artifact
try:
@@ -424,8 +420,8 @@ async def docker_v2_proxy(request: Request, remote_name: str, path: str):
if remote_config.get("package") != "docker":
raise HTTPException(status_code=400, detail=f"Remote '{remote_name}' is not a docker remote")
# Check include_patterns against the image name (e.g. "library/nginx")
patterns = config.get_repository_patterns(remote_name, "")
# Check immutable_patterns against the image name (e.g. "library/nginx")
patterns = config.get_immutable_patterns(remote_name, "")
if patterns:
path_parts = path.split("/")
image_name = "/".join(path_parts[:2]) if len(path_parts) >= 2 else path
@@ -439,11 +435,11 @@ async def docker_v2_proxy(request: Request, remote_name: str, path: str):
if not storage.exists(cached_key):
cached_key = None
is_index = cache.is_index_file(path, config.get_index_patterns(remote_name))
is_mutable = cache.is_mutable_file(path, config.get_mutable_patterns(remote_name))
if cached_key and is_index:
if cached_key and is_mutable:
if not cache.is_index_valid(remote_name, path):
logger.info(f"Index EXPIRED: {remote_name}/{path} - removing from cache")
logger.info(f"Mutable file EXPIRED: {remote_name}/{path} - removing from cache")
cache.cleanup_expired_index(storage, remote_name, path)
cached_key = None
@@ -452,11 +448,11 @@ async def docker_v2_proxy(request: Request, remote_name: str, path: str):
result = await cache_single_artifact(remote_url, remote_name, path)
if result["status"] == "error":
raise HTTPException(status_code=502, detail=f"Failed to fetch: {result['error']}")
if result["status"] == "cached" and is_index:
if result["status"] == "cached" and is_mutable:
cache_config = config.get_cache_config(remote_name)
index_ttl = cache_config.get("index_ttl", 300)
cache.mark_index_cached(remote_name, path, index_ttl)
logger.info(f"Index file cached with TTL: {remote_name}/{path} (ttl: {index_ttl}s)")
mutable_ttl = cache_config.get("mutable_ttl", 3600)
cache.mark_index_cached(remote_name, path, mutable_ttl)
logger.info(f"Mutable file cached with TTL: {remote_name}/{path} (ttl: {mutable_ttl}s)")
artifact_data = storage.download_object(storage.get_object_key(remote_name, path))
+12 -12
View File
@@ -22,47 +22,47 @@ TEST_REMOTES = {
"base_url": "https://dl-cdn.alpinelinux.org",
"type": "remote",
"package": "alpine",
"include_patterns": [".*/x86_64/.*\\.apk$"],
"cache": {"file_ttl": 0, "index_ttl": 3600},
"immutable_patterns": [".*/x86_64/.*\\.apk$"],
"cache": {"immutable_ttl": 0, "mutable_ttl": 3600},
},
"rpm-test": {
"base_url": "https://example.com/rpm",
"type": "remote",
"package": "rpm",
"include_patterns": [".*/x86_64/.*\\.rpm$", ".*/repodata/.*$"],
"cache": {"file_ttl": 0, "index_ttl": 3600},
"immutable_patterns": [".*/x86_64/.*\\.rpm$", ".*/repodata/.*$"],
"cache": {"immutable_ttl": 0, "mutable_ttl": 3600},
},
"docker-test": {
"base_url": "https://registry.example.com",
"type": "remote",
"package": "docker",
"cache": {"file_ttl": 0, "index_ttl": 300},
"cache": {"immutable_ttl": 0, "mutable_ttl": 300},
},
"docker-restricted": {
"base_url": "https://registry.example.com",
"type": "remote",
"package": "docker",
"include_patterns": ["^library/nginx"],
"cache": {"file_ttl": 0, "index_ttl": 300},
"immutable_patterns": ["^library/nginx"],
"cache": {"immutable_ttl": 0, "mutable_ttl": 300},
},
"generic-test": {
"base_url": "https://releases.example.com",
"type": "remote",
"package": "generic",
"include_patterns": [".*\\.tar\\.gz$"],
"cache": {"file_ttl": 0, "index_ttl": 0},
"immutable_patterns": [".*\\.tar\\.gz$"],
"cache": {"immutable_ttl": 0, "mutable_ttl": 0},
},
"custom-index-test": {
"base_url": "https://example.com",
"type": "remote",
"package": "generic",
"index_patterns": ["metadata\\.json$"],
"cache": {"file_ttl": 0, "index_ttl": 600},
"mutable_patterns": ["metadata\\.json$"],
"cache": {"immutable_ttl": 0, "mutable_ttl": 600},
},
"local-test": {
"type": "local",
"package": "generic",
"cache": {"file_ttl": 0, "index_ttl": 0},
"cache": {"immutable_ttl": 0, "mutable_ttl": 0},
},
}
}
+61 -61
View File
@@ -1,4 +1,4 @@
"""Tests for RedisCache, focusing on is_index_file with configurable patterns."""
"""Tests for RedisCache, focusing on is_mutable_file with configurable patterns."""
import hashlib
from unittest.mock import ANY, MagicMock, patch
@@ -6,7 +6,7 @@ from unittest.mock import ANY, MagicMock, patch
import pytest
from artifactapi.cache import RedisCache
from artifactapi.config import _PACKAGE_INDEX_PATTERNS
from artifactapi.config import _PACKAGE_MUTABLE_PATTERNS
@pytest.fixture
@@ -38,139 +38,139 @@ def cache_with_redis(mock_redis_client):
# ---------------------------------------------------------------------------
# is_index_file — alpine patterns
# is_mutable_file — alpine patterns
# ---------------------------------------------------------------------------
class TestIsIndexFileAlpine:
class TestIsMutableFileAlpine:
def test_apkindex_tarball_is_index(self, bare_cache):
patterns = _PACKAGE_INDEX_PATTERNS["alpine"]
assert bare_cache.is_index_file("alpine/v3.18/x86_64/APKINDEX.tar.gz", patterns)
patterns = _PACKAGE_MUTABLE_PATTERNS["alpine"]
assert bare_cache.is_mutable_file("alpine/v3.18/x86_64/APKINDEX.tar.gz", patterns)
def test_nested_apkindex_is_index(self, bare_cache):
patterns = _PACKAGE_INDEX_PATTERNS["alpine"]
assert bare_cache.is_index_file("mirrors/dl-cdn/alpine/v3.19/community/x86_64/APKINDEX.tar.gz", patterns)
patterns = _PACKAGE_MUTABLE_PATTERNS["alpine"]
assert bare_cache.is_mutable_file("mirrors/dl-cdn/alpine/v3.19/community/x86_64/APKINDEX.tar.gz", patterns)
def test_apk_package_is_not_index(self, bare_cache):
patterns = _PACKAGE_INDEX_PATTERNS["alpine"]
assert not bare_cache.is_index_file("alpine/v3.18/x86_64/musl-1.2.4-r2.apk", patterns)
patterns = _PACKAGE_MUTABLE_PATTERNS["alpine"]
assert not bare_cache.is_mutable_file("alpine/v3.18/x86_64/musl-1.2.4-r2.apk", patterns)
def test_random_tarball_is_not_index(self, bare_cache):
patterns = _PACKAGE_INDEX_PATTERNS["alpine"]
assert not bare_cache.is_index_file("some/path/archive.tar.gz", patterns)
patterns = _PACKAGE_MUTABLE_PATTERNS["alpine"]
assert not bare_cache.is_mutable_file("some/path/archive.tar.gz", patterns)
def test_apkindex_signature_file_is_not_index(self, bare_cache):
# Signature file adjacent to the index should not be treated as an index
patterns = _PACKAGE_INDEX_PATTERNS["alpine"]
assert not bare_cache.is_index_file("alpine/v3.18/x86_64/APKINDEX.tar.gz.sig", patterns)
patterns = _PACKAGE_MUTABLE_PATTERNS["alpine"]
assert not bare_cache.is_mutable_file("alpine/v3.18/x86_64/APKINDEX.tar.gz.sig", patterns)
def test_apkindex_tmp_file_is_not_index(self, bare_cache):
patterns = _PACKAGE_INDEX_PATTERNS["alpine"]
assert not bare_cache.is_index_file("alpine/v3.18/x86_64/APKINDEX.tar.gz.tmp", patterns)
patterns = _PACKAGE_MUTABLE_PATTERNS["alpine"]
assert not bare_cache.is_mutable_file("alpine/v3.18/x86_64/APKINDEX.tar.gz.tmp", patterns)
# ---------------------------------------------------------------------------
# is_index_file — rpm patterns
# is_mutable_file — rpm patterns
# ---------------------------------------------------------------------------
class TestIsIndexFileRpm:
class TestIsMutableFileRpm:
def test_repomd_xml_is_index(self, bare_cache):
patterns = _PACKAGE_INDEX_PATTERNS["rpm"]
assert bare_cache.is_index_file("almalinux/9/x86_64/repomd.xml", patterns)
patterns = _PACKAGE_MUTABLE_PATTERNS["rpm"]
assert bare_cache.is_mutable_file("almalinux/9/x86_64/repomd.xml", patterns)
def test_repodata_primary_xml_gz_is_index(self, bare_cache):
patterns = _PACKAGE_INDEX_PATTERNS["rpm"]
assert bare_cache.is_index_file("repo/repodata/primary.xml.gz", patterns)
patterns = _PACKAGE_MUTABLE_PATTERNS["rpm"]
assert bare_cache.is_mutable_file("repo/repodata/primary.xml.gz", patterns)
def test_repodata_sqlite_is_index(self, bare_cache):
patterns = _PACKAGE_INDEX_PATTERNS["rpm"]
assert bare_cache.is_index_file("repo/repodata/primary.sqlite", patterns)
patterns = _PACKAGE_MUTABLE_PATTERNS["rpm"]
assert bare_cache.is_mutable_file("repo/repodata/primary.sqlite", patterns)
def test_repodata_sqlite_bz2_is_index(self, bare_cache):
patterns = _PACKAGE_INDEX_PATTERNS["rpm"]
assert bare_cache.is_index_file("repo/repodata/other.sqlite.bz2", patterns)
patterns = _PACKAGE_MUTABLE_PATTERNS["rpm"]
assert bare_cache.is_mutable_file("repo/repodata/other.sqlite.bz2", patterns)
def test_repodata_yaml_xz_is_index(self, bare_cache):
patterns = _PACKAGE_INDEX_PATTERNS["rpm"]
assert bare_cache.is_index_file("repo/repodata/comps.yaml.xz", patterns)
patterns = _PACKAGE_MUTABLE_PATTERNS["rpm"]
assert bare_cache.is_mutable_file("repo/repodata/comps.yaml.xz", patterns)
def test_packages_gz_pattern_matches_any_path(self, bare_cache):
# The Packages.gz$ regex is a carryover from the original hardcoded logic and
# deliberately matches any path ending in Packages.gz — including Debian-style paths.
# This test documents that intentional behaviour.
patterns = _PACKAGE_INDEX_PATTERNS["rpm"]
assert bare_cache.is_index_file("debian/dists/stable/main/binary-amd64/Packages.gz", patterns)
patterns = _PACKAGE_MUTABLE_PATTERNS["rpm"]
assert bare_cache.is_mutable_file("debian/dists/stable/main/binary-amd64/Packages.gz", patterns)
def test_rpm_package_is_not_index(self, bare_cache):
patterns = _PACKAGE_INDEX_PATTERNS["rpm"]
assert not bare_cache.is_index_file("almalinux/9/x86_64/Packages/bash-5.1.8.x86_64.rpm", patterns)
patterns = _PACKAGE_MUTABLE_PATTERNS["rpm"]
assert not bare_cache.is_mutable_file("almalinux/9/x86_64/Packages/bash-5.1.8.x86_64.rpm", patterns)
def test_arbitrary_xml_outside_repodata_is_not_index(self, bare_cache):
patterns = _PACKAGE_INDEX_PATTERNS["rpm"]
assert not bare_cache.is_index_file("some/path/config.xml", patterns)
patterns = _PACKAGE_MUTABLE_PATTERNS["rpm"]
assert not bare_cache.is_mutable_file("some/path/config.xml", patterns)
# ---------------------------------------------------------------------------
# is_index_file — docker patterns
# is_mutable_file — docker patterns
# ---------------------------------------------------------------------------
class TestIsIndexFileDocker:
class TestIsMutableFileDocker:
def test_tag_manifest_is_index(self, bare_cache):
patterns = _PACKAGE_INDEX_PATTERNS["docker"]
assert bare_cache.is_index_file("library/nginx/manifests/latest", patterns)
patterns = _PACKAGE_MUTABLE_PATTERNS["docker"]
assert bare_cache.is_mutable_file("library/nginx/manifests/latest", patterns)
def test_version_tag_manifest_is_index(self, bare_cache):
patterns = _PACKAGE_INDEX_PATTERNS["docker"]
assert bare_cache.is_index_file("library/nginx/manifests/1.25.3", patterns)
patterns = _PACKAGE_MUTABLE_PATTERNS["docker"]
assert bare_cache.is_mutable_file("library/nginx/manifests/1.25.3", patterns)
def test_hyphenated_tag_manifest_is_index(self, bare_cache):
patterns = _PACKAGE_INDEX_PATTERNS["docker"]
assert bare_cache.is_index_file("library/nginx/manifests/latest-rc", patterns)
patterns = _PACKAGE_MUTABLE_PATTERNS["docker"]
assert bare_cache.is_mutable_file("library/nginx/manifests/latest-rc", patterns)
def test_numeric_date_tag_manifest_is_index(self, bare_cache):
patterns = _PACKAGE_INDEX_PATTERNS["docker"]
assert bare_cache.is_index_file("library/nginx/manifests/20240101", patterns)
patterns = _PACKAGE_MUTABLE_PATTERNS["docker"]
assert bare_cache.is_mutable_file("library/nginx/manifests/20240101", patterns)
def test_digest_manifest_is_not_index(self, bare_cache):
patterns = _PACKAGE_INDEX_PATTERNS["docker"]
patterns = _PACKAGE_MUTABLE_PATTERNS["docker"]
digest = "sha256:" + "a" * 64
assert not bare_cache.is_index_file(f"library/nginx/manifests/{digest}", patterns)
assert not bare_cache.is_mutable_file(f"library/nginx/manifests/{digest}", patterns)
def test_tags_list_is_index(self, bare_cache):
patterns = _PACKAGE_INDEX_PATTERNS["docker"]
assert bare_cache.is_index_file("library/nginx/tags/list", patterns)
patterns = _PACKAGE_MUTABLE_PATTERNS["docker"]
assert bare_cache.is_mutable_file("library/nginx/tags/list", patterns)
def test_blob_is_not_index(self, bare_cache):
patterns = _PACKAGE_INDEX_PATTERNS["docker"]
assert not bare_cache.is_index_file("library/nginx/blobs/sha256:abc123", patterns)
patterns = _PACKAGE_MUTABLE_PATTERNS["docker"]
assert not bare_cache.is_mutable_file("library/nginx/blobs/sha256:abc123", patterns)
# ---------------------------------------------------------------------------
# is_index_file — edge cases
# is_mutable_file — edge cases
# ---------------------------------------------------------------------------
class TestIsIndexFileEdgeCases:
class TestIsMutableFileEdgeCases:
def test_empty_patterns_nothing_is_index(self, bare_cache):
assert not bare_cache.is_index_file("APKINDEX.tar.gz", [])
assert not bare_cache.is_index_file("repomd.xml", [])
assert not bare_cache.is_index_file("library/nginx/manifests/latest", [])
assert not bare_cache.is_mutable_file("APKINDEX.tar.gz", [])
assert not bare_cache.is_mutable_file("repomd.xml", [])
assert not bare_cache.is_mutable_file("library/nginx/manifests/latest", [])
def test_none_patterns_nothing_is_index(self, bare_cache):
assert not bare_cache.is_index_file("APKINDEX.tar.gz", None)
assert not bare_cache.is_index_file("repomd.xml", None)
assert not bare_cache.is_mutable_file("APKINDEX.tar.gz", None)
assert not bare_cache.is_mutable_file("repomd.xml", None)
def test_custom_patterns_match(self, bare_cache):
patterns = [r"metadata\.json$", r"index\.yaml$"]
assert bare_cache.is_index_file("repo/metadata.json", patterns)
assert bare_cache.is_index_file("repo/subdir/index.yaml", patterns)
assert not bare_cache.is_index_file("repo/data.tar.gz", patterns)
assert bare_cache.is_mutable_file("repo/metadata.json", patterns)
assert bare_cache.is_mutable_file("repo/subdir/index.yaml", patterns)
assert not bare_cache.is_mutable_file("repo/data.tar.gz", patterns)
def test_custom_pattern_does_not_match_standard_index(self, bare_cache):
patterns = [r"metadata\.json$"]
assert not bare_cache.is_index_file("APKINDEX.tar.gz", patterns)
assert not bare_cache.is_mutable_file("APKINDEX.tar.gz", patterns)
# ---------------------------------------------------------------------------
+37 -38
View File
@@ -1,4 +1,4 @@
"""Tests for ConfigManager, focusing on get_index_patterns (new logic)."""
"""Tests for ConfigManager, focusing on get_mutable_patterns and get_immutable_patterns."""
import os
@@ -21,45 +21,44 @@ def make_config(tmp_path):
# ---------------------------------------------------------------------------
# get_index_patterns
# get_mutable_patterns
# ---------------------------------------------------------------------------
class TestGetIndexPatterns:
class TestGetMutablePatterns:
def test_alpine_returns_package_defaults(self, make_config):
cfg = make_config({"r": {"type": "remote", "package": "alpine", "base_url": "https://x.com"}})
patterns = cfg.get_index_patterns("r")
# Assert against literal strings, not the live constant, so a rename doesn't mask a regression
patterns = cfg.get_mutable_patterns("r")
assert r"APKINDEX\.tar\.gz$" in patterns
def test_rpm_returns_package_defaults(self, make_config):
cfg = make_config({"r": {"type": "remote", "package": "rpm", "base_url": "https://x.com"}})
patterns = cfg.get_index_patterns("r")
patterns = cfg.get_mutable_patterns("r")
assert r"repomd\.xml$" in patterns
assert any("repodata" in p for p in patterns)
def test_docker_returns_package_defaults(self, make_config):
cfg = make_config({"r": {"type": "remote", "package": "docker", "base_url": "https://x.com"}})
patterns = cfg.get_index_patterns("r")
patterns = cfg.get_mutable_patterns("r")
assert any("manifests" in p for p in patterns)
assert any("tags/list" in p for p in patterns)
def test_generic_returns_empty_list(self, make_config):
cfg = make_config({"r": {"type": "remote", "package": "generic", "base_url": "https://x.com"}})
assert cfg.get_index_patterns("r") == []
assert cfg.get_mutable_patterns("r") == []
def test_unknown_remote_returns_empty_list(self, make_config):
cfg = make_config({})
assert cfg.get_index_patterns("nonexistent") == []
assert cfg.get_mutable_patterns("nonexistent") == []
def test_missing_package_field_defaults_to_generic(self, make_config):
cfg = make_config({"r": {"type": "remote", "base_url": "https://x.com"}})
assert cfg.get_index_patterns("r") == []
assert cfg.get_mutable_patterns("r") == []
def test_unknown_package_type_returns_empty_list(self, make_config):
# A mis-spelled package type silently returns [] — this is a known footgun
cfg = make_config({"r": {"type": "remote", "package": "deb", "base_url": "https://x.com"}})
assert cfg.get_index_patterns("r") == []
assert cfg.get_mutable_patterns("r") == []
def test_extra_patterns_appended_after_defaults(self, make_config):
cfg = make_config(
@@ -68,11 +67,11 @@ class TestGetIndexPatterns:
"type": "remote",
"package": "alpine",
"base_url": "https://x.com",
"index_patterns": [r"custom\.json$"],
"mutable_patterns": [r"custom\.json$"],
}
}
)
patterns = cfg.get_index_patterns("r")
patterns = cfg.get_mutable_patterns("r")
assert r"APKINDEX\.tar\.gz$" in patterns
assert r"custom\.json$" in patterns
# Defaults come first
@@ -85,11 +84,11 @@ class TestGetIndexPatterns:
"type": "remote",
"package": "alpine",
"base_url": "https://x.com",
"index_patterns": [],
"mutable_patterns": [],
}
}
)
assert r"APKINDEX\.tar\.gz$" in cfg.get_index_patterns("r")
assert r"APKINDEX\.tar\.gz$" in cfg.get_mutable_patterns("r")
def test_duplicate_extra_pattern_not_added_twice(self, make_config):
existing = r"APKINDEX\.tar\.gz$"
@@ -99,11 +98,11 @@ class TestGetIndexPatterns:
"type": "remote",
"package": "alpine",
"base_url": "https://x.com",
"index_patterns": [existing],
"mutable_patterns": [existing],
}
}
)
patterns = cfg.get_index_patterns("r")
patterns = cfg.get_mutable_patterns("r")
assert patterns.count(existing) == 1
def test_generic_with_only_extra_patterns(self, make_config):
@@ -113,11 +112,11 @@ class TestGetIndexPatterns:
"type": "remote",
"package": "generic",
"base_url": "https://x.com",
"index_patterns": [r"meta\.json$", r"index\.yaml$"],
"mutable_patterns": [r"meta\.json$", r"index\.yaml$"],
}
}
)
assert cfg.get_index_patterns("r") == [r"meta\.json$", r"index\.yaml$"]
assert cfg.get_mutable_patterns("r") == [r"meta\.json$", r"index\.yaml$"]
def test_rpm_extra_patterns_merged(self, make_config):
cfg = make_config(
@@ -126,41 +125,41 @@ class TestGetIndexPatterns:
"type": "remote",
"package": "rpm",
"base_url": "https://x.com",
"index_patterns": [r"custom-meta\.xml$"],
"mutable_patterns": [r"custom-meta\.xml$"],
}
}
)
patterns = cfg.get_index_patterns("r")
patterns = cfg.get_mutable_patterns("r")
assert r"repomd\.xml$" in patterns
assert r"custom-meta\.xml$" in patterns
# ---------------------------------------------------------------------------
# get_repository_patterns
# get_immutable_patterns
# ---------------------------------------------------------------------------
class TestGetRepositoryPatterns:
def test_returns_include_patterns(self, make_config):
class TestGetImmutablePatterns:
def test_returns_immutable_patterns(self, make_config):
cfg = make_config(
{
"r": {
"type": "remote",
"package": "generic",
"base_url": "https://x.com",
"include_patterns": [r".*\.tar\.gz$"],
"immutable_patterns": [r".*\.tar\.gz$"],
}
}
)
assert cfg.get_repository_patterns("r", "") == [r".*\.tar\.gz$"]
assert cfg.get_immutable_patterns("r") == [r".*\.tar\.gz$"]
def test_returns_empty_for_missing_remote(self, make_config):
cfg = make_config({})
assert cfg.get_repository_patterns("nonexistent", "") == []
assert cfg.get_immutable_patterns("nonexistent") == []
def test_returns_empty_when_no_patterns_configured(self, make_config):
cfg = make_config({"r": {"type": "remote", "package": "generic", "base_url": "https://x.com"}})
assert cfg.get_repository_patterns("r", "") == []
assert cfg.get_immutable_patterns("r") == []
def test_multiple_patterns_returned(self, make_config):
patterns = [r".*\.rpm$", r".*/repodata/.*$"]
@@ -170,11 +169,11 @@ class TestGetRepositoryPatterns:
"type": "remote",
"package": "rpm",
"base_url": "https://x.com",
"include_patterns": patterns,
"immutable_patterns": patterns,
}
}
)
assert cfg.get_repository_patterns("r", "") == patterns
assert cfg.get_immutable_patterns("r") == patterns
def test_dict_keyed_repositories_returns_per_repo_patterns(self, make_config):
cfg = make_config(
@@ -183,14 +182,14 @@ class TestGetRepositoryPatterns:
"type": "remote",
"package": "generic",
"base_url": "https://x.com",
"include_patterns": [r".*\.tar\.gz$"],
"immutable_patterns": [r".*\.tar\.gz$"],
"repositories": {
"/path/to/repo": {"include_patterns": [r".*\.rpm$"]},
"/path/to/repo": {"immutable_patterns": [r".*\.rpm$"]},
},
}
}
)
assert cfg.get_repository_patterns("r", "/path/to/repo") == [r".*\.rpm$"]
assert cfg.get_immutable_patterns("r", "/path/to/repo") == [r".*\.rpm$"]
def test_dict_keyed_repositories_falls_back_to_remote_patterns(self, make_config):
cfg = make_config(
@@ -199,14 +198,14 @@ class TestGetRepositoryPatterns:
"type": "remote",
"package": "generic",
"base_url": "https://x.com",
"include_patterns": [r".*\.tar\.gz$"],
"immutable_patterns": [r".*\.tar\.gz$"],
"repositories": {
"/path/to/repo": {"include_patterns": [r".*\.rpm$"]},
"/path/to/repo": {"immutable_patterns": [r".*\.rpm$"]},
},
}
}
)
assert cfg.get_repository_patterns("r", "/unknown/path") == [r".*\.tar\.gz$"]
assert cfg.get_immutable_patterns("r", "/unknown/path") == [r".*\.tar\.gz$"]
# ---------------------------------------------------------------------------
@@ -222,11 +221,11 @@ class TestGetCacheConfig:
"type": "remote",
"package": "generic",
"base_url": "https://x.com",
"cache": {"file_ttl": 0, "index_ttl": 7200},
"cache": {"immutable_ttl": 0, "mutable_ttl": 7200},
}
}
)
assert cfg.get_cache_config("r") == {"file_ttl": 0, "index_ttl": 7200}
assert cfg.get_cache_config("r") == {"immutable_ttl": 0, "mutable_ttl": 7200}
def test_returns_empty_dict_for_missing_remote(self, make_config):
cfg = make_config({})
+25 -25
View File
@@ -25,7 +25,7 @@ def mock_storage():
@pytest.fixture
def mock_cache():
m = MagicMock()
m.is_index_file.return_value = False
m.is_mutable_file.return_value = False
m.is_index_valid.return_value = True
m.available = False
m.client = None
@@ -123,7 +123,7 @@ class TestDockerProxy:
).encode()
deps["storage"].exists.return_value = True
deps["storage"].download_object.return_value = manifest
deps["cache"].is_index_file.return_value = True
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")
@@ -140,7 +140,7 @@ class TestDockerProxy:
).encode()
deps["storage"].exists.return_value = True
deps["storage"].download_object.return_value = manifest
deps["cache"].is_index_file.return_value = True
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")
@@ -158,7 +158,7 @@ class TestDockerProxy:
).encode()
deps["storage"].exists.return_value = True
deps["storage"].download_object.return_value = manifest
deps["cache"].is_index_file.return_value = True
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")
@@ -170,7 +170,7 @@ class TestDockerProxy:
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_index_file.return_value = False
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)
@@ -185,7 +185,7 @@ class TestDockerProxy:
).encode()
deps["storage"].exists.return_value = True
deps["storage"].download_object.return_value = manifest
deps["cache"].is_index_file.return_value = False
deps["cache"].is_mutable_file.return_value = False
response = client.head("/v2/docker-test/library/nginx/manifests/latest")
assert response.status_code == 200
@@ -201,7 +201,7 @@ class TestDockerProxy:
).encode()
deps["storage"].exists.return_value = False
deps["storage"].download_object.return_value = manifest
deps["cache"].is_index_file.return_value = True
deps["cache"].is_mutable_file.return_value = True
with patch(
"artifactapi.main.cache_single_artifact",
@@ -223,7 +223,7 @@ class TestDockerProxy:
).encode()
deps["storage"].exists.return_value = False
deps["storage"].download_object.return_value = manifest
deps["cache"].is_index_file.return_value = True
deps["cache"].is_mutable_file.return_value = True
with patch(
"artifactapi.main.cache_single_artifact",
@@ -244,7 +244,7 @@ class TestDockerProxy:
}
).encode()
deps["storage"].exists.return_value = True # cached in S3
deps["cache"].is_index_file.return_value = True
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
@@ -278,7 +278,7 @@ class TestGenericArtifactRoute:
deps = patched_deps
deps["storage"].exists.return_value = True
deps["storage"].download_object.return_value = b"tar content"
deps["cache"].is_index_file.return_value = False
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
@@ -289,7 +289,7 @@ class TestGenericArtifactRoute:
deps = patched_deps
deps["storage"].exists.return_value = True
deps["storage"].download_object.return_value = b"content"
deps["cache"].is_index_file.return_value = False
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"]
@@ -301,7 +301,7 @@ class TestGenericArtifactRoute:
content = b"some artifact content bytes"
deps["storage"].exists.return_value = True
deps["storage"].download_object.return_value = content
deps["cache"].is_index_file.return_value = False
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))
@@ -310,7 +310,7 @@ class TestGenericArtifactRoute:
deps = patched_deps
deps["storage"].exists.return_value = True
deps["storage"].download_object.return_value = b"content"
deps["cache"].is_index_file.return_value = False
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)
@@ -319,7 +319,7 @@ class TestGenericArtifactRoute:
deps = patched_deps
deps["storage"].exists.return_value = True
deps["storage"].download_object.return_value = b"content"
deps["cache"].is_index_file.return_value = False
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()
@@ -328,7 +328,7 @@ class TestGenericArtifactRoute:
deps = patched_deps
deps["storage"].exists.return_value = True
deps["storage"].download_object.return_value = b"rpm bytes"
deps["cache"].is_index_file.return_value = False
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
@@ -338,7 +338,7 @@ class TestGenericArtifactRoute:
deps = patched_deps
deps["storage"].exists.return_value = True
deps["storage"].download_object.return_value = b"<?xml version='1.0'?>"
deps["cache"].is_index_file.return_value = False
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
@@ -348,7 +348,7 @@ class TestGenericArtifactRoute:
deps = patched_deps
deps["storage"].exists.return_value = False
deps["storage"].download_object.return_value = b"fresh content"
deps["cache"].is_index_file.return_value = False
deps["cache"].is_mutable_file.return_value = False
with patch(
"artifactapi.main.cache_single_artifact",
@@ -365,7 +365,7 @@ class TestGenericArtifactRoute:
deps = patched_deps
deps["storage"].exists.return_value = False
deps["storage"].download_object.return_value = b"fresh content"
deps["cache"].is_index_file.return_value = False
deps["cache"].is_mutable_file.return_value = False
with patch(
"artifactapi.main.cache_single_artifact",
@@ -380,7 +380,7 @@ class TestGenericArtifactRoute:
deps = patched_deps
deps["storage"].exists.return_value = False
deps["storage"].download_object.return_value = b"APKINDEX content"
deps["cache"].is_index_file.return_value = True
deps["cache"].is_mutable_file.return_value = True
with patch(
"artifactapi.main.cache_single_artifact",
@@ -395,7 +395,7 @@ class TestGenericArtifactRoute:
def test_upstream_error_returns_502(self, client, patched_deps):
deps = patched_deps
deps["storage"].exists.return_value = False
deps["cache"].is_index_file.return_value = False
deps["cache"].is_mutable_file.return_value = False
with patch(
"artifactapi.main.cache_single_artifact",
@@ -406,16 +406,16 @@ class TestGenericArtifactRoute:
assert response.status_code == 502
def test_index_file_bypasses_include_patterns(self, client, patched_deps):
"""Index files must be served even when they don't match include_patterns."""
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_index_file.return_value = True
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 include_patterns (.*.apk$),
# but since is_index_file returns True it must be allowed through.
# 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