Public Access
560 lines
21 KiB
Python
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
|