Public Access
Emil Simeonov
48e5dd5a34
The packages-API manifest.json blob sha256 is not pullable via repo@sha256:... — production deploy 6251 failed pulling the recorded digest (404 not found). Existence check stays on the packages API (keeps the 404 retry), but the digest now comes from the registry v2 Docker-Content-Digest header. Implements DEVX-173 REQ-1/REQ-2.
685 lines
26 KiB
Python
685 lines
26 KiB
Python
"""Unit tests for devx.ci.create_dependency_pr."""
|
|
|
|
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 (
|
|
cli,
|
|
create_vikunja_task,
|
|
find_existing_pr,
|
|
find_pinned_version,
|
|
update_pinned_version,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _mock_subprocess():
|
|
"""Mock subprocess so CLI tests never run a real git clone."""
|
|
with patch("devx.ci.create_dependency_pr.subprocess.run") as m:
|
|
yield m
|
|
|
|
|
|
class TestFindPinnedVersion:
|
|
def test_finds_pip_git_pin(self, tmp_path: Path) -> None:
|
|
content = "grm @ git+https://git.example.com/repo.git@v0.5.1"
|
|
path = tmp_path / "pyproject.toml"
|
|
path.write_text(content)
|
|
version = find_pinned_version("grm", str(path))
|
|
assert version == "0.5.1"
|
|
|
|
def test_finds_pyproject_pin(self, tmp_path: Path) -> None:
|
|
content = 'grm = "0.5.1"'
|
|
path = tmp_path / "pyproject.toml"
|
|
path.write_text(content)
|
|
version = find_pinned_version("grm", str(path))
|
|
assert version == "0.5.1"
|
|
|
|
def test_finds_ansible_var_pin(self, tmp_path: Path) -> None:
|
|
content = 'grm_version: "0.5.1"'
|
|
path = tmp_path / "images.yml"
|
|
path.write_text(content)
|
|
version = find_pinned_version("grm", str(path))
|
|
assert version == "0.5.1"
|
|
|
|
def test_finds_image_version_pin(self, tmp_path: Path) -> None:
|
|
content = 'sso_bridge_image_version: "1.2.3"'
|
|
path = tmp_path / "images.yml"
|
|
path.write_text(content)
|
|
version = find_pinned_version("sso_bridge", str(path))
|
|
assert version == "1.2.3"
|
|
|
|
def test_returns_none_when_not_found(self, tmp_path: Path) -> None:
|
|
path = tmp_path / "pyproject.toml"
|
|
path.write_text('other = "1.0.0"')
|
|
assert find_pinned_version("grm", str(path)) is None
|
|
|
|
def test_returns_none_when_file_missing(self, tmp_path: Path) -> None:
|
|
assert find_pinned_version("grm", str(tmp_path / "nonexistent.toml")) is None
|
|
|
|
|
|
class TestUpdatePinnedVersion:
|
|
def test_updates_pip_git_pin(self, tmp_path: Path) -> None:
|
|
content = "grm @ git+https://git.example.com/repo.git@v0.5.1"
|
|
path = tmp_path / "pyproject.toml"
|
|
path.write_text(content)
|
|
changed = update_pinned_version(str(path), "grm", "0.5.1", "0.5.2")
|
|
assert changed is True
|
|
assert "0.5.2" in path.read_text()
|
|
assert "0.5.1" not in path.read_text()
|
|
|
|
def test_updates_pyproject_pin(self, tmp_path: Path) -> None:
|
|
content = 'grm = "0.5.1"'
|
|
path = tmp_path / "pyproject.toml"
|
|
path.write_text(content)
|
|
changed = update_pinned_version(str(path), "grm", "0.5.1", "0.5.2")
|
|
assert changed is True
|
|
assert 'grm = "0.5.2"' in path.read_text()
|
|
|
|
def test_no_change_when_version_not_found(self, tmp_path: Path) -> None:
|
|
content = 'other = "1.0.0"'
|
|
path = tmp_path / "pyproject.toml"
|
|
path.write_text(content)
|
|
changed = update_pinned_version(str(path), "grm", "0.5.1", "0.5.2")
|
|
assert changed is False
|
|
|
|
def test_no_change_when_file_missing(self, tmp_path: Path) -> None:
|
|
changed = update_pinned_version(str(tmp_path / "nonexistent"), "grm", "0.5.1", "0.5.2")
|
|
assert changed is False
|
|
|
|
|
|
class TestFindExistingPr:
|
|
@patch("devx.tools.create_pr.GiteaClient")
|
|
def test_returns_pr_when_found(self, mock_client_cls: MagicMock) -> None:
|
|
mock_client = mock_client_cls.return_value
|
|
mock_client.list_prs.return_value = [
|
|
{"head": {"ref": "deps/grm-0.5.2"}, "number": 42},
|
|
{"head": {"ref": "other-branch"}, "number": 43},
|
|
]
|
|
result = find_existing_pr(mock_client, "deps/grm-0.5.2")
|
|
assert result is not None
|
|
assert result["number"] == 42
|
|
|
|
@patch("devx.tools.create_pr.GiteaClient")
|
|
def test_returns_none_when_not_found(self, mock_client_cls: MagicMock) -> None:
|
|
mock_client = mock_client_cls.return_value
|
|
mock_client.list_prs.return_value = []
|
|
result = find_existing_pr(mock_client, "deps/grm-0.5.2")
|
|
assert result is None
|
|
|
|
|
|
class TestCli:
|
|
@patch("devx.ci.create_dependency_pr.find_pinned_version")
|
|
@patch("devx.ci.create_dependency_pr.get_ci_token")
|
|
def test_same_version_no_pr(self, mock_token: MagicMock, mock_find: MagicMock) -> None:
|
|
mock_token.return_value = "fake-token"
|
|
mock_find.return_value = "0.5.2"
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"--package",
|
|
"grm",
|
|
"--new-version",
|
|
"0.5.2",
|
|
"--source-repo",
|
|
"oblachno/grm",
|
|
],
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "no pr needed" in result.output.lower()
|
|
|
|
@patch("devx.ci.create_dependency_pr.find_pinned_version")
|
|
@patch("devx.ci.create_dependency_pr.get_ci_token")
|
|
def test_dry_run(self, mock_token: MagicMock, mock_find: MagicMock) -> None:
|
|
mock_token.return_value = "fake-token"
|
|
mock_find.return_value = "0.5.1"
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"--package",
|
|
"grm",
|
|
"--new-version",
|
|
"0.5.2",
|
|
"--source-repo",
|
|
"oblachno/grm",
|
|
"--dry-run",
|
|
],
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "DRY RUN" in result.output
|
|
|
|
@patch("devx.ci.create_dependency_pr.find_pinned_version")
|
|
@patch("devx.ci.create_dependency_pr.get_ci_token")
|
|
def test_version_not_found_fails(self, mock_token: MagicMock, mock_find: MagicMock) -> None:
|
|
mock_token.return_value = "fake-token"
|
|
mock_find.return_value = None
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"--package",
|
|
"nonexistent",
|
|
"--new-version",
|
|
"1.0.0",
|
|
"--source-repo",
|
|
"oblachno/test",
|
|
],
|
|
)
|
|
assert result.exit_code != 0
|
|
|
|
|
|
class TestCreateVikunjaTask:
|
|
def test_returns_none_when_no_token(self) -> None:
|
|
with patch("devx.ci.create_dependency_pr.get_vikunja_token", side_effect=click.ClickException("no token")):
|
|
result = create_vikunja_task("Test", "desc")
|
|
assert result is None
|
|
|
|
def test_returns_identifier_on_success(self) -> None:
|
|
with (
|
|
patch("devx.ci.create_dependency_pr.get_vikunja_token", return_value="fake-token"),
|
|
patch("devx.api_clients.VikunjaClient") as mock_client_cls,
|
|
):
|
|
mock_client = mock_client_cls.return_value
|
|
mock_client.create_task.return_value = {"identifier": "OBL-INFRA-999"}
|
|
result = create_vikunja_task("Test", "desc")
|
|
assert result == "OBL-INFRA-999"
|
|
|
|
def test_explicit_project_id_used(self) -> None:
|
|
with (
|
|
patch("devx.ci.create_dependency_pr.get_vikunja_token", return_value="fake-token"),
|
|
patch("devx.api_clients.VikunjaClient") as mock_client_cls,
|
|
):
|
|
mock_client = mock_client_cls.return_value
|
|
mock_client.create_task.return_value = {"identifier": "OBL-INFRA-1"}
|
|
create_vikunja_task("Test", "desc", project_id=3)
|
|
assert mock_client.create_task.call_args.args[0] == 3
|
|
|
|
|
|
def _v2_mocks(digest: str = "sha256:deadbeef") -> list[MagicMock]:
|
|
"""Token + manifest responses for the registry v2 digest lookup."""
|
|
tok = MagicMock()
|
|
tok.raise_for_status = MagicMock()
|
|
tok.json.return_value = {"token": "bearer-tok"}
|
|
man = MagicMock()
|
|
man.raise_for_status = MagicMock()
|
|
man.headers = {"Docker-Content-Digest": digest}
|
|
return [tok, man]
|
|
|
|
|
|
class TestResolveContainerDigest:
|
|
"""REQ-1: existence check via packages API; digest via registry v2."""
|
|
|
|
def test_returns_digest_from_registry_v2(self) -> None:
|
|
from devx.ci.create_dependency_pr import resolve_container_digest
|
|
|
|
files = MagicMock()
|
|
files.raise_for_status = MagicMock()
|
|
files.json.return_value = [{"name": "manifest.json", "sha256": "blob-not-pullable"}]
|
|
with patch(
|
|
"devx.ci.create_dependency_pr.requests.get",
|
|
side_effect=[files, *_v2_mocks()],
|
|
) as mock_get:
|
|
digest = resolve_container_digest(
|
|
"https://git.example.com/api/v1", "oblachno", "sso-bridge", "0.9.1", "tok"
|
|
)
|
|
assert digest == "sha256:deadbeef"
|
|
man_call = mock_get.call_args_list[2]
|
|
assert "manifests/0.9.1" in man_call.args[0]
|
|
assert "oci.image.index" in man_call.kwargs["headers"]["Accept"]
|
|
tok_call = mock_get.call_args_list[1]
|
|
assert tok_call.kwargs["params"]["scope"] == "repository:oblachno/sso-bridge:pull"
|
|
|
|
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_v2_digest_missing(self) -> None:
|
|
from devx.ci.create_dependency_pr import resolve_container_digest
|
|
|
|
files = MagicMock()
|
|
files.raise_for_status = MagicMock()
|
|
files.json.return_value = []
|
|
tok, man = _v2_mocks(digest="")
|
|
with patch(
|
|
"devx.ci.create_dependency_pr.requests.get",
|
|
side_effect=[files, tok, man],
|
|
):
|
|
with pytest.raises(click.ClickException, match="no Docker-Content-Digest"):
|
|
resolve_container_digest("https://git.example.com/api/v1", "oblachno", "sso-bridge", "0.9.1", "tok")
|
|
|
|
def test_raises_when_v2_manifest_request_fails(self) -> None:
|
|
import requests
|
|
|
|
from devx.ci.create_dependency_pr import resolve_container_digest
|
|
|
|
files = MagicMock()
|
|
files.raise_for_status = MagicMock()
|
|
files.json.return_value = []
|
|
tok, _ = _v2_mocks()
|
|
err = requests.HTTPError("500")
|
|
err.response = MagicMock(status_code=500)
|
|
man = MagicMock()
|
|
man.raise_for_status.side_effect = err
|
|
with patch(
|
|
"devx.ci.create_dependency_pr.requests.get",
|
|
side_effect=[files, tok, man],
|
|
):
|
|
with pytest.raises(click.ClickException, match="v2 digest lookup failed"):
|
|
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
|
|
|
|
@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_container_tag_overrides_new_version(
|
|
self, mock_token: MagicMock, mock_read: MagicMock, mock_digest: MagicMock
|
|
) -> None:
|
|
"""--container-tag selects the image tag when it differs from version."""
|
|
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",
|
|
"--container-tag",
|
|
"0.2.4",
|
|
],
|
|
)
|
|
assert result.exit_code == 0
|
|
args = mock_digest.call_args[0]
|
|
assert args[1:4] == ("oblachno", "sso-bridge", "0.2.4")
|
|
|
|
|
|
class TestBranchCreation:
|
|
"""Branch creation uses the branches API (Gitea lacks POST /git/refs)."""
|
|
|
|
def _invoke(self) -> object:
|
|
runner = CliRunner()
|
|
return runner.invoke(
|
|
cli,
|
|
[
|
|
"--package",
|
|
"sso_bridge",
|
|
"--new-version",
|
|
"0.9.1",
|
|
"--source-repo",
|
|
"oblachno/sso-bridge",
|
|
"--manifest",
|
|
"deploy/m.json",
|
|
],
|
|
)
|
|
|
|
def _client(self, mock_client_cls: MagicMock) -> MagicMock:
|
|
client = mock_client_cls.return_value
|
|
client.create_pr.return_value = {"number": 1}
|
|
return client
|
|
|
|
@patch("devx.ci.create_dependency_pr.create_vikunja_task", return_value=None)
|
|
@patch("devx.ci.create_dependency_pr.update_manifest", return_value=True)
|
|
@patch("devx.ci.create_dependency_pr.read_manifest_version", return_value="0.9.0")
|
|
@patch("devx.ci.create_dependency_pr.find_existing_pr", return_value=None)
|
|
@patch("devx.ci.create_dependency_pr.GiteaClient")
|
|
@patch("devx.ci.create_dependency_pr.get_ci_token", return_value="tok")
|
|
def test_creates_branch_via_branches_api(
|
|
self,
|
|
_token: MagicMock,
|
|
mock_client_cls: MagicMock,
|
|
_find: MagicMock,
|
|
_read: MagicMock,
|
|
_update: MagicMock,
|
|
_task: MagicMock,
|
|
) -> None:
|
|
client = self._client(mock_client_cls)
|
|
result = self._invoke()
|
|
assert result.exit_code == 0
|
|
assert "Created PR" in result.output
|
|
post = client._request.call_args
|
|
assert post.args[:2] == ("POST", "/branches")
|
|
assert post.kwargs["json"] == {
|
|
"new_branch_name": "deps/sso_bridge-0.9.1",
|
|
"old_branch_name": "master",
|
|
}
|
|
|
|
@patch("devx.ci.create_dependency_pr.create_vikunja_task", return_value=None)
|
|
@patch("devx.ci.create_dependency_pr.update_manifest", return_value=True)
|
|
@patch("devx.ci.create_dependency_pr.read_manifest_version", return_value="0.9.0")
|
|
@patch("devx.ci.create_dependency_pr.find_existing_pr", return_value=None)
|
|
@patch("devx.ci.create_dependency_pr.GiteaClient")
|
|
@patch("devx.ci.create_dependency_pr.get_ci_token", return_value="tok")
|
|
def test_existing_branch_tolerated(
|
|
self,
|
|
_token: MagicMock,
|
|
mock_client_cls: MagicMock,
|
|
_find: MagicMock,
|
|
_read: MagicMock,
|
|
_update: MagicMock,
|
|
_task: MagicMock,
|
|
) -> None:
|
|
from devx.exceptions import APIError
|
|
|
|
client = self._client(mock_client_cls)
|
|
client._request.side_effect = APIError(422, "branch already exists")
|
|
result = self._invoke()
|
|
assert result.exit_code == 0
|
|
assert "already exists" in result.output.lower()
|
|
|
|
@patch("devx.ci.create_dependency_pr.create_vikunja_task", return_value=None)
|
|
@patch("devx.ci.create_dependency_pr.update_manifest", return_value=True)
|
|
@patch("devx.ci.create_dependency_pr.read_manifest_version", return_value="0.9.0")
|
|
@patch("devx.ci.create_dependency_pr.find_existing_pr", return_value=None)
|
|
@patch("devx.ci.create_dependency_pr.GiteaClient")
|
|
@patch("devx.ci.create_dependency_pr.get_ci_token", return_value="tok")
|
|
def test_branch_api_error_fails(
|
|
self,
|
|
_token: MagicMock,
|
|
mock_client_cls: MagicMock,
|
|
_find: MagicMock,
|
|
_read: MagicMock,
|
|
_update: MagicMock,
|
|
_task: MagicMock,
|
|
) -> None:
|
|
from devx.exceptions import APIError
|
|
|
|
client = self._client(mock_client_cls)
|
|
client._request.side_effect = APIError(500, "boom")
|
|
result = self._invoke()
|
|
assert result.exit_code != 0
|
|
assert "Failed to create branch" in result.output
|
|
|
|
@patch("devx.ci.create_dependency_pr.create_vikunja_task", return_value=None)
|
|
@patch("devx.ci.create_dependency_pr.update_manifest", return_value=True)
|
|
@patch("devx.ci.create_dependency_pr.read_manifest_version", return_value="0.9.0")
|
|
@patch("devx.ci.create_dependency_pr.find_existing_pr", return_value=None)
|
|
@patch("devx.ci.create_dependency_pr.GiteaClient")
|
|
@patch("devx.ci.create_dependency_pr.get_ci_token", return_value="tok")
|
|
def test_checks_out_remote_tracking_branch(
|
|
self,
|
|
_token: MagicMock,
|
|
mock_client_cls: MagicMock,
|
|
_find: MagicMock,
|
|
_read: MagicMock,
|
|
_update: MagicMock,
|
|
_task: MagicMock,
|
|
) -> None:
|
|
import subprocess
|
|
|
|
self._client(mock_client_cls)
|
|
result = self._invoke()
|
|
assert result.exit_code == 0
|
|
calls = [c.args[0] for c in subprocess.run.call_args_list]
|
|
assert ["git", "fetch", "origin", "deps/sso_bridge-0.9.1:refs/remotes/origin/deps/sso_bridge-0.9.1"] in calls
|
|
assert ["git", "checkout", "-B", "deps/sso_bridge-0.9.1", "origin/deps/sso_bridge-0.9.1"] in calls
|
|
|
|
@patch("devx.ci.create_dependency_pr.create_vikunja_task", return_value="OBL-INFRA-581")
|
|
@patch("devx.ci.create_dependency_pr.update_manifest", return_value=True)
|
|
@patch("devx.ci.create_dependency_pr.read_manifest_version", return_value="0.9.0")
|
|
@patch("devx.ci.create_dependency_pr.find_existing_pr", return_value=None)
|
|
@patch("devx.ci.create_dependency_pr.GiteaClient")
|
|
@patch("devx.ci.create_dependency_pr.get_ci_token", return_value="tok")
|
|
def test_task_id_in_branch_and_spec_committed(
|
|
self,
|
|
_token: MagicMock,
|
|
mock_client_cls: MagicMock,
|
|
_find: MagicMock,
|
|
_read: MagicMock,
|
|
_update: MagicMock,
|
|
mock_task: MagicMock,
|
|
) -> None:
|
|
import subprocess
|
|
|
|
client = self._client(mock_client_cls)
|
|
result = self._invoke()
|
|
assert result.exit_code == 0
|
|
# Branch embeds the target-repo task ID
|
|
post = client._request.call_args
|
|
assert post.kwargs["json"]["new_branch_name"] == "deps/OBL-INFRA-581-sso_bridge-0.9.1"
|
|
# Spec file added alongside the manifest change
|
|
add = next(c for c in subprocess.run.call_args_list if c.args[0][:2] == ["git", "add"])
|
|
assert "docs/specs/OBL-INFRA-581.md" in add.args[0]
|
|
# PR title carries the task ID
|
|
assert client.create_pr.call_args.kwargs["title"] == "OBL-INFRA-581: Bump sso_bridge to 0.9.1"
|
|
mock_task.assert_called_once()
|
|
|
|
def test_retries_404_until_published(self) -> None:
|
|
import requests
|
|
|
|
from devx.ci.create_dependency_pr import resolve_container_digest
|
|
|
|
err = requests.HTTPError("404")
|
|
err.response = MagicMock(status_code=404)
|
|
fail = MagicMock()
|
|
fail.raise_for_status.side_effect = err
|
|
ok = MagicMock()
|
|
ok.raise_for_status = MagicMock()
|
|
ok.json.return_value = [{"name": "manifest.json", "sha256": "cafe"}]
|
|
with (
|
|
patch("devx.ci.create_dependency_pr.requests.get", side_effect=[fail, ok, *_v2_mocks("sha256:cafe")]),
|
|
patch("devx.ci.create_dependency_pr.time.sleep"),
|
|
):
|
|
digest = resolve_container_digest(
|
|
"https://git.example.com/api/v1", "oblachno", "sso-bridge", "0.9.1", "tok", timeout_s=60
|
|
)
|
|
assert digest == "sha256:cafe"
|