DEVX-56: feat: add --force flag to classify_changes, fix api_clients coverage
Post-merge / detect-type (push) Successful in 7s
Post-merge / validate-commit-msg (push) Successful in 11s
Post-merge / configure-repo (push) Successful in 13s
Post-merge / release (push) Successful in 1m5s
Post-merge / vikunja (push) Successful in 32s
Post-merge / badges (push) Successful in 58s
Post-merge / sync-wiki (push) Successful in 1m18s

This commit was merged in pull request #92.
This commit is contained in:
2026-06-25 22:56:31 +00:00
parent 904812dfae
commit 8450f33e88
4 changed files with 164 additions and 1 deletions
+34
View File
@@ -192,6 +192,17 @@ class GiteaClient:
r = self._request("GET", f"/pulls/{pr_number}")
return r.json()
def list_prs(self, state: str = "all", **params: Any) -> list[dict[str, Any]]:
"""List pull requests, optionally filtered by state.
Args:
state: ``open``, ``closed``, ``all`` (default).
**params: Additional query params (e.g. ``q="keyword"`` for title search).
"""
params.setdefault("state", state)
r = self._request("GET", "/pulls", params=params)
return r.json()
def get_pr_files(self, pr_number: str | int) -> list[dict[str, Any]]:
"""Fetch the list of files changed in a pull request."""
r = self._request("GET", f"/pulls/{pr_number}/files")
@@ -345,5 +356,28 @@ class VikunjaClient:
def post_comment(self, task_id: int, comment: str) -> None:
self._request("PUT", f"/tasks/{task_id}/comments", json={"comment": comment})
def list_comments(self, task_id: int) -> list[dict[str, Any]]:
"""List all comments on a task."""
r = self._request("GET", f"/tasks/{task_id}/comments")
return r.json()
def update_task(self, task_id: int, **fields: Any) -> None:
"""Update task fields via POST (full replacement semantics).
Warning: Vikunja's POST /tasks/{id} replaces the entire task body.
Unspecified fields are reset to their type defaults. Use
``update_task_safe`` to preserve existing fields.
"""
self._request("POST", f"/tasks/{task_id}", json=fields)
def update_task_safe(self, task_id: int, **fields: Any) -> dict[str, Any]:
"""Safely update task fields using read-merge-write pattern.
Fetches the full task body, merges the provided fields on top,
and POSTs the complete body back. This prevents accidental
resets of done status, title, etc.
"""
task = self.get_task(task_id)
task.update(fields)
r = self._request("POST", f"/tasks/{task_id}", json=task)
return r.json()
+15 -1
View File
@@ -615,11 +615,25 @@ def _write_github_output(key: str, value: str) -> None:
help="Write results to $GITHUB_OUTPUT file (for CI workflow steps). "
"Outputs 'user-facing-changed' and '<tag>-changed' for each configured tag.",
)
def main(base: str | None, head: str, quiet: bool, check: str, github_output: bool) -> None:
@click.option(
"--force",
is_flag=True,
default=False,
help="Force user-facing-changed=true regardless of actual changes. "
"Used by workflow_dispatch with force-deploy input.",
)
def main(base: str | None, head: str, quiet: bool, check: str, github_output: bool, force: bool) -> None:
"""Classify git changes and output results."""
classifier = _get_classifier()
available_tags = list(classifier.config.tags.keys())
if force and github_output:
_write_github_output("user-facing-changed", "true")
for tag in available_tags:
_write_github_output(f"{tag}-changed", "true")
click.echo("Forced user-facing-changed=true via --force flag.")
return
if base is None:
base = get_latest_tag()
if not base:
+69
View File
@@ -278,6 +278,35 @@ class TestGiteaClient:
timeout=DEFAULT_TIMEOUT,
)
def test_list_prs(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(
return_value=_mock_response([{"number": 1, "title": "feat: add"}, {"number": 2, "title": "fix: bug"}])
)
result = client.list_prs()
assert len(result) == 2
assert result[0]["number"] == 1
client._session.request.assert_called_once_with(
"GET",
"https://git.example.com/repos/owner/repo/pulls",
params={"state": "all"},
timeout=DEFAULT_TIMEOUT,
)
def test_list_prs_with_params(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response([{"number": 3, "title": "docs: update"}]))
result = client.list_prs(state="closed", q="docs")
assert len(result) == 1
client._session.request.assert_called_once_with(
"GET",
"https://git.example.com/repos/owner/repo/pulls",
params={"state": "closed", "q": "docs"},
timeout=DEFAULT_TIMEOUT,
)
def test_get_pr_reviews(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response([{"id": 1, "state": "APPROVED"}]))
@@ -566,6 +595,46 @@ class TestVikunjaClient:
json={"done": True},
)
def test_list_comments(self) -> None:
client = VikunjaClient("https://work.example.com", "tok")
client._session.request = MagicMock(
return_value=_mock_response([{"id": 1, "comment": "first"}, {"id": 2, "comment": "second"}])
)
result = client.list_comments(42)
assert len(result) == 2
assert result[0]["comment"] == "first"
client._session.request.assert_called_once_with(
"GET",
"https://work.example.com/tasks/42/comments",
timeout=DEFAULT_TIMEOUT,
)
def test_update_task_safe(self) -> None:
client = VikunjaClient("https://work.example.com", "tok")
client._session.request = MagicMock(
side_effect=[
_mock_response({"id": 42, "title": "My task", "done": False}),
_mock_response({"id": 42, "title": "My task", "done": True}),
]
)
result = client.update_task_safe(42, done=True)
assert result["done"] is True
assert result["title"] == "My task"
assert client._session.request.call_count == 2
client._session.request.assert_any_call(
"GET",
"https://work.example.com/tasks/42",
timeout=DEFAULT_TIMEOUT,
)
client._session.request.assert_any_call(
"POST",
"https://work.example.com/tasks/42",
timeout=DEFAULT_TIMEOUT,
json={"id": 42, "title": "My task", "done": True},
)
@patch("devx.api_clients.time.sleep")
def test_http_error_raises_api_error(self, mock_sleep: MagicMock) -> None:
client = VikunjaClient("https://work.example.com", "tok")
+46
View File
@@ -742,3 +742,49 @@ class TestGithubOutput:
assert "user-facing-changed=true" in content
# No tag outputs since no tags are configured
assert "ansible-changed" not in content
@patch("devx.ci.classify_changes._get_classifier")
def test_force_outputs_true(self, mock_clf: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""--force with --github-output writes user-facing-changed=true and all tags true."""
mock_clf.return_value = self._make_classifier_with_ansible()
gh_file = tmp_path / "output.txt"
monkeypatch.setenv("GITHUB_OUTPUT", str(gh_file))
runner = CliRunner()
result = runner.invoke(main, ["--github-output", "--force"])
assert result.exit_code == 0
content = gh_file.read_text()
assert "user-facing-changed=true" in content
assert "ansible-changed=true" in content
assert "Forced user-facing-changed=true" in result.output
@patch("devx.ci.classify_changes._get_classifier")
def test_force_without_github_output_does_nothing(
self, mock_clf: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""--force without --github-output falls through to normal classification."""
mock_clf.return_value = self._make_classifier_with_ansible()
monkeypatch.setenv("GITHUB_OUTPUT", str(tmp_path / "output.txt"))
with patch.object(classify_changes_mod, "get_latest_tag", return_value="v1.0"):
with patch.object(classify_changes_mod, "get_changed_files", return_value=[]):
runner = CliRunner()
result = runner.invoke(main, ["--force", "--quiet"])
assert result.exit_code == 0
assert result.output.strip() == "false"
@patch("devx.ci.classify_changes._get_classifier")
def test_force_no_tags(self, mock_clf: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""--force with --github-output and no tags writes only user-facing-changed=true."""
mock_clf.return_value = ChangeClassifier(
ClassifierConfig(
infrastructure=[".gitea/**"],
tags={},
)
)
gh_file = tmp_path / "output.txt"
monkeypatch.setenv("GITHUB_OUTPUT", str(gh_file))
runner = CliRunner()
result = runner.invoke(main, ["--github-output", "--force"])
assert result.exit_code == 0
content = gh_file.read_text()
assert "user-facing-changed=true" in content
assert "ansible-changed" not in content