Files
devx/tests/unit/test_create_dependency_pr.py
T
Emil SimeonovandDevin <158243242+devin-ai-integration[bot]@users.noreply.github.com> 872d757bb5
CI / validate (pull_request) Successful in 1m1s
CI / auto-merge (pull_request) Successful in 12s
fix(ci): create dep-PR branches via POST /branches
Gitea lacks the POST /git/refs endpoint (HTTP 405). Use the branches
API with new_branch_name/old_branch_name; tolerate already-exists.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-09-19 21:10:42 +02:00

560 lines
21 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"
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
@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