Compare commits

...
9 Commits
Author SHA1 Message Date
devx-ci-bot 8bb1813715 release: v0.26.0 [skip ci] 2026-06-28 01:59:17 +00:00
emil f7f53941a1 DEVX-92: feat: add distribute_items CI tool for parallel VM deployment
Post-merge / detect-type (push) Successful in 9s
Post-merge / validate-commit-msg (push) Successful in 8s
Post-merge / vikunja (push) Successful in 12s
Post-merge / sync-wiki (push) Successful in 17s
Post-merge / configure-repo (push) Successful in 9s
Post-merge / release (push) Successful in 25s
Build Images / detect-type (push) Successful in 41s
Post-merge / badges (push) Successful in 28s
Post-merge / publish (push) Successful in 15s
Build Images / build-and-push (push) Successful in 2m53s
Build Images / cleanup (push) Successful in 7m33s
2026-06-28 01:58:42 +00:00
gitea-actions-bot 5edfdaa7aa chore: update badge URLs to commit 3a6bff69 [skip ci] 2026-06-28 00:18:22 +00:00
devx-ci-bot 893da8ba34 release: v0.25.0 [skip ci] 2026-06-28 00:18:14 +00:00
emil 64a58874b6 DEVX-91: feat: add manual review support to pr_review (--event, --body, --checklist-confirmed)
Post-merge / detect-type (push) Successful in 8s
Post-merge / validate-commit-msg (push) Successful in 9s
Post-merge / vikunja (push) Successful in 13s
Post-merge / configure-repo (push) Successful in 10s
Post-merge / sync-wiki (push) Successful in 19s
Build Images / detect-type (push) Successful in 34s
Post-merge / release (push) Successful in 27s
Post-merge / badges (push) Successful in 32s
Post-merge / publish (push) Successful in 16s
Build Images / build-and-push (push) Successful in 3m38s
Build Images / cleanup (push) Successful in 7m17s
2026-06-28 00:17:33 +00:00
gitea-actions-bot c1c2041ca4 chore: update badge URLs to commit 0fa6360d [skip ci] 2026-06-28 00:07:16 +00:00
devx-ci-bot 49ff8870b1 release: v0.24.1 [skip ci] 2026-06-28 00:07:07 +00:00
emil 4c1ecbf4fa DEVX-91: refactor: add find_task_by_identifier, config fallbacks for tools
Post-merge / release (push) Successful in 27s
Post-merge / detect-type (push) Successful in 8s
Post-merge / validate-commit-msg (push) Successful in 10s
Post-merge / vikunja (push) Successful in 11s
Post-merge / sync-wiki (push) Successful in 18s
Post-merge / configure-repo (push) Successful in 10s
Build Images / detect-type (push) Successful in 42s
Post-merge / badges (push) Successful in 30s
Post-merge / publish (push) Successful in 15s
Build Images / build-and-push (push) Successful in 2m51s
Build Images / cleanup (push) Successful in 1m44s
2026-06-28 00:06:29 +00:00
gitea-actions-bot 93a1cb9945 chore: update badge URLs to commit 6c0ce9c6 [skip ci] 2026-06-27 23:39:13 +00:00
35 changed files with 1103 additions and 177 deletions
+3 -1
View File
@@ -62,12 +62,13 @@ src/devx/
│ ├── classify_changes.py # User-facing vs workflow-only change detection
│ ├── detect_release_commit.py # Detect release commits on master
│ ├── validate_commit_msg.py # Conventional commit validation
│ ├── pr_review.py # Automated PR review
│ ├── pr_review.py # Automated PR review + manual reviews (--event, --body, --checklist-confirmed)
│ ├── post_merge.py # Vikunja task updates after merge
│ ├── sync_wiki.py # Sync documentation to Gitea wiki
│ ├── push_badges.py # Generate and push quality badges (--retries for retry on git push failures)
│ ├── notify_failure.py # Create Gitea issues on CI failures (--auto-login)
│ ├── distribute_files.py # Distribute files across parallel runners (LPT scheduling)
│ ├── distribute_items.py # Distribute generic items (VMs, hosts) across parallel runners (LPT)
│ ├── integration_guard.py # Run pytest with cross-runner fail-fast
│ ├── check_translations.py # Translation completeness check
│ └── doc_coverage.py # Documentation coverage check
@@ -406,6 +407,7 @@ projects.
| `devx-pr-status` | Check CI status for a PR (`PR=`, `WAIT=`, `TIMEOUT=`) |
| `devx-pr-logs` | Fetch logs for failed CI jobs (`PR=`, `JOB=`, `TAIL=`) |
| `devx-pr-label` | Add a label to a PR (`PR=`, `LABEL=ready-to-merge`) |
| `devx-pr-review` | Post a review on a PR (`PR=`, `EVENT=`, `BODY=`, `CHECKLIST=`) |
| `devx-check-config` | Validate devx configuration |
| `devx-configure-gitea-pypi` | Configure Gitea private PyPI registry |
| `devx-env` | Create .env from .env.example |
+18
View File
@@ -2,6 +2,24 @@
All notable changes to this project will be documented in this file.
## [0.26.0] - 2026-06-28
### Features
- Add distribute_items CI tool for parallel VM deployment
## [0.25.0] - 2026-06-28
### Features
- Add manual review support to pr_review (--event, --body, --checklist-confirmed)
## [0.24.1] - 2026-06-28
### Refactor
- Add find_task_by_identifier, config fallbacks for tools
## [0.24.0] - 2026-06-27
### Features
+6 -6
View File
@@ -16,12 +16,12 @@ quality badges.
[![CI](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions/workflows/ci.yml/badge.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![License: GPL-3.0](https://img.shields.io/badge/license-GPL--3.0-blue)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/src/branch/master/LICENSE)
[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/d312f7b79097e12c4239e00e7eb7cd63fa37b8cf/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/d312f7b79097e12c4239e00e7eb7cd63fa37b8cf/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/d312f7b79097e12c4239e00e7eb7cd63fa37b8cf/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/d312f7b79097e12c4239e00e7eb7cd63fa37b8cf/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/d312f7b79097e12c4239e00e7eb7cd63fa37b8cf/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/d312f7b79097e12c4239e00e7eb7cd63fa37b8cf/python.svg)](https://www.python.org/downloads/)
[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/3a6bff698a366dc91551fbe4845501459afa1b54/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/3a6bff698a366dc91551fbe4845501459afa1b54/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/3a6bff698a366dc91551fbe4845501459afa1b54/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/3a6bff698a366dc91551fbe4845501459afa1b54/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/3a6bff698a366dc91551fbe4845501459afa1b54/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/3a6bff698a366dc91551fbe4845501459afa1b54/python.svg)](https://www.python.org/downloads/)
## Why devx?
+6 -6
View File
@@ -12,12 +12,12 @@ project to be reusable across all oblachno-oss repositories.
[![CI](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions/workflows/ci.yml/badge.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![License: GPL-3.0](https://img.shields.io/badge/license-GPL--3.0-blue)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/src/branch/master/LICENSE)
[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/d312f7b79097e12c4239e00e7eb7cd63fa37b8cf/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/d312f7b79097e12c4239e00e7eb7cd63fa37b8cf/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/d312f7b79097e12c4239e00e7eb7cd63fa37b8cf/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/d312f7b79097e12c4239e00e7eb7cd63fa37b8cf/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/d312f7b79097e12c4239e00e7eb7cd63fa37b8cf/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/d312f7b79097e12c4239e00e7eb7cd63fa37b8cf/python.svg)](https://www.python.org/downloads/)
[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/3a6bff698a366dc91551fbe4845501459afa1b54/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/3a6bff698a366dc91551fbe4845501459afa1b54/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/3a6bff698a366dc91551fbe4845501459afa1b54/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/3a6bff698a366dc91551fbe4845501459afa1b54/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/3a6bff698a366dc91551fbe4845501459afa1b54/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/3a6bff698a366dc91551fbe4845501459afa1b54/python.svg)](https://www.python.org/downloads/)
## Overview
+1 -1
View File
@@ -1,3 +1,3 @@
"""devx — reusable development and CI/CD tools for oblachno-oss projects."""
__version__ = "0.24.0"
__version__ = "0.26.0"
+19
View File
@@ -407,6 +407,25 @@ class VikunjaClient:
r = self._request("GET", f"/projects/{project_id}/tasks", params=params)
return r.json()
def find_task_by_identifier(self, project_id: int, identifier: str, per_page: int = 50) -> dict[str, Any] | None:
"""Find a task by its identifier (e.g. ``DEVX-42``) in a project.
Paginates through all tasks in the project. Returns the task dict
or None if not found.
"""
page = 1
while True:
tasks = self.list_project_tasks(project_id, page=page, per_page=per_page)
if not tasks:
break
for t in tasks:
if t.get("identifier") == identifier:
return t
if len(tasks) < per_page:
break
page += 1
return None
def create_task(self, project_id: int, title: str, description: str = "") -> dict[str, Any]:
"""Create a task in a project and return the created task dict.
+15 -5
View File
@@ -174,15 +174,25 @@ def validate_pr_title_matches_vikunja(pr_title: str, task_id: str) -> None:
def extract_conventional_msg(commits: list[dict[str, Any]]) -> str:
"""Extract the conventional commit message from PR commits.
Iterates commits in reverse order (newest first) to find the first
message matching the conventional commit format. Falls back to the
newest commit message if none match.
Picks the highest-priority conventional commit message from the PR.
Priority: feat > fix > refactor > docs > chore > other.
Falls back to the newest commit message if none match.
"""
priority = {"feat": 5, "fix": 4, "refactor": 3, "docs": 2, "chore": 1, "ci": 1, "style": 1, "test": 1}
best_msg = ""
best_score = 0
for commit in reversed(commits):
commit_info = commit.get("commit", {})
message = str(commit_info.get("message", "") if isinstance(commit_info, dict) else "").split("\n")[0]
if CONVENTIONAL_RE.match(message):
return message
m = CONVENTIONAL_RE.match(message)
if m:
prefix = m.group(1).split("(")[0].strip() # e.g. "feat" from "feat(scope)"
score = priority.get(prefix, 0)
if score > best_score:
best_score = score
best_msg = message
if best_msg:
return best_msg
# Fallback: use the newest commit's first line
if commits:
commit_info = commits[-1].get("commit", {})
+3 -3
View File
@@ -29,7 +29,7 @@ import os
import click
import requests
from devx.config import GITEA_API_URL
from devx.config import GITEA_API_URL, REPO_NAME, REPO_OWNER
DEFAULT_MAX_RUNNERS = 3
@@ -151,9 +151,9 @@ def main(
token = os.environ.get("CI_GITEA_TOKEN", "")
if owner is None:
owner = os.environ.get("DEVX_REPO_OWNER", "oblachno-oss")
owner = os.environ.get("DEVX_REPO_OWNER", "") or REPO_OWNER
if repo is None:
repo = os.environ.get("DEVX_REPO_NAME", "devx")
repo = os.environ.get("DEVX_REPO_NAME", "") or REPO_NAME
count = get_runner_count(GITEA_API_URL, token, owner, repo)
indices = generate_indices(count)
+210
View File
@@ -0,0 +1,210 @@
#!/usr/bin/env python3
"""Distribute a list of items across N parallel runners using LPT scheduling.
Generic item distribution for CI matrix jobs. Items are read from a JSON
array on stdin (or from a file via --items-file), sorted for deterministic
ordering, then assigned to *max_runners* groups using LPT (Longest
Processing Time first) scheduling.
Each item is a string (e.g. an Ansible ``--limit`` pattern like
``observability`` or ``infra-314-vm``). Optionally, items can be objects
with ``{"id": "...", "weight": N}`` to provide explicit weights.
The assigned group for *runner_index* is written to ``$GITHUB_ENV`` as
``ASSIGNED_ITEMS`` (space-delimited) for use by subsequent steps.
Usage::
echo '["observability", "infra-314-vm"]' | \\
python3 -m devx.ci.distribute_items \\
--runner-index 1 --max-runners 3 \\
--github-env --skip-if-excess
# With weights:
echo '[{"id": "observability", "weight": 5}, {"id": "customer-1", "weight": 3}]' | \\
python3 -m devx.ci.distribute_items \\
--runner-index 1 --max-runners 3 --github-env
"""
from __future__ import annotations
import json
import os
import sys
import click
from devx.i18n import _
DEFAULT_MAX_RUNNERS = 3
DEFAULT_WEIGHT = 1
def parse_items(raw: str) -> list[str]:
"""Parse a JSON array into a list of item identifier strings.
Accepts both plain string arrays (``["a", "b"]``) and object arrays
(``[{"id": "a", "weight": 2}]``). Returns just the identifier strings.
"""
data = json.loads(raw)
if not isinstance(data, list):
raise click.ClickException(_("Items input must be a JSON array, got {type}", type=type(data).__name__))
items: list[str] = []
for entry in data:
if isinstance(entry, str):
items.append(entry)
elif isinstance(entry, dict) and "id" in entry:
items.append(str(entry["id"]))
else:
raise click.ClickException(
_("Each item must be a string or an object with 'id', got {type}", type=type(entry).__name__)
)
return items
def parse_weighted_items(raw: str) -> tuple[list[str], list[int]]:
"""Parse a JSON array into (items, weights) lists.
For plain string arrays, all items get ``DEFAULT_WEIGHT``.
For object arrays, the ``weight`` field is used (default: ``DEFAULT_WEIGHT``).
"""
data = json.loads(raw)
if not isinstance(data, list):
raise click.ClickException(_("Items input must be a JSON array, got {type}", type=type(data).__name__))
items: list[str] = []
weights: list[int] = []
for entry in data:
if isinstance(entry, str):
items.append(entry)
weights.append(DEFAULT_WEIGHT)
elif isinstance(entry, dict) and "id" in entry:
items.append(str(entry["id"]))
weights.append(int(entry.get("weight", DEFAULT_WEIGHT)))
else:
raise click.ClickException(
_("Each item must be a string or an object with 'id', got {type}", type=type(entry).__name__)
)
return items, weights
def distribute(items: list[str], weights: list[int], max_runners: int) -> list[list[str]]:
"""Split *items* into *max_runners* balanced groups using LPT scheduling.
Items are sorted by weight (descending), then assigned to the runner
with the least total weight.
"""
groups: list[list[str]] = [[] for _ in range(max_runners)]
loads = [0] * max_runners
indexed = sorted(enumerate(items), key=lambda x: (-weights[x[0]], x[0]))
for orig_idx, item in indexed:
min_runner = min(range(max_runners), key=lambda r: loads[r])
groups[min_runner].append(item)
loads[min_runner] += weights[orig_idx]
return groups
def items_for_runner(items: list[str], weights: list[int], runner_index: int, max_runners: int) -> list[str]:
"""Return the subset of items assigned to *runner_index* (0-based)."""
groups = distribute(items, weights, max_runners)
if runner_index < 0 or runner_index >= len(groups):
raise click.ClickException(
_("Runner index {index} out of range (0..{max})", index=runner_index, max=max_runners - 1)
)
return groups[runner_index]
def _write_github_env(key: str, value: str) -> None:
gh_env = os.environ.get("GITHUB_ENV")
if not gh_env:
raise click.ClickException("GITHUB_ENV environment variable is not set")
with open(gh_env, "a") as f: # noqa: PTH123
if "\n" in value:
delimiter = "EOF"
f.write(f"{key}<<{delimiter}\n{value}\n{delimiter}\n")
else:
f.write(f"{key}={value}\n")
@click.command()
@click.option(
"--items-file",
type=click.Path(exists=True, file_okay=True, path_type=None),
default=None,
help="Read items from a JSON file instead of stdin.",
)
@click.option(
"--runner-index",
type=int,
default=None,
help="One-based runner index. If omitted, prints all groups.",
)
@click.option(
"--max-runners",
type=int,
default=DEFAULT_MAX_RUNNERS,
show_default=True,
help="Total number of parallel runners.",
)
@click.option(
"--github-env",
is_flag=True,
default=False,
help="Write ASSIGNED_ITEMS and SKIP to $GITHUB_ENV.",
)
@click.option(
"--skip-if-excess",
is_flag=True,
default=False,
help="With --github-env: write SKIP=true when runner-index exceeds max-runners.",
)
def main(
items_file: str | None,
runner_index: int | None,
max_runners: int,
github_env: bool,
skip_if_excess: bool,
) -> None:
# Read items from file or stdin
if items_file is not None:
with open(items_file) as f: # noqa: PTH123
raw = f.read()
else:
raw = sys.stdin.read()
raw = raw.strip()
if not raw:
raw = "[]"
items, weights = parse_weighted_items(raw)
if runner_index is None:
groups = distribute(items, weights, max_runners)
for i, group in enumerate(groups):
labels = " ".join(group) if group else "(none)"
click.echo(f"Runner {i}: {labels}")
return
if skip_if_excess and github_env and runner_index > max_runners:
click.echo(f"Skipping — runner index {runner_index} > max runners {max_runners}")
_write_github_env("ASSIGNED_ITEMS", "")
_write_github_env("SKIP", "true")
return
if runner_index < 1:
raise click.ClickException(f"Runner index {runner_index} is out of range (must be >= 1)")
zero_based = runner_index - 1
assigned = items_for_runner(items, weights, zero_based, max_runners)
encoded = " ".join(assigned)
if github_env:
_write_github_env("ASSIGNED_ITEMS", encoded)
_write_github_env("SKIP", "false")
click.echo(f"Assigned {len(assigned)} items to runner {runner_index}: {encoded}")
return
click.echo(encoded)
if __name__ == "__main__": # pragma: no cover
main()
+3 -2
View File
@@ -36,6 +36,7 @@ import time
import click
from devx.config import REPO_NAME, REPO_OWNER
from devx.i18n import _
from devx.molecule.molecule_ci_guard import (
poll_for_other_failures,
@@ -53,10 +54,10 @@ def cli(pytest_args: tuple[str, ...]) -> None:
run_id = int(os.environ.get("RUN_ID", "0"))
job_name = os.environ.get("JOB_NAME", "integration-tests")
current_index = int(os.environ.get("MATRIX_INDEX", "0"))
repository = os.environ.get("GITEA_REPOSITORY", "oblachno-oss/devx")
repository = os.environ.get("GITEA_REPOSITORY", "")
owner, _sep, repo = repository.partition("/")
if not owner or not repo:
owner, repo = "oblachno-oss", "devx"
owner, repo = REPO_OWNER, REPO_NAME
if not all([gitea_url, token, run_id]):
click.echo(_("GITEA_URL/CI_GITEA_TOKEN/RUN_ID not set; running without cross-runner cancellation."))
+96 -2
View File
@@ -520,12 +520,102 @@ def post_review(client: GiteaClient, pr_number: str, result: ReviewResult) -> di
return client.create_review(pr_number, event=event, body=body, comments=comments)
def _post_manual_review(
client: GiteaClient,
pr_number: str,
event: str,
body: str | None,
checklist_confirmed: bool,
checklist_categories: str | None,
dry_run: bool,
) -> None:
"""Post a manual review with validation for APPROVE events."""
if not body or len(body) < 50:
raise click.ClickException(_("Review body must be at least 50 characters."))
if event == "APPROVE":
if not checklist_confirmed:
raise click.ClickException(
_("--checklist-confirmed is required for APPROVE events."),
)
cats = [c.strip() for c in (checklist_categories or "").split(",") if c.strip()]
cat_nums: list[int] = []
for c in cats:
try:
cat_nums.append(int(c))
except ValueError:
raise click.ClickException(
_("Invalid checklist category: {cat}. Must be numbers.", cat=c),
) from None
if len(cat_nums) < 8:
raise click.ClickException(
_("--checklist-categories must list at least 8 of 13 categories. Got {count}.", count=len(cat_nums)),
)
click.echo(f"Manual review event: {event}")
click.echo(f"Body: {body[:80]}...")
if checklist_confirmed:
click.echo(f"Checklist confirmed: {checklist_categories}")
if dry_run:
click.echo("\n[dry-run] Review not posted.")
return
try:
review = client.create_review(pr_number, event=event, body=body)
except APIError as e:
if "approve" in e.message.lower() or "422" in str(e.status):
click.echo(_("Note: Self-approval not allowed. Posting COMMENT instead."))
review = client.create_review(pr_number, event="COMMENT", body=body)
else:
raise
review_id = review.get("id", "?")
click.echo(
_(
"\nReview #{review_id} posted on PR #{pr_number} with event '{event}'.",
review_id=review_id,
pr_number=pr_number,
event=event,
)
)
@click.command()
@click.argument("pr_number")
@click.argument("repo")
@click.option("--dry-run", is_flag=True, default=False, help="Print review without posting.")
def main(pr_number: str, repo: str, dry_run: bool) -> None:
"""Run automated PR review and post results to Gitea."""
@click.option(
"--event",
type=click.Choice(["APPROVE", "REQUEST_CHANGES", "COMMENT"], case_sensitive=False),
default=None,
help="Post a manual review with the given event (skips automated checks).",
)
@click.option("--body", default=None, help="Review body text (required with --event).")
@click.option(
"--checklist-confirmed",
is_flag=True,
default=False,
help="Attest that REVIEW_CHECKLIST.md categories were checked (required for APPROVE).",
)
@click.option(
"--checklist-categories",
default=None,
help="Comma-separated checklist category numbers (required for APPROVE, min 8 of 13).",
)
def main(
pr_number: str,
repo: str,
dry_run: bool,
event: str | None,
body: str | None,
checklist_confirmed: bool,
checklist_categories: str | None,
) -> None:
"""Run automated PR review and post results to Gitea.
Without --event: runs automated checks and posts COMMENT/REQUEST_CHANGES.
With --event: posts a manual review (skips automated checks).
"""
token = os.environ.get("CI_GITEA_TOKEN", "")
if not token:
raise click.ClickException(_("ERROR: CI_GITEA_TOKEN is not set."))
@@ -533,6 +623,10 @@ def main(pr_number: str, repo: str, dry_run: bool) -> None:
owner, repo_name = repo.split("/")
client = GiteaClient(GITEA_API_URL, token, owner, repo_name)
if event is not None:
_post_manual_review(client, pr_number, event.upper(), body, checklist_confirmed, checklist_categories, dry_run)
return
result = run_review(client, pr_number)
body = build_review_body(result)
+2 -2
View File
@@ -28,7 +28,7 @@ from pathlib import Path
import click
from dotenv import load_dotenv # pyright: ignore[reportMissingImports,reportUnknownVariableType]
from devx.config import GITEA_API_URL
from devx.config import GITEA_API_URL, REPO_OWNER
from devx.gitea_cli import TeaCLI, TeaCLIError, configure_tea_login
from devx.i18n import _
@@ -165,7 +165,7 @@ def _default_gitea_registry_url() -> str:
base = base[: -len("/api/v1")]
elif base.endswith("/api"):
base = base[: -len("/api")]
owner = os.environ.get("DEVX_REPO_OWNER", "oblachno-oss")
owner = os.environ.get("DEVX_REPO_OWNER", "") or REPO_OWNER
return f"{base}/api/packages/{owner}/pypi"
+3 -3
View File
@@ -28,7 +28,7 @@ import click
from dotenv import load_dotenv # pyright: ignore[reportMissingImports,reportUnknownVariableType]
from devx.api_clients import GiteaClient
from devx.config import GITEA_API_URL
from devx.config import GITEA_API_URL, REPO_NAME, REPO_OWNER
from devx.exceptions import APIError
from devx.i18n import _
@@ -230,8 +230,8 @@ def main(dry_run: bool, repo: str | None, verify: bool, strict: bool) -> None:
raise click.ClickException(_("ERROR: CI_GITEA_TOKEN is not set."))
if repo is None:
owner = os.environ.get("DEVX_REPO_OWNER", "oblachno-oss")
repo_name = os.environ.get("DEVX_REPO_NAME", "devx")
owner = os.environ.get("DEVX_REPO_OWNER", "") or REPO_OWNER
repo_name = os.environ.get("DEVX_REPO_NAME", "") or REPO_NAME
else:
owner, repo_name = repo.split("/")
+11 -1
View File
@@ -63,7 +63,7 @@ DEVX_PIP_INSTALL := if [ -z "$$CI_GITEA_TOKEN" ]; then . ./.env 2>/dev/null; fi;
$(DEVX_BIN)/pip
.PHONY: devx-create-task devx-create-pr devx-push devx-push-with-pr devx-check-config
.PHONY: devx-pr-status devx-pr-logs devx-pr-label
.PHONY: devx-pr-status devx-pr-logs devx-pr-label devx-pr-review
.PHONY: devx-configure-gitea-pypi devx-install-tools devx-install-checkmake devx-checkmake
.PHONY: devx-workflow-lint devx-workflow-dryrun devx-workflow-dryrun-safe devx-workflow-check
.PHONY: devx-notify-failure devx-install-hooks devx-activate-scripts
@@ -124,6 +124,16 @@ devx-pr-label:
$(if $(PR),--pr $(PR)) \
--label $(or $(LABEL),ready-to-merge)
# Usage: make devx-pr-review PR=42 EVENT=APPROVE BODY="..." CHECKLIST=1,2,3,4,5,6,7,8,9,10,11,12,13
# make devx-pr-review PR=42 EVENT=REQUEST_CHANGES BODY="..."
# make devx-pr-review PR=42 (auto review)
devx-pr-review:
@$(DEVX_PYTHON) -m devx.ci.pr_review \
$(PR) $(DEVX_REPO_OWNER)/$(DEVX_REPO_NAME) \
$(if $(EVENT),--event $(EVENT)) \
$(if $(BODY),--body "$(BODY)") \
$(if $(CHECKLIST),--checklist-confirmed --checklist-categories $(CHECKLIST))
# ── Environment setup ─────────────────────────────────────────────────────────
# Configure Gitea private PyPI registry so pip can find devx and other
+3 -3
View File
@@ -29,7 +29,7 @@ import os
import click
import requests
from devx.config import GITEA_API_URL
from devx.config import GITEA_API_URL, REPO_NAME, REPO_OWNER
DEFAULT_MAX_RUNNERS = 3
@@ -145,9 +145,9 @@ def main(
token = os.environ.get("CI_GITEA_TOKEN", "")
if owner is None:
owner = os.environ.get("DEVX_REPO_OWNER", "oblachno-oss")
owner = os.environ.get("DEVX_REPO_OWNER", "") or REPO_OWNER
if repo is None:
repo = os.environ.get("DEVX_REPO_NAME", "devx")
repo = os.environ.get("DEVX_REPO_NAME", "") or REPO_NAME
count = get_runner_count(GITEA_API_URL, token, owner, repo)
indices = generate_indices(count)
+3 -2
View File
@@ -43,6 +43,7 @@ from pathlib import Path
import click
import requests
from devx.config import REPO_NAME, REPO_OWNER
from devx.i18n import _
POLL_INTERVAL = 10
@@ -167,10 +168,10 @@ def cli(pairs: tuple[str, ...], roles_root: Path | None) -> None:
run_id = int(os.environ.get("RUN_ID", "0"))
job_name = os.environ.get("JOB_NAME", "molecule-tests")
current_index = int(os.environ.get("MATRIX_INDEX", "0"))
repository = os.environ.get("GITEA_REPOSITORY", "oblachno-oss/devx")
repository = os.environ.get("GITEA_REPOSITORY", "")
owner, _sep, repo = repository.partition("/")
if not owner or not repo:
owner, repo = "oblachno-oss", "devx"
owner, repo = REPO_OWNER, REPO_NAME
if not all([gitea_url, token, run_id]):
click.echo(_("GITEA_URL/CI_GITEA_TOKEN/RUN_ID not set; running without cross-runner cancellation."))
+2 -4
View File
@@ -13,7 +13,6 @@ Usage::
from __future__ import annotations
import re
import sys
import tomllib
from pathlib import Path
@@ -27,8 +26,7 @@ def cli() -> None:
"""Validate devx configuration in pyproject.toml."""
path = Path("pyproject.toml")
if not path.exists():
click.echo(_("pyproject.toml not found in current directory."))
sys.exit(1)
raise click.ClickException(_("pyproject.toml not found in current directory."))
with open(path, "rb") as f: # noqa: PTH123
data = tomllib.load(f)
@@ -65,7 +63,7 @@ def cli() -> None:
if errors:
for err in errors:
click.echo(f"ERROR: {err}", err=True)
sys.exit(1)
raise click.ClickException(_("Configuration validation failed."))
click.echo(_("Configuration OK: [tool.devx] present, devx versions consistent."))
+21 -33
View File
@@ -23,12 +23,12 @@ Usage::
from __future__ import annotations
import argparse
import fnmatch
import subprocess # nosec B404
import sys
from pathlib import Path
import click
from devx.config import _load_pyproject_devx
from devx.i18n import _
@@ -203,49 +203,37 @@ def _find_missing_tests(
return missing
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
description=_("Check that changed files have corresponding tests"),
)
parser.add_argument(
"--staged-only",
action="store_true",
help=_("Only check staged files (for pre-commit)"),
)
parser.add_argument(
"--warn-only",
action="store_true",
help=_("Print warnings but always exit 0"),
)
args = parser.parse_args(argv)
@click.command()
@click.option("--staged-only", is_flag=True, help=_("Only check staged files (for pre-commit)"))
@click.option("--warn-only", is_flag=True, help=_("Print warnings but always exit 0"))
def cli(staged_only: bool, warn_only: bool) -> None:
"""Check that changed files have corresponding tests."""
repo_root = Path.cwd()
rules, skip_patterns, test_indicators, skip_extensions = _load_rules()
files = _changed_files(args.staged_only, repo_root)
files = _changed_files(staged_only, repo_root)
if not files:
print(_("[check_test_coverage] No changed files to check."))
return 0
click.echo(_("[check_test_coverage] No changed files to check."))
return
missing = _find_missing_tests(files, repo_root, rules, skip_patterns, test_indicators, skip_extensions)
if not missing:
print(f"[check_test_coverage] All {len(files)} changed file(s) have tests.")
return 0
click.echo(f"[check_test_coverage] All {len(files)} changed file(s) have tests.")
return
print("[check_test_coverage] FAILED: missing tests for changed files:\n", file=sys.stderr)
click.echo("[check_test_coverage] FAILED: missing tests for changed files:\n", err=True)
for f, reason in missing.items():
print(f" {f}", file=sys.stderr)
print(f" -> {reason}", file=sys.stderr)
click.echo(f" {f}", err=True)
click.echo(f" -> {reason}", err=True)
print(
"\n[check_test_coverage] Fix: add the missing test file(s) before committing.",
file=sys.stderr,
click.echo(
_("\n[check_test_coverage] Fix: add the missing test file(s) before committing."),
err=True,
)
if args.warn_only:
return 0
return 1
if not warn_only:
raise click.ClickException(_("Missing tests for changed files."))
if __name__ == "__main__": # pragma: no cover
sys.exit(main())
cli() # pragma: no cover
+8 -4
View File
@@ -40,7 +40,7 @@ from typing import Any
import click
import requests
from devx.config import GITEA_API_URL
from devx.config import GITEA_API_URL, REPO_OWNER
from devx.i18n import _
@@ -151,8 +151,8 @@ def select_for_deletion(
@click.command()
@click.option(
"--owner",
required=True,
help="Package owner (user or org).",
default=None,
help="Package owner (user or org, default: from [tool.devx] repo_owner).",
)
@click.option(
"--name",
@@ -180,7 +180,7 @@ def select_for_deletion(
help="Gitea API URL (defaults to DEVX_GITEA_API_URL or built-in default).",
)
def main(
owner: str,
owner: str | None,
names: tuple[str, ...],
keep: int,
dry_run: bool,
@@ -190,6 +190,10 @@ def main(
token = os.environ.get("CI_GITEA_TOKEN", "")
if not token:
raise click.ClickException(_("CI_GITEA_TOKEN environment variable required"))
if not owner:
owner = REPO_OWNER
if not owner:
raise click.ClickException(_("Package owner not specified. Use --owner or set [tool.devx] repo_owner."))
base_url = api_url or GITEA_API_URL
total_deleted = 0
+2 -2
View File
@@ -19,7 +19,7 @@ from typing import Any, cast
import click
from devx.api_clients import GiteaClient
from devx.config import GITEA_API_URL, REPO_OWNER
from devx.config import GITEA_API_URL, REPO_NAME, REPO_OWNER
from devx.exceptions import APIError
from devx.i18n import _
@@ -151,7 +151,7 @@ def main(repo: str | None, owner: str | None, branch: str, api_url: str | None)
token = os.environ.get("CI_GITEA_TOKEN", "")
if repo is None:
repo = os.environ.get("DEVX_REPO_NAME", "")
repo = os.environ.get("DEVX_REPO_NAME", "") or REPO_NAME
if not repo:
raise click.ClickException(_("ERROR: Repository name not specified. Use --repo or set DEVX_REPO_NAME."))
+10 -18
View File
@@ -76,24 +76,16 @@ def get_vikunja_task_title(task_id: str) -> str:
if not token:
raise click.ClickException(_("VIKUNJA_TOKEN is not set. Required to derive PR title."))
client = VikunjaClient(VIKUNJA_API_URL, token)
page = 1
while True:
tasks = client.list_project_tasks(VIKUNJA_PROJECT_ID, page=page, per_page=DEFAULT_PER_PAGE)
if not tasks:
break
matches = [t for t in tasks if t.get("identifier") == task_id]
if matches:
return str(matches[0].get("title", ""))
if len(tasks) < DEFAULT_PER_PAGE:
break
page += 1
raise click.ClickException(
_(
"Could not find Vikunja task {task_id} in project {project_id}.",
task_id=task_id,
project_id=VIKUNJA_PROJECT_ID,
),
)
task = client.find_task_by_identifier(VIKUNJA_PROJECT_ID, task_id, per_page=DEFAULT_PER_PAGE)
if not task:
raise click.ClickException(
_(
"Could not find Vikunja task {task_id} in project {project_id}.",
task_id=task_id,
project_id=VIKUNJA_PROJECT_ID,
),
)
return str(task.get("title", ""))
def find_existing_pr(client: GiteaClient, branch: str) -> dict | None:
+1 -1
View File
@@ -109,7 +109,7 @@ def _generate(prefix: str) -> str:
@click.option(
"--prefix",
default=TASK_PREFIX,
help="Task ID prefix for commit preprocessor (default: DEVX_TASK_PREFIX env var or 'DEVX').",
help="Task ID prefix for commit preprocessor (default: from [tool.devx] task_prefix in pyproject.toml).",
)
@click.option(
"--output",
+3 -2
View File
@@ -56,7 +56,8 @@ def _download_binary() -> None:
TARGET_PATH.chmod(0o755)
def main() -> None:
@click.command()
def cli() -> None:
"""Install checkmake if not already present."""
if shutil.which("checkmake") is not None:
return
@@ -66,4 +67,4 @@ def main() -> None:
if __name__ == "__main__": # pragma: no cover
main() # pragma: no cover
cli() # pragma: no cover
+1 -11
View File
@@ -59,17 +59,7 @@ def task_exists(task_id: str) -> bool:
if not token:
return False
client = VikunjaClient(VIKUNJA_API_URL, token)
page = 1
while True:
tasks = client.list_project_tasks(VIKUNJA_PROJECT_ID, page=page, per_page=DEFAULT_PER_PAGE)
if not tasks:
break
if any(t.get("identifier") == task_id for t in tasks):
return True
if len(tasks) < DEFAULT_PER_PAGE:
break
page += 1
return False
return client.find_task_by_identifier(VIKUNJA_PROJECT_ID, task_id, per_page=DEFAULT_PER_PAGE) is not None
def validate(branch: str) -> None:
+88 -8
View File
@@ -463,14 +463,6 @@
"ru": "Bumping version: {current} -> v{new_version}",
"zh": "Bumping version: {current} -> v{new_version}"
},
"Check that changed files have corresponding tests": {
"bg": "Check that changed files have corresponding tests",
"de": "Check that changed files have corresponding tests",
"en": "Check that changed files have corresponding tests",
"pl": "Check that changed files have corresponding tests",
"ru": "Check that changed files have corresponding tests",
"zh": "Check that changed files have corresponding tests"
},
"Checking CLI command documentation...": {
"bg": "Checking CLI command documentation...",
"de": "Checking CLI command documentation...",
@@ -647,6 +639,14 @@
"ru": "Dockerfile not found: {path}",
"zh": "Dockerfile not found: {path}"
},
"Each item must be a string or an object with 'id', got {type}": {
"bg": "Всеки елемент трябва да е низ или обект с 'id', получено {type}",
"de": "Jedes Element muss ein String oder ein Objekt mit 'id' sein, erhalten {type}",
"en": "Each item must be a string or an object with 'id', got {type}",
"pl": "Każdy element musi być ciągiem lub obiektem z 'id', otrzymano {type}",
"ru": "Каждый элемент должен быть строкой или объектом с 'id', получено {type}",
"zh": "每个元素必须是字符串或带有 'id' 的对象,得到 {type}"
},
"Dry-run mode: on branch '{branch}' (not master). Some checks may behave differently.": {
"bg": "Dry-run mode: on branch '{branch}' (not master). Some checks may behave differently.",
"de": "Dry-run mode: on branch '{branch}' (not master). Some checks may behave differently.",
@@ -2190,5 +2190,85 @@
"pl": "Waiting for CI checks to complete (timeout: {timeout}s)...",
"ru": "Waiting for CI checks to complete (timeout: {timeout}s)...",
"zh": "Waiting for CI checks to complete (timeout: {timeout}s)..."
},
"\n[check_test_coverage] Fix: add the missing test file(s) before committing.": {
"en": "\n[check_test_coverage] Fix: add the missing test file(s) before committing.",
"bg": "\n[check_test_coverage] Fix: add the missing test file(s) before committing.",
"de": "\n[check_test_coverage] Fix: add the missing test file(s) before committing.",
"pl": "\n[check_test_coverage] Fix: add the missing test file(s) before committing.",
"ru": "\n[check_test_coverage] Fix: add the missing test file(s) before committing.",
"zh": "\n[check_test_coverage] Fix: add the missing test file(s) before committing."
},
"Package owner not specified. Use --owner or set [tool.devx] repo_owner.": {
"en": "Package owner not specified. Use --owner or set [tool.devx] repo_owner.",
"bg": "Package owner not specified. Use --owner or set [tool.devx] repo_owner.",
"de": "Package owner not specified. Use --owner or set [tool.devx] repo_owner.",
"pl": "Package owner not specified. Use --owner or set [tool.devx] repo_owner.",
"ru": "Package owner not specified. Use --owner or set [tool.devx] repo_owner.",
"zh": "Package owner not specified. Use --owner or set [tool.devx] repo_owner."
},
"Configuration validation failed.": {
"en": "Configuration validation failed.",
"bg": "Configuration validation failed.",
"de": "Configuration validation failed.",
"pl": "Configuration validation failed.",
"ru": "Configuration validation failed.",
"zh": "Configuration validation failed."
},
"Missing tests for changed files.": {
"en": "Missing tests for changed files.",
"bg": "Missing tests for changed files.",
"de": "Missing tests for changed files.",
"pl": "Missing tests for changed files.",
"ru": "Missing tests for changed files.",
"zh": "Missing tests for changed files."
},
"\nReview #{review_id} posted on PR #{pr_number} with event '{event}'.": {
"en": "\nReview #{review_id} posted on PR #{pr_number} with event '{event}'.",
"bg": "\nReview #{review_id} posted on PR #{pr_number} with event '{event}'.",
"de": "\nReview #{review_id} posted on PR #{pr_number} with event '{event}'.",
"pl": "\nReview #{review_id} posted on PR #{pr_number} with event '{event}'.",
"ru": "\nReview #{review_id} posted on PR #{pr_number} with event '{event}'.",
"zh": "\nReview #{review_id} posted on PR #{pr_number} with event '{event}'."
},
"--checklist-categories must list at least 8 of 13 categories. Got {count}.": {
"en": "--checklist-categories must list at least 8 of 13 categories. Got {count}.",
"bg": "--checklist-categories must list at least 8 of 13 categories. Got {count}.",
"de": "--checklist-categories must list at least 8 of 13 categories. Got {count}.",
"pl": "--checklist-categories must list at least 8 of 13 categories. Got {count}.",
"ru": "--checklist-categories must list at least 8 of 13 categories. Got {count}.",
"zh": "--checklist-categories must list at least 8 of 13 categories. Got {count}."
},
"--checklist-confirmed is required for APPROVE events.": {
"en": "--checklist-confirmed is required for APPROVE events.",
"bg": "--checklist-confirmed is required for APPROVE events.",
"de": "--checklist-confirmed is required for APPROVE events.",
"pl": "--checklist-confirmed is required for APPROVE events.",
"ru": "--checklist-confirmed is required for APPROVE events.",
"zh": "--checklist-confirmed is required for APPROVE events."
},
"Invalid checklist category: {cat}. Must be numbers.": {
"en": "Invalid checklist category: {cat}. Must be numbers.",
"bg": "Invalid checklist category: {cat}. Must be numbers.",
"de": "Invalid checklist category: {cat}. Must be numbers.",
"pl": "Invalid checklist category: {cat}. Must be numbers.",
"ru": "Invalid checklist category: {cat}. Must be numbers.",
"zh": "Invalid checklist category: {cat}. Must be numbers."
},
"Items input must be a JSON array, got {type}": {
"bg": "Входните данни трябва да са JSON масив, получено {type}",
"de": "Eingabe muss ein JSON-Array sein, erhalten {type}",
"en": "Items input must be a JSON array, got {type}",
"pl": "Dane wejściowe muszą być tablicą JSON, otrzymano {type}",
"ru": "Входные данные должны быть JSON-массивом, получено {type}",
"zh": "输入必须是 JSON 数组,得到 {type}"
},
"Review body must be at least 50 characters.": {
"en": "Review body must be at least 50 characters.",
"bg": "Review body must be at least 50 characters.",
"de": "Review body must be at least 50 characters.",
"pl": "Review body must be at least 50 characters.",
"ru": "Review body must be at least 50 characters.",
"zh": "Review body must be at least 50 characters."
}
}
+36
View File
@@ -781,6 +781,42 @@ class TestVikunjaClient:
call_kwargs = client._session.request.call_args.kwargs
assert call_kwargs["json"]["description"] == ""
def test_find_task_by_identifier_found(self) -> None:
client = VikunjaClient("https://work.example.com", "tok")
client._session.request = MagicMock(
return_value=_mock_response([{"identifier": "DEVX-1"}, {"identifier": "DEVX-42", "title": "Found"}])
)
result = client.find_task_by_identifier(6, "DEVX-42", per_page=50)
assert result is not None
assert result["title"] == "Found"
def test_find_task_by_identifier_not_found(self) -> None:
client = VikunjaClient("https://work.example.com", "tok")
client._session.request = MagicMock(
return_value=_mock_response([{"identifier": "DEVX-1"}, {"identifier": "DEVX-2"}])
)
result = client.find_task_by_identifier(6, "DEVX-99", per_page=50)
assert result is None
def test_find_task_by_identifier_empty_project(self) -> None:
client = VikunjaClient("https://work.example.com", "tok")
client._session.request = MagicMock(return_value=_mock_response([]))
result = client.find_task_by_identifier(6, "DEVX-1", per_page=50)
assert result is None
def test_find_task_by_identifier_paginates(self) -> None:
client = VikunjaClient("https://work.example.com", "tok")
full_page = [{"identifier": f"DEVX-{i}"} for i in range(50)]
client._session.request = MagicMock(
side_effect=[
_mock_response(full_page),
_mock_response([{"identifier": "DEVX-50", "title": "Found on page 2"}]),
]
)
result = client.find_task_by_identifier(6, "DEVX-50", per_page=50)
assert result is not None
assert result["title"] == "Found on page 2"
class TestIsRetryable:
def test_connection_error_is_retryable(self) -> None:
+23
View File
@@ -195,6 +195,29 @@ class TestExtractConventionalMsg:
]
assert extract_conventional_msg(commits) == "feat: add feature"
def test_prefers_feat_over_refactor(self) -> None:
"""When both feat and refactor commits exist, feat wins."""
commits = [
{"commit": {"message": "refactor: add find_task_by_identifier"}},
{"commit": {"message": "fix: remove hardcoded fallbacks"}},
{"commit": {"message": "feat: add manual review support"}},
]
assert extract_conventional_msg(commits) == "feat: add manual review support"
def test_prefers_fix_over_docs(self) -> None:
commits = [
{"commit": {"message": "docs: update README"}},
{"commit": {"message": "fix: resolve bug"}},
]
assert extract_conventional_msg(commits) == "fix: resolve bug"
def test_scope_in_prefix(self) -> None:
commits = [
{"commit": {"message": "refactor(ci): cleanup code"}},
{"commit": {"message": "feat(api): add endpoint"}},
]
assert extract_conventional_msg(commits) == "feat(api): add endpoint"
# -- run_cmd --
+30 -1
View File
@@ -705,4 +705,33 @@ class TestCLICleanImages:
)
assert result.exit_code != 0
assert "FAILED" in result.output
assert "failed" in result.output.lower()
@patch("devx.tools.clean_images.REPO_OWNER", "oblachno-oss")
def test_owner_from_config(self) -> None:
from devx.tools.clean_images import main as clean_main
runner = CliRunner()
mock_resp = MagicMock()
mock_resp.json.return_value = []
mock_resp.raise_for_status = MagicMock()
with patch.dict("os.environ", {"CI_GITEA_TOKEN": "fake"}):
with patch("devx.tools.clean_images.requests.get", return_value=mock_resp):
result = runner.invoke(
clean_main,
["--name", "ci-base", "--dry-run"],
)
assert result.exit_code == 0
assert "oblachno-oss/ci-base" in result.output
@patch("devx.tools.clean_images.REPO_OWNER", "")
def test_no_owner_raises(self) -> None:
from devx.tools.clean_images import main as clean_main
runner = CliRunner()
with patch.dict("os.environ", {"CI_GITEA_TOKEN": "fake"}):
result = runner.invoke(
clean_main,
["--name", "ci-base"],
)
assert result.exit_code != 0
assert "owner" in result.output.lower()
+17 -5
View File
@@ -3,6 +3,8 @@
from pathlib import Path
from unittest.mock import MagicMock, patch
from click.testing import CliRunner
from devx.tools.check_test_coverage import (
BUILTIN_RULES,
DEFAULT_SKIP_EXTENSIONS,
@@ -13,7 +15,7 @@ from devx.tools.check_test_coverage import (
_load_rules,
_resolve_test_path,
_should_skip_file,
main,
cli,
)
@@ -201,7 +203,10 @@ class TestMain:
),
patch("devx.tools.check_test_coverage.Path.cwd", return_value=tmp_path),
):
assert main([]) == 0
runner = CliRunner()
result = runner.invoke(cli, [])
assert result.exit_code == 0
assert "No changed files" in result.output
def test_all_have_tests(self, tmp_path: Path) -> None:
(tmp_path / "scripts" / "tests").mkdir(parents=True)
@@ -214,7 +219,10 @@ class TestMain:
),
patch("devx.tools.check_test_coverage.Path.cwd", return_value=tmp_path),
):
assert main([]) == 0
runner = CliRunner()
result = runner.invoke(cli, [])
assert result.exit_code == 0
assert "have tests" in result.output
def test_missing_test_returns_1(self, tmp_path: Path) -> None:
with (
@@ -225,7 +233,9 @@ class TestMain:
),
patch("devx.tools.check_test_coverage.Path.cwd", return_value=tmp_path),
):
assert main([]) == 1
runner = CliRunner()
result = runner.invoke(cli, [])
assert result.exit_code != 0
def test_warn_only_returns_0(self, tmp_path: Path) -> None:
with (
@@ -236,4 +246,6 @@ class TestMain:
),
patch("devx.tools.check_test_coverage.Path.cwd", return_value=tmp_path),
):
assert main(["--warn-only"]) == 0
runner = CliRunner()
result = runner.invoke(cli, ["--warn-only"])
assert result.exit_code == 0
+17
View File
@@ -145,6 +145,7 @@ class TestMain:
assert "CI_GITEA_TOKEN" in result.output
@patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True)
@patch("devx.tools.configure_repo.REPO_NAME", "")
def test_main_no_repo(self) -> None:
runner = CliRunner()
result = runner.invoke(main, [])
@@ -197,3 +198,19 @@ class TestMain:
call_args = mock_client_cls.call_args
assert call_args[0][2] == "oblachno" # owner
assert call_args[0][3] == "infra" # repo
@patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True)
@patch("devx.tools.configure_repo.REPO_NAME", "devx")
@patch("devx.tools.configure_repo.REPO_OWNER", "oblachno-oss")
@patch("devx.tools.configure_repo.GiteaClient")
def test_main_repo_from_pyproject(self, mock_client_cls: MagicMock) -> None:
"""When no env var is set, repo name should come from pyproject.toml."""
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
runner = CliRunner()
result = runner.invoke(main, [])
assert result.exit_code == 0
call_args = mock_client_cls.call_args
assert call_args[0][2] == "oblachno-oss" # owner
assert call_args[0][3] == "devx" # repo
+2 -15
View File
@@ -56,7 +56,7 @@ class TestGetVikunjaTaskTitle:
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"})
def test_found(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.list_project_tasks.return_value = [{"identifier": "DEVX-42", "title": "Add feature"}]
mock_client.find_task_by_identifier.return_value = {"identifier": "DEVX-42", "title": "Add feature"}
mock_client_cls.return_value = mock_client
assert get_vikunja_task_title("DEVX-42") == "Add feature"
@@ -69,20 +69,7 @@ class TestGetVikunjaTaskTitle:
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"})
def test_not_found(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.list_project_tasks.return_value = []
mock_client_cls.return_value = mock_client
with pytest.raises(click.ClickException, match="Could not find"):
get_vikunja_task_title("DEVX-42")
@patch("devx.tools.create_pr.VikunjaClient")
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"})
def test_pagination_not_found(self, mock_client_cls: MagicMock) -> None:
from devx.config import DEFAULT_PER_PAGE
mock_client = MagicMock()
page1 = [{"identifier": f"OTHER-{i}"} for i in range(DEFAULT_PER_PAGE)]
page2 = [{"identifier": "OTHER-99"}]
mock_client.list_project_tasks.side_effect = [page1, page2]
mock_client.find_task_by_identifier.return_value = None
mock_client_cls.return_value = mock_client
with pytest.raises(click.ClickException, match="Could not find"):
get_vikunja_task_title("DEVX-42")
+250
View File
@@ -0,0 +1,250 @@
"""Unit tests for devx.ci.distribute_items."""
import pytest
from click.testing import CliRunner
from devx.ci.distribute_items import (
DEFAULT_WEIGHT,
distribute,
items_for_runner,
main,
parse_items,
parse_weighted_items,
)
class TestParseItems:
def test_string_array(self) -> None:
assert parse_items('["a", "b", "c"]') == ["a", "b", "c"]
def test_object_array(self) -> None:
raw = '[{"id": "a", "weight": 2}, {"id": "b"}]'
assert parse_items(raw) == ["a", "b"]
def test_empty_array(self) -> None:
assert parse_items("[]") == []
def test_not_an_array(self) -> None:
with pytest.raises(Exception, match="must be a JSON array"):
parse_items('{"key": "value"}')
def test_invalid_entry_type(self) -> None:
with pytest.raises(Exception, match="must be a string or an object"):
parse_items("[42]")
def test_object_without_id(self) -> None:
with pytest.raises(Exception, match="must be a string or an object"):
parse_items('[{"weight": 2}]')
class TestParseWeightedItems:
def test_string_array_default_weights(self) -> None:
items, weights = parse_weighted_items('["a", "b"]')
assert items == ["a", "b"]
assert weights == [DEFAULT_WEIGHT, DEFAULT_WEIGHT]
def test_object_array_with_weights(self) -> None:
items, weights = parse_weighted_items('[{"id": "a", "weight": 5}, {"id": "b", "weight": 1}]')
assert items == ["a", "b"]
assert weights == [5, 1]
def test_object_array_missing_weight(self) -> None:
items, weights = parse_weighted_items('[{"id": "a"}]')
assert items == ["a"]
assert weights == [DEFAULT_WEIGHT]
def test_not_an_array(self) -> None:
with pytest.raises(Exception, match="must be a JSON array"):
parse_weighted_items('"hello"')
def test_invalid_entry(self) -> None:
with pytest.raises(Exception, match="must be a string or an object"):
parse_weighted_items("[true]")
class TestDistribute:
def test_even_split(self) -> None:
items = [f"vm-{i}" for i in range(6)]
weights = [1] * 6
groups = distribute(items, weights, 3)
assert len(groups) == 3
assert all(len(g) == 2 for g in groups)
def test_uneven_split(self) -> None:
items = [f"vm-{i}" for i in range(5)]
weights = [1] * 5
groups = distribute(items, weights, 3)
assert len(groups[0]) == 2
assert len(groups[1]) == 2
assert len(groups[2]) == 1
def test_more_runners_than_items(self) -> None:
items = ["vm-a"]
weights = [1]
groups = distribute(items, weights, 5)
assert len(groups) == 5
assert len(groups[0]) == 1
assert all(len(g) == 0 for g in groups[1:])
def test_lpt_heavy_item_on_least_loaded(self) -> None:
items = ["heavy", "light1", "light2", "light3"]
weights = [10, 1, 1, 1]
groups = distribute(items, weights, 2)
# Heavy item goes to runner 0, lights go to runner 1 (least loaded)
assert "heavy" in groups[0]
# Runner 1 should have more items but less total weight
assert len(groups[1]) >= 2
def test_empty_items(self) -> None:
groups = distribute([], [], 3)
assert len(groups) == 3
assert all(len(g) == 0 for g in groups)
def test_single_runner(self) -> None:
items = ["a", "b", "c"]
weights = [1, 2, 3]
groups = distribute(items, weights, 1)
assert len(groups) == 1
assert len(groups[0]) == 3
class TestItemsForRunner:
def test_returns_assigned_subset(self) -> None:
items = ["a", "b", "c", "d", "e", "f"]
weights = [1] * 6
result = items_for_runner(items, weights, 0, 3)
assert len(result) == 2
assert all(item in items for item in result)
def test_out_of_range(self) -> None:
with pytest.raises(Exception, match="out of range"):
items_for_runner(["a"], [1], 5, 3)
def test_negative_index(self) -> None:
with pytest.raises(Exception, match="out of range"):
items_for_runner(["a"], [1], -1, 3)
class TestMain:
def test_stdin_string_array(self) -> None:
runner = CliRunner()
result = runner.invoke(main, ["--runner-index", "1", "--max-runners", "2"], input='["a", "b", "c"]')
assert result.exit_code == 0
# LPT: heaviest first, so "a" goes to runner 0, "b" to runner 1, "c" to runner 0
# All weights equal, so round-robin-ish: runner 0 gets "a","c"; runner 1 gets "b"
assert "a" in result.output
def test_stdin_object_array(self) -> None:
runner = CliRunner()
result = runner.invoke(
main,
["--runner-index", "1", "--max-runners", "2"],
input='[{"id": "a", "weight": 5}, {"id": "b", "weight": 1}]',
)
assert result.exit_code == 0
assert "a" in result.output
def test_items_file(self, tmp_path: object) -> None:
import pathlib
items_file = pathlib.Path(str(tmp_path)) / "items.json"
items_file.write_text('["x", "y", "z"]')
runner = CliRunner()
result = runner.invoke(main, ["--items-file", str(items_file), "--runner-index", "1", "--max-runners", "3"])
assert result.exit_code == 0
assert "x" in result.output
def test_print_all_groups_no_runner_index(self) -> None:
runner = CliRunner()
result = runner.invoke(main, ["--max-runners", "2"], input='["a", "b"]')
assert result.exit_code == 0
assert "Runner 0:" in result.output
assert "Runner 1:" in result.output
def test_empty_stdin(self) -> None:
runner = CliRunner()
result = runner.invoke(main, ["--runner-index", "1", "--max-runners", "3"], input="")
assert result.exit_code == 0
# Empty input → empty assigned items
assert result.output.strip() == ""
def test_github_env(self, tmp_path: object, monkeypatch: pytest.MonkeyPatch) -> None:
import pathlib
gh_env = pathlib.Path(str(tmp_path)) / "gh_env"
gh_env.write_text("")
monkeypatch.setenv("GITHUB_ENV", str(gh_env))
runner = CliRunner()
result = runner.invoke(
main,
["--runner-index", "1", "--max-runners", "2", "--github-env"],
input='["a", "b"]',
)
assert result.exit_code == 0
content = gh_env.read_text()
assert "ASSIGNED_ITEMS=" in content
assert "SKIP=false" in content
def test_skip_if_excess(self, tmp_path: object, monkeypatch: pytest.MonkeyPatch) -> None:
import pathlib
gh_env = pathlib.Path(str(tmp_path)) / "gh_env"
gh_env.write_text("")
monkeypatch.setenv("GITHUB_ENV", str(gh_env))
runner = CliRunner()
result = runner.invoke(
main,
["--runner-index", "5", "--max-runners", "3", "--github-env", "--skip-if-excess"],
input='["a"]',
)
assert result.exit_code == 0
content = gh_env.read_text()
assert "ASSIGNED_ITEMS=" in content
assert "SKIP=true" in content
def test_runner_index_zero(self) -> None:
runner = CliRunner()
result = runner.invoke(main, ["--runner-index", "0"], input='["a"]')
assert result.exit_code != 0
assert "out of range" in result.output
def test_default_max_runners(self) -> None:
runner = CliRunner()
result = runner.invoke(main, ["--runner-index", "1"], input='["a"]')
assert result.exit_code == 0
assert "a" in result.output
def test_github_env_not_set(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("GITHUB_ENV", raising=False)
runner = CliRunner()
result = runner.invoke(
main,
["--runner-index", "1", "--github-env"],
input='["a"]',
)
assert result.exit_code != 0
assert "GITHUB_ENV" in result.output
def test_invalid_json(self) -> None:
runner = CliRunner()
result = runner.invoke(main, ["--runner-index", "1"], input="not json")
assert result.exit_code != 0
def test_multiline_github_env(self, tmp_path: object, monkeypatch: pytest.MonkeyPatch) -> None:
import pathlib
gh_env = pathlib.Path(str(tmp_path)) / "gh_env"
gh_env.write_text("")
monkeypatch.setenv("GITHUB_ENV", str(gh_env))
runner = CliRunner()
# Items with newlines in their IDs would trigger multiline syntax
# Normal items don't have newlines, but test the path anyway
result = runner.invoke(
main,
["--runner-index", "1", "--max-runners", "1", "--github-env"],
input='["a\\nb"]',
)
assert result.exit_code == 0
content = gh_env.read_text()
# Item "a\nb" contains a newline → heredoc syntax
assert "ASSIGNED_ITEMS<<" in content
+12 -3
View File
@@ -63,13 +63,19 @@ class TestDownloadBinary:
class TestMain:
def test_already_installed(self) -> None:
from click.testing import CliRunner
with patch("shutil.which", return_value="/usr/bin/checkmake"):
install_checkmake.main()
runner = CliRunner()
runner.invoke(install_checkmake.cli, [])
def test_install_with_go(self) -> None:
from click.testing import CliRunner
with patch("shutil.which", side_effect=[None, "/usr/bin/go"]):
with patch("subprocess.run") as mock_run:
install_checkmake.main()
runner = CliRunner()
runner.invoke(install_checkmake.cli, [])
mock_run.assert_called_once_with(
[
"/usr/bin/go",
@@ -80,6 +86,8 @@ class TestMain:
)
def test_download_when_no_go(self, tmp_path: Path) -> None:
from click.testing import CliRunner
target = tmp_path / "checkmake"
def _write_file(url: str, path: str) -> tuple[str, None]:
@@ -90,5 +98,6 @@ class TestMain:
with patch("shutil.which", side_effect=[None, None]):
with patch.object(platform, "machine", return_value="x86_64"):
with patch("urllib.request.urlretrieve", side_effect=_write_file) as mock_retrieve:
install_checkmake.main()
runner = CliRunner()
runner.invoke(install_checkmake.cli, [])
mock_retrieve.assert_called_once()
+176
View File
@@ -733,6 +733,182 @@ class TestMain:
assert "CI_GITEA_TOKEN" in result.output
class TestManualReview:
@patch("devx.ci.pr_review.GiteaClient")
def test_approve_success(self, mock_client_class: MagicMock) -> None:
mock_client_class.return_value.create_review.return_value = {"id": 200}
runner = CliRunner()
result = runner.invoke(
main,
[
"42",
"oblachno-oss/devx",
"--event",
"APPROVE",
"--body",
"All 13 REVIEW_CHECKLIST.md categories verified. Architecture: clean. Security: no issues.",
"--checklist-confirmed",
"--checklist-categories",
"1,2,3,4,5,6,7,8,9,10,11,12,13",
],
env={"CI_GITEA_TOKEN": "fake"},
)
assert result.exit_code == 0
assert "Review #200" in result.output
@patch("devx.ci.pr_review.GiteaClient")
def test_approve_without_checklist_confirmed_fails(self, mock_client_class: MagicMock) -> None:
runner = CliRunner()
result = runner.invoke(
main,
[
"42",
"oblachno-oss/devx",
"--event",
"APPROVE",
"--body",
"x" * 60,
"--checklist-categories",
"1,2,3,4,5,6,7,8",
],
env={"CI_GITEA_TOKEN": "fake"},
)
assert result.exit_code != 0
assert "checklist-confirmed" in result.output
@patch("devx.ci.pr_review.GiteaClient")
def test_approve_with_too_few_categories_fails(self, mock_client_class: MagicMock) -> None:
runner = CliRunner()
result = runner.invoke(
main,
[
"42",
"oblachno-oss/devx",
"--event",
"APPROVE",
"--body",
"x" * 60,
"--checklist-confirmed",
"--checklist-categories",
"1,2,3",
],
env={"CI_GITEA_TOKEN": "fake"},
)
assert result.exit_code != 0
assert "at least 8" in result.output
@patch("devx.ci.pr_review.GiteaClient")
def test_approve_with_short_body_fails(self, mock_client_class: MagicMock) -> None:
runner = CliRunner()
result = runner.invoke(
main,
[
"42",
"oblachno-oss/devx",
"--event",
"APPROVE",
"--body",
"LGTM",
"--checklist-confirmed",
"--checklist-categories",
"1,2,3,4,5,6,7,8",
],
env={"CI_GITEA_TOKEN": "fake"},
)
assert result.exit_code != 0
assert "50 characters" in result.output
@patch("devx.ci.pr_review.GiteaClient")
def test_approve_with_invalid_category_fails(self, mock_client_class: MagicMock) -> None:
runner = CliRunner()
result = runner.invoke(
main,
[
"42",
"oblachno-oss/devx",
"--event",
"APPROVE",
"--body",
"x" * 60,
"--checklist-confirmed",
"--checklist-categories",
"1,2,abc,4",
],
env={"CI_GITEA_TOKEN": "fake"},
)
assert result.exit_code != 0
assert "Invalid" in result.output
@patch("devx.ci.pr_review.GiteaClient")
def test_request_changes_success(self, mock_client_class: MagicMock) -> None:
mock_client_class.return_value.create_review.return_value = {"id": 201}
runner = CliRunner()
result = runner.invoke(
main,
[
"42",
"oblachno-oss/devx",
"--event",
"REQUEST_CHANGES",
"--body",
"Please fix the architecture issues in the CLI module before merging.",
],
env={"CI_GITEA_TOKEN": "fake"},
)
assert result.exit_code == 0
assert "Review #201" in result.output
@patch("devx.ci.pr_review.GiteaClient")
def test_manual_review_dry_run(self, mock_client_class: MagicMock) -> None:
runner = CliRunner()
result = runner.invoke(
main,
["42", "oblachno-oss/devx", "--event", "COMMENT", "--body", "x" * 60, "--dry-run"],
env={"CI_GITEA_TOKEN": "fake"},
)
assert result.exit_code == 0
assert "[dry-run]" in result.output
mock_client_class.return_value.create_review.assert_not_called()
@patch("devx.ci.pr_review.GiteaClient")
def test_manual_review_self_approval_fallback(self, mock_client_class: MagicMock) -> None:
client = mock_client_class.return_value
client.create_review.side_effect = [
APIError(422, "approve your own pull is not allowed"),
{"id": 202},
]
runner = CliRunner()
result = runner.invoke(
main,
[
"42",
"oblachno-oss/devx",
"--event",
"APPROVE",
"--body",
"x" * 60,
"--checklist-confirmed",
"--checklist-categories",
"1,2,3,4,5,6,7,8",
],
env={"CI_GITEA_TOKEN": "fake"},
)
assert result.exit_code == 0
assert "Review #202" in result.output
@patch("devx.ci.pr_review.GiteaClient")
def test_manual_review_other_error_re_raises(self, mock_client_class: MagicMock) -> None:
client = mock_client_class.return_value
client.create_review.side_effect = APIError(500, "Internal server error")
runner = CliRunner()
result = runner.invoke(
main,
["42", "oblachno-oss/devx", "--event", "COMMENT", "--body", "x" * 60],
env={"CI_GITEA_TOKEN": "fake"},
)
assert result.exit_code != 0
def test_main_module_block() -> None:
import devx.ci.pr_review as pr
+2 -33
View File
@@ -43,7 +43,7 @@ class TestTaskExists:
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"})
def test_found(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.list_project_tasks.return_value = [{"identifier": "DEVX-42"}]
mock_client.find_task_by_identifier.return_value = {"identifier": "DEVX-42"}
mock_client_cls.return_value = mock_client
assert task_exists("DEVX-42") is True
@@ -51,7 +51,7 @@ class TestTaskExists:
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"})
def test_not_found(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.list_project_tasks.return_value = [{"identifier": "DEVX-99"}]
mock_client.find_task_by_identifier.return_value = None
mock_client_cls.return_value = mock_client
assert task_exists("DEVX-42") is False
@@ -59,37 +59,6 @@ class TestTaskExists:
def test_no_token(self) -> None:
assert task_exists("DEVX-42") is False
@patch("devx.tools.pre_push_check.VikunjaClient")
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"})
def test_pagination(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
# First page: full page (50 items, none matching), second page: match
page1 = [{"identifier": f"OTHER-{i}"} for i in range(50)]
page2 = [{"identifier": "DEVX-42"}]
mock_client.list_project_tasks.side_effect = [page1, page2]
mock_client_cls.return_value = mock_client
assert task_exists("DEVX-42") is True
@patch("devx.tools.pre_push_check.VikunjaClient")
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"})
def test_empty_project(self, mock_client_cls: MagicMock) -> None:
mock_client = MagicMock()
mock_client.list_project_tasks.return_value = []
mock_client_cls.return_value = mock_client
assert task_exists("DEVX-42") is False
@patch("devx.tools.pre_push_check.VikunjaClient")
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"})
def test_pagination_not_found(self, mock_client_cls: MagicMock) -> None:
from devx.config import DEFAULT_PER_PAGE
mock_client = MagicMock()
page1 = [{"identifier": f"OTHER-{i}"} for i in range(DEFAULT_PER_PAGE)]
page2 = [{"identifier": "OTHER-99"}]
mock_client.list_project_tasks.side_effect = [page1, page2]
mock_client_cls.return_value = mock_client
assert task_exists("DEVX-42") is False
class TestValidate:
def test_master_branch_skips(self) -> None: