DEVX-91: refactor: add find_task_by_identifier, config fallbacks for tools
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
Post-merge / release (push) Successful in 27s
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

This commit was merged in pull request #145.
This commit is contained in:
2026-06-28 00:06:29 +00:00
parent 93a1cb9945
commit 4c1ecbf4fa
23 changed files with 232 additions and 155 deletions
+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.
+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)
+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."))
+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("/")
+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:
+32 -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...",
@@ -2190,5 +2182,37 @@
"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."
}
}