feat(ci): manifest-aware dependency PRs and cleanup protection
CI / validate (pull_request) Failing after 49s
CI / auto-merge (pull_request) Skipped

Implements: DEVX-165 REQ-1..3

- create_dependency_pr gains --manifest/--verify-container/--source-ref:
  resolves the producer image's OCI digest via the packages API before
  opening the PR, then writes structured manifest fields
  (version/git_ref/image_tag/image_digest/updated_at) in the PR branch
- clean_images gains --protect: pinned versions survive --keep trimming
  so a manifest-referenced digest can't be garbage-collected
- regression tests for digest resolution, manifest helpers, verify
  ordering, and protect filtering
This commit is contained in:
Emil Simeonov
2026-09-19 04:13:11 +02:00
parent 9b90816be4
commit 8cbf148745
7 changed files with 618 additions and 27 deletions
+32
View File
@@ -485,6 +485,38 @@ class TestSelectForDeletion:
to_delete = select_for_deletion(versions, keep=5)
assert len(to_delete) == 0
def test_protect_exempts_version(self) -> None:
"""REQ-2: a manifest-pinned version survives cleanup beyond --keep."""
versions = [
{"version": "0.1.0", "created_at": "2025-01-01"},
{"version": "0.2.0", "created_at": "2025-02-01"},
{"version": "0.3.0", "created_at": "2025-03-01"},
{"version": "0.4.0", "created_at": "2025-04-01"},
]
to_delete = select_for_deletion(versions, keep=2, protect=frozenset({"0.1.0"}))
deleted = {v["version"] for v in to_delete}
assert "0.1.0" not in deleted
assert deleted == {"0.2.0"}
def test_protect_multiple_versions(self) -> None:
versions = [
{"version": "0.1.0", "created_at": "2025-01-01"},
{"version": "0.2.0", "created_at": "2025-02-01"},
{"version": "0.3.0", "created_at": "2025-03-01"},
{"version": "0.4.0", "created_at": "2025-04-01"},
]
to_delete = select_for_deletion(versions, keep=1, protect=frozenset({"0.1.0", "0.2.0"}))
assert {v["version"] for v in to_delete} == {"0.3.0"}
def test_protect_default_empty_behaves_as_before(self) -> None:
versions = [
{"version": "0.1.0", "created_at": "2025-01-01"},
{"version": "0.2.0", "created_at": "2025-02-01"},
{"version": "0.3.0", "created_at": "2025-03-01"},
]
to_delete = select_for_deletion(versions, keep=2)
assert {v["version"] for v in to_delete} == {"0.1.0"}
class TestCleanImagesAPI:
"""Tests for the clean_images module's API functions."""
+241
View File
@@ -4,6 +4,7 @@ from pathlib import Path
from unittest.mock import MagicMock, patch
import click
import pytest
from click.testing import CliRunner
from devx.ci.create_dependency_pr import (
@@ -180,3 +181,243 @@ class TestCreateVikunjaTask:
mock_client.create_task.return_value = {"identifier": "OBL-INFRA-999"}
result = create_vikunja_task("Test", "desc")
assert result == "OBL-INFRA-999"
class TestResolveContainerDigest:
"""REQ-1: pre-PR artifact verification via the packages API."""
def test_returns_digest_from_manifest_blob(self) -> None:
from devx.ci.create_dependency_pr import resolve_container_digest
mock_resp = MagicMock()
mock_resp.raise_for_status = MagicMock()
mock_resp.json.return_value = [
{"name": "sha256_layer", "sha256": "abc"},
{"name": "manifest.json", "sha256": "deadbeef"},
]
with patch("devx.ci.create_dependency_pr.requests.get", return_value=mock_resp):
digest = resolve_container_digest(
"https://git.example.com/api/v1", "oblachno", "sso-bridge", "0.9.1", "tok"
)
assert digest == "sha256:deadbeef"
def test_raises_when_version_missing(self) -> None:
import requests
from devx.ci.create_dependency_pr import resolve_container_digest
mock_resp = MagicMock()
http_err = requests.HTTPError("404")
http_err.response = MagicMock(status_code=404)
mock_resp.raise_for_status.side_effect = http_err
with patch("devx.ci.create_dependency_pr.requests.get", return_value=mock_resp):
with pytest.raises(click.ClickException, match="unpublished artifact"):
resolve_container_digest("https://git.example.com/api/v1", "oblachno", "sso-bridge", "9.9.9", "tok")
def test_raises_when_no_manifest_blob(self) -> None:
from devx.ci.create_dependency_pr import resolve_container_digest
mock_resp = MagicMock()
mock_resp.raise_for_status = MagicMock()
mock_resp.json.return_value = [{"name": "sha256_layer", "sha256": "abc"}]
with patch("devx.ci.create_dependency_pr.requests.get", return_value=mock_resp):
with pytest.raises(click.ClickException, match="no manifest blob"):
resolve_container_digest("https://git.example.com/api/v1", "oblachno", "sso-bridge", "0.9.1", "tok")
def test_raises_on_connection_error(self) -> None:
import requests
from devx.ci.create_dependency_pr import resolve_container_digest
with patch(
"devx.ci.create_dependency_pr.requests.get",
side_effect=requests.ConnectionError("refused"),
):
with pytest.raises(click.ClickException, match="Registry lookup failed"):
resolve_container_digest("https://git.example.com/api/v1", "oblachno", "sso-bridge", "0.9.1", "tok")
class TestManifestHelpers:
"""REQ-1: manifest read/update helpers."""
def test_read_version(self, tmp_path: Path) -> None:
import json
from devx.ci.create_dependency_pr import read_manifest_version
p = tmp_path / "m.json"
p.write_text(json.dumps({"schema_version": 1, "sso_bridge": {"version": "0.9.0"}}))
assert read_manifest_version(str(p), "sso_bridge") == "0.9.0"
def test_read_version_missing_file(self, tmp_path: Path) -> None:
from devx.ci.create_dependency_pr import read_manifest_version
assert read_manifest_version(str(tmp_path / "nope.json"), "sso_bridge") is None
def test_read_version_bad_json(self, tmp_path: Path) -> None:
from devx.ci.create_dependency_pr import read_manifest_version
p = tmp_path / "m.json"
p.write_text("not json{")
assert read_manifest_version(str(p), "sso_bridge") is None
def test_read_version_missing_section(self, tmp_path: Path) -> None:
import json
from devx.ci.create_dependency_pr import read_manifest_version
p = tmp_path / "m.json"
p.write_text(json.dumps({"schema_version": 1, "other": "x"}))
assert read_manifest_version(str(p), "sso_bridge") is None
def test_update_manifest_fields(self, tmp_path: Path) -> None:
import json
from devx.ci.create_dependency_pr import update_manifest
p = tmp_path / "m.json"
p.write_text(json.dumps({"schema_version": 1, "sso_bridge": {"version": "0.9.0"}}))
changed = update_manifest(
str(p),
"sso_bridge",
{"version": "0.9.1", "git_ref": "v0.9.1", "image_digest": "sha256:x"},
)
assert changed is True
data = json.loads(p.read_text())
assert data["sso_bridge"]["version"] == "0.9.1"
assert data["sso_bridge"]["git_ref"] == "v0.9.1"
assert data["sso_bridge"]["image_digest"] == "sha256:x"
def test_update_manifest_no_change(self, tmp_path: Path) -> None:
import json
from devx.ci.create_dependency_pr import update_manifest
p = tmp_path / "m.json"
p.write_text(json.dumps({"sso_bridge": {"version": "0.9.1"}}))
assert update_manifest(str(p), "sso_bridge", {"version": "0.9.1"}) is False
def test_update_manifest_missing_file(self, tmp_path: Path) -> None:
from devx.ci.create_dependency_pr import update_manifest
with pytest.raises(click.ClickException, match="not found"):
update_manifest(str(tmp_path / "nope.json"), "sso_bridge", {"version": "1"})
def test_update_manifest_bad_json(self, tmp_path: Path) -> None:
from devx.ci.create_dependency_pr import update_manifest
p = tmp_path / "m.json"
p.write_text("broken{")
with pytest.raises(click.ClickException, match="not valid JSON"):
update_manifest(str(p), "sso_bridge", {"version": "1"})
def test_update_manifest_missing_section(self, tmp_path: Path) -> None:
import json
from devx.ci.create_dependency_pr import update_manifest
p = tmp_path / "m.json"
p.write_text(json.dumps({"schema_version": 1}))
with pytest.raises(click.ClickException, match="no object section"):
update_manifest(str(p), "sso_bridge", {"version": "1"})
class TestCliManifestMode:
@patch("devx.ci.create_dependency_pr.read_manifest_version")
@patch("devx.ci.create_dependency_pr.get_ci_token")
def test_manifest_same_version_no_pr(self, mock_token: MagicMock, mock_read: MagicMock) -> None:
mock_token.return_value = "fake-token"
mock_read.return_value = "0.9.1"
runner = CliRunner()
result = runner.invoke(
cli,
[
"--package",
"sso_bridge",
"--new-version",
"0.9.1",
"--source-repo",
"oblachno/sso-bridge",
"--manifest",
"deploy/sso-bridge-release.json",
],
)
assert result.exit_code == 0
assert "no pr needed" in result.output.lower()
@patch("devx.ci.create_dependency_pr.resolve_container_digest")
@patch("devx.ci.create_dependency_pr.read_manifest_version")
@patch("devx.ci.create_dependency_pr.get_ci_token")
def test_verify_container_runs_before_lookup(
self, mock_token: MagicMock, mock_read: MagicMock, mock_digest: MagicMock
) -> None:
"""Verification failure aborts before the version lookup/PR steps."""
mock_token.return_value = "fake-token"
mock_digest.side_effect = click.ClickException("unpublished artifact")
runner = CliRunner()
result = runner.invoke(
cli,
[
"--package",
"sso_bridge",
"--new-version",
"0.9.1",
"--source-repo",
"oblachno/sso-bridge",
"--manifest",
"deploy/m.json",
"--verify-container",
"oblachno/sso-bridge",
],
)
assert result.exit_code != 0
mock_read.assert_not_called()
@patch("devx.ci.create_dependency_pr.resolve_container_digest")
@patch("devx.ci.create_dependency_pr.read_manifest_version")
@patch("devx.ci.create_dependency_pr.get_ci_token")
def test_verify_container_resolves_digest(
self, mock_token: MagicMock, mock_read: MagicMock, mock_digest: MagicMock
) -> None:
mock_token.return_value = "fake-token"
mock_read.return_value = "0.9.1"
mock_digest.return_value = "sha256:abc"
runner = CliRunner()
result = runner.invoke(
cli,
[
"--package",
"sso_bridge",
"--new-version",
"0.9.1",
"--source-repo",
"oblachno/sso-bridge",
"--manifest",
"deploy/m.json",
"--verify-container",
"oblachno/sso-bridge",
],
)
assert result.exit_code == 0
mock_digest.assert_called_once()
args = mock_digest.call_args[0]
assert args[1:4] == ("oblachno", "sso-bridge", "0.9.1")
@patch("devx.ci.create_dependency_pr.get_ci_token")
def test_verify_container_invalid_format(self, mock_token: MagicMock) -> None:
mock_token.return_value = "fake-token"
runner = CliRunner()
result = runner.invoke(
cli,
[
"--package",
"sso_bridge",
"--new-version",
"0.9.1",
"--source-repo",
"oblachno/sso-bridge",
"--verify-container",
"no-slash",
],
)
assert result.exit_code != 0