diff --git a/AGENTS.md b/AGENTS.md index 7ef3006..a42c799 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -83,7 +83,12 @@ src/devx/ │ ├── check_test_coverage.py # Ensure changed files have corresponding tests (configurable rules) │ ├── check_agent_docs.py # Validate docs for stale file references (configurable patterns) │ ├── configure_repo.py # Branch protection and label setup -│ └── generate_badges.py # Badge SVG generation +│ ├── generate_badges.py # Badge SVG generation +│ ├── create_task.py # Create Vikunja tasks +│ ├── create_pr.py # Create PRs with auto-derived title from Vikunja +│ ├── pr_status.py # Check CI status for a PR/commit (--wait polls) +│ ├── pr_logs.py # Fetch logs for failed CI jobs +│ └── pr_label.py # Add labels to PRs (idempotent) ├── opentofu.py # OpenTofu output helpers (get_tofu_output, get_tofu_vm_ip, get_tofu_vm_field) └── molecule/ # Optional molecule testing helpers (for Ansible projects) ├── discover_runners.py # Dynamic Gitea runner discovery @@ -398,6 +403,9 @@ projects. | `devx-create-pr` | Create a PR with auto-derived title | | `devx-push` | Push current branch to origin | | `devx-push-with-pr` | Push and create PR in one step | +| `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-check-config` | Validate devx configuration | | `devx-configure-gitea-pypi` | Configure Gitea private PyPI registry | | `devx-env` | Create .env from .env.example | diff --git a/src/devx/api_clients.py b/src/devx/api_clients.py index 64388db..afb4152 100644 --- a/src/devx/api_clients.py +++ b/src/devx/api_clients.py @@ -223,6 +223,20 @@ class GiteaClient: r = self._request("GET", f"/pulls/{pr_number}/files") return r.json() + def add_pr_label(self, pr_number: str | int, label_names: list[str]) -> None: + """Attach labels to a PR/issue by name. + + Args: + pr_number: PR or issue number. + label_names: List of label names to attach. + """ + self._request("POST", f"/issues/{pr_number}/labels", json={"labels": label_names}) + + def get_pr_label_names(self, pr_number: str | int) -> list[str]: + """Return label names currently attached to a PR/issue.""" + r = self._request("GET", f"/issues/{pr_number}/labels") + return [label.get("name", "") for label in r.json()] + def get_pr_commits(self, pr_number: str | int) -> list[dict[str, Any]]: """Fetch the commits included in a pull request.""" r = self._request("GET", f"/pulls/{pr_number}/commits") @@ -301,6 +315,31 @@ class GiteaClient: return existing return self.create_release(tag=tag, name=name, body=body, draft=draft, prerelease=prerelease) + # -- actions (CI/CD) -- + + def list_action_runs(self, **params: Any) -> dict[str, Any]: + """List workflow runs for the repository. + + Returns the raw API response dict (includes ``workflow_runs`` and + ``total_count`` keys per Gitea API). + """ + r = self._request("GET", "/actions/runs", params=params) + return r.json() + + def get_action_run_jobs(self, run_id: str | int) -> list[dict[str, Any]]: + """List jobs for a specific workflow run.""" + r = self._request("GET", f"/actions/runs/{run_id}/jobs") + data = r.json() + return data.get("jobs", []) + + def get_action_job_logs(self, job_id: str | int) -> str: + """Fetch logs for a specific CI job. + + Returns the raw log text. Raises APIError if logs are unavailable. + """ + r = self._request("GET", f"/actions/jobs/{job_id}/logs") + return r.text + class VikunjaClient: """Low-level Vikunja REST API client with connection pooling.""" @@ -368,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. diff --git a/src/devx/ci/discover_runners.py b/src/devx/ci/discover_runners.py index 7ab703d..3573a21 100644 --- a/src/devx/ci/discover_runners.py +++ b/src/devx/ci/discover_runners.py @@ -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) diff --git a/src/devx/ci/integration_guard.py b/src/devx/ci/integration_guard.py index 42bd0ad..bafdd1a 100644 --- a/src/devx/ci/integration_guard.py +++ b/src/devx/ci/integration_guard.py @@ -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.")) diff --git a/src/devx/ci/publish.py b/src/devx/ci/publish.py index d0a5621..3dea9c7 100644 --- a/src/devx/ci/publish.py +++ b/src/devx/ci/publish.py @@ -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" diff --git a/src/devx/ci/sync_wiki.py b/src/devx/ci/sync_wiki.py index 6c6d9ab..2c3feea 100644 --- a/src/devx/ci/sync_wiki.py +++ b/src/devx/ci/sync_wiki.py @@ -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("/") diff --git a/src/devx/config.py b/src/devx/config.py index b44a4b0..931fa46 100644 --- a/src/devx/config.py +++ b/src/devx/config.py @@ -74,6 +74,7 @@ VIKUNJA_API_URL = _get("vikunja_api_url", "DEVX_VIKUNJA_API_URL", "https://work. # Organization defaults — each project MUST set DEVX_REPO_OWNER explicitly. # No default: prevents silent 404s when the wrong owner is used. REPO_OWNER = _get("repo_owner", "DEVX_REPO_OWNER", "") +REPO_NAME = _get("repo_name", "DEVX_REPO_NAME", "") # Task prefix for Vikunja task IDs — each project sets its own (GRM, DEVX, INFRA, etc.) TASK_PREFIX = _get("task_prefix", "DEVX_TASK_PREFIX", "DEVX") diff --git a/src/devx/make/devx.mak b/src/devx/make/devx.mak index 9dc5df7..a7ad179 100644 --- a/src/devx/make/devx.mak +++ b/src/devx/make/devx.mak @@ -63,6 +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-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 @@ -94,6 +95,35 @@ devx-check-config: # Push and create PR in one step devx-push-with-pr: devx-push devx-create-pr +# Check CI status for a PR (auto-detects current branch's PR) +# Usage: make devx-pr-status +# make devx-pr-status PR=42 +# make devx-pr-status PR=42 WAIT=1 TIMEOUT=600 +devx-pr-status: + @$(DEVX_PYTHON) -m devx.tools.pr_status \ + $(if $(PR),--pr $(PR)) \ + $(if $(WAIT),--wait) \ + $(if $(TIMEOUT),--timeout $(TIMEOUT)) + +# Fetch logs for failed CI jobs on a PR +# Usage: make devx-pr-logs +# make devx-pr-logs PR=42 +# make devx-pr-logs PR=42 JOB=quality TAIL=50 +devx-pr-logs: + @$(DEVX_PYTHON) -m devx.tools.pr_logs \ + $(if $(PR),--pr $(PR)) \ + $(if $(JOB),--job $(JOB)) \ + $(if $(TAIL),--tail $(TAIL)) + +# Add a label to a PR (default: ready-to-merge) +# Usage: make devx-pr-label +# make devx-pr-label PR=42 +# make devx-pr-label PR=42 LABEL=ready-to-merge +devx-pr-label: + @$(DEVX_PYTHON) -m devx.tools.pr_label \ + $(if $(PR),--pr $(PR)) \ + --label $(or $(LABEL),ready-to-merge) + # ── Environment setup ───────────────────────────────────────────────────────── # Configure Gitea private PyPI registry so pip can find devx and other diff --git a/src/devx/molecule/discover_runners.py b/src/devx/molecule/discover_runners.py index b61c7b4..d16688b 100644 --- a/src/devx/molecule/discover_runners.py +++ b/src/devx/molecule/discover_runners.py @@ -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) diff --git a/src/devx/molecule/molecule_ci_guard.py b/src/devx/molecule/molecule_ci_guard.py index e043a3b..7724b5f 100644 --- a/src/devx/molecule/molecule_ci_guard.py +++ b/src/devx/molecule/molecule_ci_guard.py @@ -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.")) diff --git a/src/devx/tools/check_config.py b/src/devx/tools/check_config.py index ce91d5b..dc30cba 100644 --- a/src/devx/tools/check_config.py +++ b/src/devx/tools/check_config.py @@ -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.")) diff --git a/src/devx/tools/check_test_coverage.py b/src/devx/tools/check_test_coverage.py index 84fe7ab..79454af 100644 --- a/src/devx/tools/check_test_coverage.py +++ b/src/devx/tools/check_test_coverage.py @@ -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 diff --git a/src/devx/tools/clean_images.py b/src/devx/tools/clean_images.py index b25d49d..e09d542 100644 --- a/src/devx/tools/clean_images.py +++ b/src/devx/tools/clean_images.py @@ -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 diff --git a/src/devx/tools/configure_repo.py b/src/devx/tools/configure_repo.py index 1db27c2..a36f248 100644 --- a/src/devx/tools/configure_repo.py +++ b/src/devx/tools/configure_repo.py @@ -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.")) diff --git a/src/devx/tools/create_pr.py b/src/devx/tools/create_pr.py index 3e68f46..a1d95b4 100644 --- a/src/devx/tools/create_pr.py +++ b/src/devx/tools/create_pr.py @@ -34,6 +34,7 @@ from devx.api_clients import GiteaClient, VikunjaClient from devx.config import ( DEFAULT_PER_PAGE, GITEA_API_URL, + REPO_NAME, REPO_OWNER, TASK_ID_RE, TASK_PREFIX, @@ -46,15 +47,17 @@ load_dotenv() def get_repo_name() -> str: - """Auto-detect repository name from env vars or git remote.""" + """Auto-detect repository name from env vars, pyproject.toml, or git remote.""" name = os.environ.get("DEVX_REPO_NAME", "") if name: return name github_repo = os.environ.get("GITHUB_REPOSITORY", "") if github_repo and "/" in github_repo: return github_repo.split("/", 1)[1] + if REPO_NAME: + return REPO_NAME raise click.ClickException( - _("Repository name not set. Use DEVX_REPO_NAME or GITHUB_REPOSITORY env var."), + _("Repository name not set. Use DEVX_REPO_NAME, [tool.devx] repo_name, or GITHUB_REPOSITORY env var."), ) @@ -73,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: diff --git a/src/devx/tools/generate_cliff_config.py b/src/devx/tools/generate_cliff_config.py index 70cd37b..3e77843 100644 --- a/src/devx/tools/generate_cliff_config.py +++ b/src/devx/tools/generate_cliff_config.py @@ -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", diff --git a/src/devx/tools/install_checkmake.py b/src/devx/tools/install_checkmake.py index 743cdc1..0c0ae6d 100644 --- a/src/devx/tools/install_checkmake.py +++ b/src/devx/tools/install_checkmake.py @@ -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 diff --git a/src/devx/tools/pr_label.py b/src/devx/tools/pr_label.py new file mode 100644 index 0000000..fb3b581 --- /dev/null +++ b/src/devx/tools/pr_label.py @@ -0,0 +1,81 @@ +#!/usr/bin/env python3 +"""Add a label to a pull request (idempotent). + +Commonly used to add the ``ready-to-merge`` label after CI passes and +review is complete. The operation is idempotent — if the label is already +attached, it succeeds without error. + +Usage:: + + # Add ready-to-merge to PR #42 + python -m devx.tools.pr_label --pr 42 --label ready-to-merge + + # Add label to current branch's PR + python -m devx.tools.pr_label --label ready-to-merge + + # Add multiple labels + python -m devx.tools.pr_label --pr 42 --label ready-to-merge --label reviewed + +The repository is auto-detected from ``DEVX_REPO_OWNER`` / +``DEVX_REPO_NAME`` or ``GITHUB_REPOSITORY`` environment variables. +""" + +from __future__ import annotations + +import os + +import click +from dotenv import load_dotenv + +from devx.api_clients import GiteaClient +from devx.config import GITEA_API_URL, REPO_OWNER +from devx.i18n import _ +from devx.tools.create_pr import get_repo_name +from devx.tools.pr_status import _get_current_branch_pr + +load_dotenv() + + +@click.command() +@click.option("--pr", "pr_number", type=int, default=None, help="PR number (default: auto-detect from current branch).") +@click.option("--label", "labels", multiple=True, required=True, help="Label name(s) to add (can be repeated).") +@click.option("--owner", default=None, help="Repository owner (default: DEVX_REPO_OWNER).") +@click.option("--repo", default=None, help="Repository name (default: DEVX_REPO_NAME or GITHUB_REPOSITORY).") +def cli( + pr_number: int | None, + labels: tuple[str, ...], + owner: str | None, + repo: str | None, +) -> None: + """Add one or more labels to a pull request (idempotent).""" + token = os.environ.get("CI_GITEA_TOKEN", "") + if not token: + raise click.ClickException(_("CI_GITEA_TOKEN is not set.")) + + repo_owner = owner or REPO_OWNER + if not repo_owner: + raise click.ClickException(_("Repository owner not set. Use --owner or DEVX_REPO_OWNER env var.")) + repo_name = repo or get_repo_name() + + client = GiteaClient(GITEA_API_URL, token, repo_owner, repo_name) + + if pr_number is None: + pr_number = _get_current_branch_pr(client) + + label_list = list(labels) + existing = client.get_pr_label_names(pr_number) + to_add = [lbl for lbl in label_list if lbl not in existing] + already = [lbl for lbl in label_list if lbl in existing] + + if already: + for lbl in already: + click.echo(_("Label '{label}' already on PR #{pr}.", label=lbl, pr=pr_number)) + + if to_add: + client.add_pr_label(pr_number, to_add) + for lbl in to_add: + click.echo(_("Added label '{label}' to PR #{pr}.", label=lbl, pr=pr_number)) + + +if __name__ == "__main__": # pragma: no cover + cli() # pragma: no cover diff --git a/src/devx/tools/pr_logs.py b/src/devx/tools/pr_logs.py new file mode 100644 index 0000000..4c1cadf --- /dev/null +++ b/src/devx/tools/pr_logs.py @@ -0,0 +1,187 @@ +#!/usr/bin/env python3 +"""Fetch logs for failed CI jobs on a pull request. + +Lists CI jobs for the latest workflow run of a PR's branch, then fetches +and prints the logs of any failed jobs. Useful for diagnosing CI failures +without navigating the web UI. + +Usage:: + + # Show failed job logs for PR #42 + python -m devx.tools.pr_logs --pr 42 + + # Show failed job logs for current branch's PR + python -m devx.tools.pr_logs + + # Show logs for a specific job (by name) + python -m devx.tools.pr_logs --pr 42 --job quality + + # Show last N lines of each failed job's logs + python -m devx.tools.pr_logs --pr 42 --tail 50 + +The repository is auto-detected from ``DEVX_REPO_OWNER`` / +``DEVX_REPO_NAME`` or ``GITHUB_REPOSITORY`` environment variables. +""" + +from __future__ import annotations + +import os + +import click +from dotenv import load_dotenv + +from devx.api_clients import APIError, GiteaClient +from devx.config import GITEA_API_URL, REPO_OWNER +from devx.i18n import _ +from devx.tools.create_pr import get_repo_name +from devx.tools.pr_status import _get_current_branch_pr + +load_dotenv() + + +def _get_pr_sha(client: GiteaClient, pr_number: int) -> str: + """Fetch the head SHA of a PR.""" + pr = client.get_pr(pr_number) + return pr.get("head", {}).get("sha", "") + + +def _find_latest_run_by_sha(client: GiteaClient, sha: str) -> dict | None: + """Find the latest workflow run for a commit SHA. + + Gitea Actions API doesn't set head_branch for pull_request events, + so we filter by head_sha instead. + """ + data = client.list_action_runs(limit=50) + for run in data.get("workflow_runs", []): + if run.get("head_sha", "").startswith(sha): + return run + return None + + +def _find_failed_jobs(jobs: list[dict]) -> list[dict]: + """Return jobs with conclusion 'failure'.""" + return [j for j in jobs if j.get("conclusion") == "failure"] + + +def _find_job_by_name(jobs: list[dict], name: str) -> dict | None: + """Find a job by name (case-insensitive partial match).""" + name_lower = name.lower() + for j in jobs: + if name_lower in j.get("name", "").lower(): + return j + return None + + +def _print_job_summary(jobs: list[dict]) -> None: + """Print a summary table of all jobs and their status.""" + for j in jobs: + name = j.get("name", "?") + conclusion = j.get("conclusion", "pending") + status = j.get("status", "?") + symbol = "[FAIL]" if conclusion == "failure" else "[OK]" if conclusion == "success" else f"[{conclusion}]" + click.echo(f" {symbol} {name} (status: {status}, conclusion: {conclusion})") + + +def _print_failed_steps(job: dict) -> list[int]: + """Print failed steps for a job. Returns list of failed step numbers.""" + failed_steps = [] + for step in job.get("steps", []): + if step.get("conclusion") == "failure": + name = step.get("name", "?") + num = step.get("number", "?") + click.echo(f" FAILED step #{num}: {name}") + failed_steps.append(num) + return failed_steps + + +def _print_logs(client: GiteaClient, job_id: int, tail: int = 0) -> None: + """Fetch and print logs for a job. If tail > 0, print only last N lines.""" + try: + logs = client.get_action_job_logs(job_id) + except APIError as e: + click.echo(_(" Could not fetch logs: {error}", error=str(e))) + return + + if tail > 0: + lines = logs.strip().split("\n") + if len(lines) > tail: + click.echo(f" ... (showing last {tail} of {len(lines)} lines)") + logs = "\n".join(lines[-tail:]) + + for line in logs.split("\n"): + click.echo(f" {line}") + + +@click.command() +@click.option("--pr", "pr_number", type=int, default=None, help="PR number (default: auto-detect from current branch).") +@click.option("--job", default=None, help="Job name to show logs for (partial match, case-insensitive).") +@click.option("--tail", type=int, default=80, show_default=True, help="Show last N lines of logs (0 = all).") +@click.option("--owner", default=None, help="Repository owner (default: DEVX_REPO_OWNER).") +@click.option("--repo", default=None, help="Repository name (default: DEVX_REPO_NAME or GITHUB_REPOSITORY).") +def cli( + pr_number: int | None, + job: str | None, + tail: int, + owner: str | None, + repo: str | None, +) -> None: + """Fetch logs for failed CI jobs on a pull request.""" + token = os.environ.get("CI_GITEA_TOKEN", "") + if not token: + raise click.ClickException(_("CI_GITEA_TOKEN is not set.")) + + repo_owner = owner or REPO_OWNER + if not repo_owner: + raise click.ClickException(_("Repository owner not set. Use --owner or DEVX_REPO_OWNER env var.")) + repo_name = repo or get_repo_name() + + client = GiteaClient(GITEA_API_URL, token, repo_owner, repo_name) + + if pr_number is None: + pr_number = _get_current_branch_pr(client) + click.echo(_("Fetching logs for PR #{pr_number}...", pr_number=pr_number)) + + sha = _get_pr_sha(client, pr_number) + if not sha: + raise click.ClickException(_("Could not determine head SHA for PR #{pr_number}.", pr_number=pr_number)) + + run = _find_latest_run_by_sha(client, sha) + if not run: + raise click.ClickException(_("No workflow runs found for SHA {sha}.", sha=sha[:8])) + + run_id = run.get("id", 0) + run_status = run.get("status", "?") + click.echo(_("Latest run: #{run_id} (status: {status})", run_id=run_id, status=run_status)) + click.echo("") + + jobs = client.get_action_run_jobs(run_id) + if not jobs: + click.echo(_("No jobs found for run #{run_id}.", run_id=run_id)) + return + + _print_job_summary(jobs) + click.echo("") + + if job: + target = _find_job_by_name(jobs, job) + if not target: + raise click.ClickException(_("No job matching '{job}' found.", job=job)) + click.echo(f"Logs for job '{target.get('name', '?')}' (id={target.get('id')}):") + _print_failed_steps(target) + click.echo("") + _print_logs(client, target["id"], tail) + else: + failed = _find_failed_jobs(jobs) + if not failed: + click.echo(_("No failed jobs.")) + return + for fj in failed: + click.echo(f"Logs for failed job '{fj.get('name', '?')}' (id={fj.get('id')}):") + _print_failed_steps(fj) + click.echo("") + _print_logs(client, fj["id"], tail) + click.echo("") + + +if __name__ == "__main__": # pragma: no cover + cli() # pragma: no cover diff --git a/src/devx/tools/pr_status.py b/src/devx/tools/pr_status.py new file mode 100644 index 0000000..3df8245 --- /dev/null +++ b/src/devx/tools/pr_status.py @@ -0,0 +1,174 @@ +#!/usr/bin/env python3 +"""Check CI status for a pull request or commit. + +Displays the status of all CI checks for a PR (or a specific commit SHA). +Optionally polls until all checks complete (``--wait``). + +Usage:: + + # Check status of PR #42 + python -m devx.tools.pr_status --pr 42 + + # Check status of current branch's PR + python -m devx.tools.pr_status + + # Wait for all checks to complete (timeout 600s) + python -m devx.tools.pr_status --pr 42 --wait --timeout 600 + + # Check a specific commit SHA + python -m devx.tools.pr_status --sha abc1234 + +The repository is auto-detected from ``DEVX_REPO_OWNER`` / +``DEVX_REPO_NAME`` or ``GITHUB_REPOSITORY`` environment variables. +""" + +from __future__ import annotations + +import os +import subprocess # nosec B404 +import time + +import click +from dotenv import load_dotenv + +from devx.api_clients import GiteaClient +from devx.config import GITEA_API_URL, REPO_OWNER +from devx.i18n import _ +from devx.tools.create_pr import get_repo_name + +load_dotenv() + +# Status symbols for terminal output +_STATUS_SYMBOLS = { + "success": "[OK]", + "failure": "[FAIL]", + "error": "[FAIL]", + "pending": "[..]", + "skipped": "[SKIP]", + "none": "[--]", +} + + +def _get_symbol(status: str) -> str: + return _STATUS_SYMBOLS.get(status, f"[{status}]") + + +def _get_pr_sha(client: GiteaClient, pr_number: int) -> str: + """Fetch the head SHA of a PR.""" + pr = client.get_pr(pr_number) + return pr.get("head", {}).get("sha", "") + + +def _get_current_branch_pr(client: GiteaClient) -> int: + """Find the open PR for the current git branch.""" + result = subprocess.run( # nosec + ["git", "rev-parse", "--abbrev-ref", "HEAD"], + capture_output=True, + text=True, + check=False, + ) + if result.returncode != 0: + raise click.ClickException(_("Could not detect current branch: {error}", error=result.stderr.strip())) + branch = result.stdout.strip() + + prs = client.list_prs(state="open") + for pr in prs: + if pr.get("head", {}).get("ref") == branch: + return int(pr["number"]) + raise click.ClickException(_("No open PR found for branch '{branch}'.", branch=branch)) + + +def print_status(client: GiteaClient, sha: str) -> str: + """Print CI check statuses for a commit SHA. Returns the overall state.""" + statuses = client.get_commit_status(sha) + if not statuses: + click.echo(_("No CI checks found for commit {sha}.", sha=sha[:8])) + return "none" + + overall = "success" + for s in statuses: + context = s.get("context", "?") + status = s.get("status", "pending") + symbol = _get_symbol(status) + click.echo(f" {symbol} {context}") + if status in ("failure", "error"): + overall = "failure" + elif status == "pending" and overall != "failure": + overall = "pending" + elif status == "skipped" and overall == "success": + overall = "success" + + click.echo(f"\n Overall: {_get_symbol(overall)} {overall}") + return overall + + +def wait_for_completion( + client: GiteaClient, + sha: str, + timeout: int = 600, + interval: int = 30, +) -> str: + """Poll CI status until all checks complete or timeout. Returns final state.""" + click.echo(_("Waiting for CI checks to complete (timeout: {timeout}s)...", timeout=timeout)) + deadline = time.time() + timeout + while time.time() < deadline: + state = print_status(client, sha) + if state in ("success", "failure", "error", "none"): + return state + click.echo(f" ...still pending, retrying in {interval}s\n") + time.sleep(interval) + click.echo(_("Timeout reached after {timeout}s.", timeout=timeout)) + return "pending" + + +@click.command() +@click.option("--pr", "pr_number", type=int, default=None, help="PR number (default: auto-detect from current branch).") +@click.option("--sha", default=None, help="Commit SHA to check (alternative to --pr).") +@click.option("--wait", "do_wait", is_flag=True, help="Poll until all checks complete.") +@click.option("--timeout", type=int, default=600, show_default=True, help="Wait timeout in seconds.") +@click.option("--interval", type=int, default=30, show_default=True, help="Poll interval in seconds.") +@click.option("--owner", default=None, help="Repository owner (default: DEVX_REPO_OWNER).") +@click.option("--repo", default=None, help="Repository name (default: DEVX_REPO_NAME or GITHUB_REPOSITORY).") +def cli( + pr_number: int | None, + sha: str | None, + do_wait: bool, + timeout: int, + interval: int, + owner: str | None, + repo: str | None, +) -> None: + """Check CI status for a pull request or commit.""" + token = os.environ.get("CI_GITEA_TOKEN", "") + if not token: + raise click.ClickException(_("CI_GITEA_TOKEN is not set.")) + + repo_owner = owner or REPO_OWNER + if not repo_owner: + raise click.ClickException(_("Repository owner not set. Use --owner or DEVX_REPO_OWNER env var.")) + repo_name = repo or get_repo_name() + + client = GiteaClient(GITEA_API_URL, token, repo_owner, repo_name) + + if sha is None: + if pr_number is None: + pr_number = _get_current_branch_pr(client) + click.echo(_("Checking status for PR #{pr_number}...", pr_number=pr_number)) + pr = client.get_pr(pr_number) + sha = pr.get("head", {}).get("sha", "") + if not sha: + raise click.ClickException(_("Could not determine head SHA for PR #{pr_number}.", pr_number=pr_number)) + + click.echo(_("Commit: {sha}", sha=sha[:12])) + click.echo("") + + state = wait_for_completion(client, sha, timeout, interval) if do_wait else print_status(client, sha) + + if state in ("failure", "error"): + raise click.ClickException(_("CI checks failed.")) + if state == "pending" and do_wait: + raise click.ClickException(_("CI checks did not complete within timeout.")) + + +if __name__ == "__main__": # pragma: no cover + cli() # pragma: no cover diff --git a/src/devx/tools/pre_push_check.py b/src/devx/tools/pre_push_check.py index 9a328eb..952b8b1 100644 --- a/src/devx/tools/pre_push_check.py +++ b/src/devx/tools/pre_push_check.py @@ -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: diff --git a/src/devx/translations.json b/src/devx/translations.json index 5e768f3..87f84c9 100644 --- a/src/devx/translations.json +++ b/src/devx/translations.json @@ -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...", @@ -1471,14 +1463,6 @@ "ru": "Repository in owner/name format", "zh": "Repository in owner/name format" }, - "Repository name not set. Use DEVX_REPO_NAME or GITHUB_REPOSITORY env var.": { - "bg": "Името на хранилището не е зададено. Използвайте DEVX_REPO_NAME или GITHUB_REPOSITORY env var.", - "de": "Repository-Name nicht gesetzt. Verwende DEVX_REPO_NAME oder GITHUB_REPOSITORY env var.", - "en": "Repository name not set. Use DEVX_REPO_NAME or GITHUB_REPOSITORY env var.", - "pl": "Nazwa repozytorium nie jest ustawiona. Użyj DEVX_REPO_NAME lub GITHUB_REPOSITORY env var.", - "ru": "Имя репозитория не установлено. Используйте DEVX_REPO_NAME или GITHUB_REPOSITORY env var.", - "zh": "仓库名称未设置。使用 DEVX_REPO_NAME 或 GITHUB_REPOSITORY 环境变量。" - }, "Repository owner not set. Use --owner or DEVX_REPO_OWNER env var.": { "bg": "Собственикът на хранилището не е зададен. Използвайте --owner или DEVX_REPO_OWNER env var.", "de": "Repository-Owner nicht gesetzt. Verwende --owner oder DEVX_REPO_OWNER env var.", @@ -2038,5 +2022,197 @@ "pl": "Configuring tea login '{name}' for {url}...", "ru": "Configuring tea login '{name}' for {url}...", "zh": "Configuring tea login '{name}' for {url}..." + }, + " Could not fetch logs: {error}": { + "en": " Could not fetch logs: {error}", + "bg": " Could not fetch logs: {error}", + "de": " Could not fetch logs: {error}", + "pl": " Could not fetch logs: {error}", + "ru": " Could not fetch logs: {error}", + "zh": " Could not fetch logs: {error}" + }, + "Added label '{label}' to PR #{pr}.": { + "en": "Added label '{label}' to PR #{pr}.", + "bg": "Added label '{label}' to PR #{pr}.", + "de": "Added label '{label}' to PR #{pr}.", + "pl": "Added label '{label}' to PR #{pr}.", + "ru": "Added label '{label}' to PR #{pr}.", + "zh": "Added label '{label}' to PR #{pr}." + }, + "CI checks did not complete within timeout.": { + "en": "CI checks did not complete within timeout.", + "bg": "CI checks did not complete within timeout.", + "de": "CI checks did not complete within timeout.", + "pl": "CI checks did not complete within timeout.", + "ru": "CI checks did not complete within timeout.", + "zh": "CI checks did not complete within timeout." + }, + "CI checks failed.": { + "en": "CI checks failed.", + "bg": "CI checks failed.", + "de": "CI checks failed.", + "pl": "CI checks failed.", + "ru": "CI checks failed.", + "zh": "CI checks failed." + }, + "CI_GITEA_TOKEN is not set.": { + "en": "CI_GITEA_TOKEN is not set.", + "bg": "CI_GITEA_TOKEN is not set.", + "de": "CI_GITEA_TOKEN is not set.", + "pl": "CI_GITEA_TOKEN is not set.", + "ru": "CI_GITEA_TOKEN is not set.", + "zh": "CI_GITEA_TOKEN is not set." + }, + "Checking status for PR #{pr_number}...": { + "en": "Checking status for PR #{pr_number}...", + "bg": "Checking status for PR #{pr_number}...", + "de": "Checking status for PR #{pr_number}...", + "pl": "Checking status for PR #{pr_number}...", + "ru": "Checking status for PR #{pr_number}...", + "zh": "Checking status for PR #{pr_number}..." + }, + "Commit: {sha}": { + "en": "Commit: {sha}", + "bg": "Commit: {sha}", + "de": "Commit: {sha}", + "pl": "Commit: {sha}", + "ru": "Commit: {sha}", + "zh": "Commit: {sha}" + }, + "Could not determine head SHA for PR #{pr_number}.": { + "en": "Could not determine head SHA for PR #{pr_number}.", + "bg": "Could not determine head SHA for PR #{pr_number}.", + "de": "Could not determine head SHA for PR #{pr_number}.", + "pl": "Could not determine head SHA for PR #{pr_number}.", + "ru": "Could not determine head SHA for PR #{pr_number}.", + "zh": "Could not determine head SHA for PR #{pr_number}." + }, + "Fetching logs for PR #{pr_number}...": { + "en": "Fetching logs for PR #{pr_number}...", + "bg": "Fetching logs for PR #{pr_number}...", + "de": "Fetching logs for PR #{pr_number}...", + "pl": "Fetching logs for PR #{pr_number}...", + "ru": "Fetching logs for PR #{pr_number}...", + "zh": "Fetching logs for PR #{pr_number}..." + }, + "Label '{label}' already on PR #{pr}.": { + "en": "Label '{label}' already on PR #{pr}.", + "bg": "Label '{label}' already on PR #{pr}.", + "de": "Label '{label}' already on PR #{pr}.", + "pl": "Label '{label}' already on PR #{pr}.", + "ru": "Label '{label}' already on PR #{pr}.", + "zh": "Label '{label}' already on PR #{pr}." + }, + "Latest run: #{run_id} (status: {status})": { + "en": "Latest run: #{run_id} (status: {status})", + "bg": "Latest run: #{run_id} (status: {status})", + "de": "Latest run: #{run_id} (status: {status})", + "pl": "Latest run: #{run_id} (status: {status})", + "ru": "Latest run: #{run_id} (status: {status})", + "zh": "Latest run: #{run_id} (status: {status})" + }, + "No CI checks found for commit {sha}.": { + "en": "No CI checks found for commit {sha}.", + "bg": "No CI checks found for commit {sha}.", + "de": "No CI checks found for commit {sha}.", + "pl": "No CI checks found for commit {sha}.", + "ru": "No CI checks found for commit {sha}.", + "zh": "No CI checks found for commit {sha}." + }, + "No failed jobs.": { + "en": "No failed jobs.", + "bg": "No failed jobs.", + "de": "No failed jobs.", + "pl": "No failed jobs.", + "ru": "No failed jobs.", + "zh": "No failed jobs." + }, + "No job matching '{job}' found.": { + "en": "No job matching '{job}' found.", + "bg": "No job matching '{job}' found.", + "de": "No job matching '{job}' found.", + "pl": "No job matching '{job}' found.", + "ru": "No job matching '{job}' found.", + "zh": "No job matching '{job}' found." + }, + "No jobs found for run #{run_id}.": { + "en": "No jobs found for run #{run_id}.", + "bg": "No jobs found for run #{run_id}.", + "de": "No jobs found for run #{run_id}.", + "pl": "No jobs found for run #{run_id}.", + "ru": "No jobs found for run #{run_id}.", + "zh": "No jobs found for run #{run_id}." + }, + "No open PR found for branch '{branch}'.": { + "en": "No open PR found for branch '{branch}'.", + "bg": "No open PR found for branch '{branch}'.", + "de": "No open PR found for branch '{branch}'.", + "pl": "No open PR found for branch '{branch}'.", + "ru": "No open PR found for branch '{branch}'.", + "zh": "No open PR found for branch '{branch}'." + }, + "No workflow runs found for SHA {sha}.": { + "en": "No workflow runs found for SHA {sha}.", + "bg": "No workflow runs found for SHA {sha}.", + "de": "No workflow runs found for SHA {sha}.", + "pl": "No workflow runs found for SHA {sha}.", + "ru": "No workflow runs found for SHA {sha}.", + "zh": "No workflow runs found for SHA {sha}." + }, + "Repository name not set. Use DEVX_REPO_NAME, [tool.devx] repo_name, or GITHUB_REPOSITORY env var.": { + "en": "Repository name not set. Use DEVX_REPO_NAME, [tool.devx] repo_name, or GITHUB_REPOSITORY env var.", + "bg": "Repository name not set. Use DEVX_REPO_NAME, [tool.devx] repo_name, or GITHUB_REPOSITORY env var.", + "de": "Repository name not set. Use DEVX_REPO_NAME, [tool.devx] repo_name, or GITHUB_REPOSITORY env var.", + "pl": "Repository name not set. Use DEVX_REPO_NAME, [tool.devx] repo_name, or GITHUB_REPOSITORY env var.", + "ru": "Repository name not set. Use DEVX_REPO_NAME, [tool.devx] repo_name, or GITHUB_REPOSITORY env var.", + "zh": "Repository name not set. Use DEVX_REPO_NAME, [tool.devx] repo_name, or GITHUB_REPOSITORY env var." + }, + "Timeout reached after {timeout}s.": { + "en": "Timeout reached after {timeout}s.", + "bg": "Timeout reached after {timeout}s.", + "de": "Timeout reached after {timeout}s.", + "pl": "Timeout reached after {timeout}s.", + "ru": "Timeout reached after {timeout}s.", + "zh": "Timeout reached after {timeout}s." + }, + "Waiting for CI checks to complete (timeout: {timeout}s)...": { + "en": "Waiting for CI checks to complete (timeout: {timeout}s)...", + "bg": "Waiting for CI checks to complete (timeout: {timeout}s)...", + "de": "Waiting for CI checks to complete (timeout: {timeout}s)...", + "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." } } diff --git a/tests/unit/test_api_clients.py b/tests/unit/test_api_clients.py index 8b024a9..ee658f7 100644 --- a/tests/unit/test_api_clients.py +++ b/tests/unit/test_api_clients.py @@ -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: @@ -797,5 +833,88 @@ class TestIsRetryable: err = _mock_http_error(404, "not found") assert _is_retryable(err) is False + +class TestGiteaClientPrLabels: + def test_add_pr_label(self) -> None: + client = GiteaClient("https://git.example.com", "tok", "owner", "repo") + client._session.request = MagicMock(return_value=_mock_response({})) + client.add_pr_label(42, ["ready-to-merge"]) + client._session.request.assert_called_once_with( + "POST", + "https://git.example.com/repos/owner/repo/issues/42/labels", + timeout=DEFAULT_TIMEOUT, + json={"labels": ["ready-to-merge"]}, + ) + + def test_add_pr_label_multiple(self) -> None: + client = GiteaClient("https://git.example.com", "tok", "owner", "repo") + client._session.request = MagicMock(return_value=_mock_response({})) + client.add_pr_label(42, ["ready-to-merge", "reviewed"]) + call_kwargs = client._session.request.call_args.kwargs + assert call_kwargs["json"]["labels"] == ["ready-to-merge", "reviewed"] + + def test_get_pr_label_names(self) -> None: + client = GiteaClient("https://git.example.com", "tok", "owner", "repo") + client._session.request = MagicMock(return_value=_mock_response([{"name": "bug"}, {"name": "ready-to-merge"}])) + result = client.get_pr_label_names(42) + assert result == ["bug", "ready-to-merge"] + client._session.request.assert_called_once_with( + "GET", + "https://git.example.com/repos/owner/repo/issues/42/labels", + timeout=DEFAULT_TIMEOUT, + ) + + +class TestGiteaClientActions: + def test_list_action_runs(self) -> None: + client = GiteaClient("https://git.example.com", "tok", "owner", "repo") + client._session.request = MagicMock( + return_value=_mock_response({"workflow_runs": [{"id": 1, "status": "completed"}], "total_count": 1}) + ) + result = client.list_action_runs(branch="feature-branch", limit=1) + assert result["total_count"] == 1 + client._session.request.assert_called_once_with( + "GET", + "https://git.example.com/repos/owner/repo/actions/runs", + timeout=DEFAULT_TIMEOUT, + params={"branch": "feature-branch", "limit": 1}, + ) + + def test_get_action_run_jobs(self) -> None: + client = GiteaClient("https://git.example.com", "tok", "owner", "repo") + client._session.request = MagicMock( + return_value=_mock_response({"jobs": [{"id": 100, "name": "quality", "conclusion": "failure"}]}) + ) + result = client.get_action_run_jobs(1410) + assert len(result) == 1 + assert result[0]["name"] == "quality" + client._session.request.assert_called_once_with( + "GET", + "https://git.example.com/repos/owner/repo/actions/runs/1410/jobs", + timeout=DEFAULT_TIMEOUT, + ) + + def test_get_action_run_jobs_empty(self) -> None: + client = GiteaClient("https://git.example.com", "tok", "owner", "repo") + client._session.request = MagicMock(return_value=_mock_response({})) + result = client.get_action_run_jobs(1410) + assert result == [] + + def test_get_action_job_logs(self) -> None: + client = GiteaClient("https://git.example.com", "tok", "owner", "repo") + mock_resp = MagicMock() + mock_resp.text = "log line 1\nlog line 2" + mock_resp.raise_for_status = MagicMock() + client._session.request = MagicMock(return_value=mock_resp) + result = client.get_action_job_logs(10026) + assert "log line 1" in result + client._session.request.assert_called_once_with( + "GET", + "https://git.example.com/repos/owner/repo/actions/jobs/10026/logs", + timeout=DEFAULT_TIMEOUT, + ) + + +class TestIsRetryableGeneric: def test_generic_exception_is_not_retryable(self) -> None: assert _is_retryable(ValueError("oops")) is False diff --git a/tests/unit/test_build_image.py b/tests/unit/test_build_image.py index 1fd40cc..20ac496 100644 --- a/tests/unit/test_build_image.py +++ b/tests/unit/test_build_image.py @@ -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() diff --git a/tests/unit/test_check_test_coverage.py b/tests/unit/test_check_test_coverage.py index e0cb60a..e5ecbba 100644 --- a/tests/unit/test_check_test_coverage.py +++ b/tests/unit/test_check_test_coverage.py @@ -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 diff --git a/tests/unit/test_configure_repo.py b/tests/unit/test_configure_repo.py index 24d8e50..3d8f559 100644 --- a/tests/unit/test_configure_repo.py +++ b/tests/unit/test_configure_repo.py @@ -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 diff --git a/tests/unit/test_create_pr.py b/tests/unit/test_create_pr.py index a6f0f56..5c08df0 100644 --- a/tests/unit/test_create_pr.py +++ b/tests/unit/test_create_pr.py @@ -29,10 +29,22 @@ class TestGetRepoName: def test_from_env(self) -> None: assert get_repo_name() == "infra" + @patch("devx.tools.create_pr.REPO_NAME", "devx") + @patch.dict("os.environ", {"GITHUB_REPOSITORY": "oblachno/infra"}, clear=True) + def test_env_overrides_pyproject(self) -> None: + assert get_repo_name() == "infra" + + @patch("devx.tools.create_pr.REPO_NAME", "devx") + @patch.dict("os.environ", {}, clear=True) + def test_from_pyproject(self) -> None: + assert get_repo_name() == "devx" + + @patch("devx.tools.create_pr.REPO_NAME", "") @patch.dict("os.environ", {"GITHUB_REPOSITORY": "oblachno/infra"}, clear=True) def test_from_github(self) -> None: assert get_repo_name() == "infra" + @patch("devx.tools.create_pr.REPO_NAME", "") @patch.dict("os.environ", {}, clear=True) def test_missing_raises(self) -> None: with pytest.raises(click.ClickException, match="Repository name"): @@ -44,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" @@ -57,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") diff --git a/tests/unit/test_install_checkmake.py b/tests/unit/test_install_checkmake.py index 73e6360..50b351f 100644 --- a/tests/unit/test_install_checkmake.py +++ b/tests/unit/test_install_checkmake.py @@ -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() diff --git a/tests/unit/test_pr_label.py b/tests/unit/test_pr_label.py new file mode 100644 index 0000000..10faca6 --- /dev/null +++ b/tests/unit/test_pr_label.py @@ -0,0 +1,84 @@ +"""Unit tests for devx.tools.pr_label.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest +from click.testing import CliRunner + +from devx.tools.pr_label import cli + + +class TestCli: + def test_no_token_raises(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("CI_GITEA_TOKEN", raising=False) + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42", "--label", "ready-to-merge"]) + assert result.exit_code != 0 + assert "CI_GITEA_TOKEN" in result.output + + @patch("devx.tools.pr_label.REPO_OWNER", "") + def test_no_owner_raises(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42", "--label", "ready-to-merge"]) + assert result.exit_code != 0 + assert "owner" in result.output.lower() + + @patch("devx.tools.pr_label.GiteaClient") + def test_adds_new_label(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + client = mock_client_cls.return_value + client.get_pr_label_names.return_value = [] + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42", "--label", "ready-to-merge"]) + assert result.exit_code == 0 + client.add_pr_label.assert_called_once_with(42, ["ready-to-merge"]) + assert "Added label" in result.output + + @patch("devx.tools.pr_label.GiteaClient") + def test_skips_existing_label(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + client = mock_client_cls.return_value + client.get_pr_label_names.return_value = ["ready-to-merge"] + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42", "--label", "ready-to-merge"]) + assert result.exit_code == 0 + client.add_pr_label.assert_not_called() + assert "already" in result.output + + @patch("devx.tools.pr_label.GiteaClient") + def test_mixed_new_and_existing(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + client = mock_client_cls.return_value + client.get_pr_label_names.return_value = ["reviewed"] + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42", "--label", "ready-to-merge", "--label", "reviewed"]) + assert result.exit_code == 0 + client.add_pr_label.assert_called_once_with(42, ["ready-to-merge"]) + assert "Added label" in result.output + assert "already" in result.output + + @patch("devx.tools.pr_label.GiteaClient") + @patch("devx.tools.pr_status.subprocess.run") + def test_auto_detect_pr( + self, mock_subprocess: MagicMock, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + mock_subprocess.return_value = MagicMock(returncode=0, stdout="feature-branch\n") + client = mock_client_cls.return_value + client.list_prs.return_value = [{"number": 42, "head": {"ref": "feature-branch"}}] + client.get_pr_label_names.return_value = [] + runner = CliRunner() + result = runner.invoke(cli, ["--label", "ready-to-merge"]) + assert result.exit_code == 0 + client.add_pr_label.assert_called_once_with(42, ["ready-to-merge"]) diff --git a/tests/unit/test_pr_logs.py b/tests/unit/test_pr_logs.py new file mode 100644 index 0000000..aae92a2 --- /dev/null +++ b/tests/unit/test_pr_logs.py @@ -0,0 +1,312 @@ +"""Unit tests for devx.tools.pr_logs.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest +from click.testing import CliRunner + +from devx.api_clients import APIError, GiteaClient +from devx.tools.pr_logs import ( + _find_failed_jobs, + _find_job_by_name, + _find_latest_run_by_sha, + _get_pr_sha, + _print_failed_steps, + _print_job_summary, + _print_logs, + cli, +) + + +class TestGetPrSha: + def test_returns_sha(self) -> None: + client = MagicMock(spec=GiteaClient) + client.get_pr.return_value = {"head": {"sha": "abc123"}} + assert _get_pr_sha(client, 42) == "abc123" + + def test_returns_empty_when_missing(self) -> None: + client = MagicMock(spec=GiteaClient) + client.get_pr.return_value = {"head": {}} + assert _get_pr_sha(client, 42) == "" + + +class TestFindLatestRunBySha: + def test_returns_matching_run(self) -> None: + client = MagicMock(spec=GiteaClient) + client.list_action_runs.return_value = { + "workflow_runs": [ + {"id": 2, "head_sha": "def456"}, + {"id": 1, "head_sha": "abc123def"}, + ], + } + result = _find_latest_run_by_sha(client, "abc123") + assert result is not None + assert result["id"] == 1 + + def test_returns_none_when_no_match(self) -> None: + client = MagicMock(spec=GiteaClient) + client.list_action_runs.return_value = { + "workflow_runs": [{"id": 1, "head_sha": "def456"}], + } + result = _find_latest_run_by_sha(client, "abc123") + assert result is None + + def test_returns_none_when_empty(self) -> None: + client = MagicMock(spec=GiteaClient) + client.list_action_runs.return_value = {"workflow_runs": []} + result = _find_latest_run_by_sha(client, "abc123") + assert result is None + + +class TestFindFailedJobs: + def test_returns_failed(self) -> None: + jobs = [ + {"id": 1, "name": "quality", "conclusion": "failure"}, + {"id": 2, "name": "lint", "conclusion": "success"}, + ] + result = _find_failed_jobs(jobs) + assert len(result) == 1 + assert result[0]["name"] == "quality" + + def test_empty_when_none_failed(self) -> None: + jobs = [{"id": 1, "name": "quality", "conclusion": "success"}] + assert _find_failed_jobs(jobs) == [] + + +class TestFindJobByName: + def test_case_insensitive_partial(self) -> None: + jobs = [{"id": 1, "name": "CI / quality (pull_request)"}] + result = _find_job_by_name(jobs, "QUALITY") + assert result is not None + assert result["id"] == 1 + + def test_returns_none_when_not_found(self) -> None: + jobs = [{"id": 1, "name": "quality"}] + assert _find_job_by_name(jobs, "molecule") is None + + +class TestPrintJobSummary: + def test_prints_all_jobs(self, capsys: pytest.CaptureFixture) -> None: + jobs = [ + {"id": 1, "name": "quality", "conclusion": "failure", "status": "completed"}, + {"id": 2, "name": "lint", "conclusion": "success", "status": "completed"}, + ] + _print_job_summary(jobs) + out = capsys.readouterr().out + assert "[FAIL]" in out + assert "[OK]" in out + assert "quality" in out + assert "lint" in out + + +class TestPrintFailedSteps: + def test_prints_failed_steps(self, capsys: pytest.CaptureFixture) -> None: + job = { + "steps": [ + {"name": "checkout", "number": 1, "conclusion": "success"}, + {"name": "Unit tests", "number": 3, "conclusion": "failure"}, + ] + } + result = _print_failed_steps(job) + assert result == [3] + out = capsys.readouterr().out + assert "FAILED step #3" in out + assert "Unit tests" in out + + def test_no_failed_steps(self, capsys: pytest.CaptureFixture) -> None: + job = {"steps": [{"name": "checkout", "number": 1, "conclusion": "success"}]} + result = _print_failed_steps(job) + assert result == [] + + def test_no_steps_key(self, capsys: pytest.CaptureFixture) -> None: + result = _print_failed_steps({}) + assert result == [] + + +class TestPrintLogs: + def test_prints_all_lines(self, capsys: pytest.CaptureFixture) -> None: + client = MagicMock(spec=GiteaClient) + client.get_action_job_logs.return_value = "line 1\nline 2\nline 3" + _print_logs(client, 100, tail=0) + out = capsys.readouterr().out + assert "line 1" in out + assert "line 3" in out + + def test_tail_truncates(self, capsys: pytest.CaptureFixture) -> None: + client = MagicMock(spec=GiteaClient) + client.get_action_job_logs.return_value = "\n".join(f"line {i}" for i in range(100)) + _print_logs(client, 100, tail=10) + out = capsys.readouterr().out + assert "line 99" in out + assert "line 0" not in out + assert "showing last 10" in out + + def test_api_error_handled(self, capsys: pytest.CaptureFixture) -> None: + client = MagicMock(spec=GiteaClient) + client.get_action_job_logs.side_effect = APIError(404, "not found") + _print_logs(client, 100, tail=0) + out = capsys.readouterr().out + assert "Could not fetch logs" in out + + +class TestCli: + def test_no_token_raises(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("CI_GITEA_TOKEN", raising=False) + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42"]) + assert result.exit_code != 0 + assert "CI_GITEA_TOKEN" in result.output + + @patch("devx.tools.pr_logs.REPO_OWNER", "") + def test_no_owner_raises(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42"]) + assert result.exit_code != 0 + assert "owner" in result.output.lower() + + @patch("devx.tools.pr_logs.GiteaClient") + @patch("devx.tools.pr_status.subprocess.run") + def test_auto_detect_pr( + self, mock_subprocess: MagicMock, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + mock_subprocess.return_value = MagicMock(returncode=0, stdout="feature-branch\n") + client = mock_client_cls.return_value + client.list_prs.return_value = [{"number": 42, "head": {"ref": "feature-branch"}}] + client.get_pr.return_value = {"head": {"sha": "abc123"}} + client.list_action_runs.return_value = {"workflow_runs": []} + runner = CliRunner() + result = runner.invoke(cli, []) + assert result.exit_code != 0 + assert "Fetching logs for PR #42" in result.output + + @patch("devx.tools.pr_logs.GiteaClient") + def test_no_runs_found(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + client = mock_client_cls.return_value + client.get_pr.return_value = {"head": {"sha": "abc123"}} + client.list_action_runs.return_value = {"workflow_runs": []} + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42"]) + assert result.exit_code != 0 + assert "No workflow runs" in result.output + + @patch("devx.tools.pr_logs.GiteaClient") + def test_no_jobs(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + client = mock_client_cls.return_value + client.get_pr.return_value = {"head": {"sha": "abc123"}} + client.list_action_runs.return_value = { + "workflow_runs": [{"id": 1, "status": "completed", "head_sha": "abc123"}], + } + client.get_action_run_jobs.return_value = [] + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42"]) + assert result.exit_code == 0 + assert "No jobs" in result.output + + @patch("devx.tools.pr_logs.GiteaClient") + def test_no_failed_jobs(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + client = mock_client_cls.return_value + client.get_pr.return_value = {"head": {"sha": "abc123"}} + client.list_action_runs.return_value = { + "workflow_runs": [{"id": 1, "status": "completed", "head_sha": "abc123"}], + } + client.get_action_run_jobs.return_value = [ + {"id": 100, "name": "quality", "conclusion": "success", "status": "completed", "steps": []} + ] + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42"]) + assert result.exit_code == 0 + assert "No failed jobs" in result.output + + @patch("devx.tools.pr_logs.GiteaClient") + def test_failed_job_logs(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + client = mock_client_cls.return_value + client.get_pr.return_value = {"head": {"sha": "abc123"}} + client.list_action_runs.return_value = { + "workflow_runs": [{"id": 1, "status": "completed", "head_sha": "abc123"}], + } + client.get_action_run_jobs.return_value = [ + { + "id": 100, + "name": "quality", + "conclusion": "failure", + "status": "completed", + "steps": [ + {"name": "checkout", "number": 1, "conclusion": "success"}, + {"name": "Unit tests", "number": 3, "conclusion": "failure"}, + ], + } + ] + client.get_action_job_logs.return_value = "error: test failed" + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42", "--tail", "0"]) + assert result.exit_code == 0 + assert "FAILED step #3" in result.output + assert "error: test failed" in result.output + + @patch("devx.tools.pr_logs.GiteaClient") + def test_specific_job(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + client = mock_client_cls.return_value + client.get_pr.return_value = {"head": {"sha": "abc123"}} + client.list_action_runs.return_value = { + "workflow_runs": [{"id": 1, "status": "completed", "head_sha": "abc123"}], + } + client.get_action_run_jobs.return_value = [ + {"id": 100, "name": "quality", "conclusion": "success", "status": "completed", "steps": []}, + {"id": 101, "name": "lint", "conclusion": "success", "status": "completed", "steps": []}, + ] + client.get_action_job_logs.return_value = "lint output here" + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42", "--job", "lint", "--tail", "0"]) + assert result.exit_code == 0 + assert "lint output here" in result.output + + @patch("devx.tools.pr_logs.GiteaClient") + def test_job_not_found(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + client = mock_client_cls.return_value + client.get_pr.return_value = {"head": {"sha": "abc123"}} + client.list_action_runs.return_value = { + "workflow_runs": [{"id": 1, "status": "completed", "head_sha": "abc123"}], + } + client.get_action_run_jobs.return_value = [ + {"id": 100, "name": "quality", "conclusion": "success", "status": "completed", "steps": []} + ] + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42", "--job", "nonexistent"]) + assert result.exit_code != 0 + assert "No job matching" in result.output + + @patch("devx.tools.pr_logs.GiteaClient") + def test_no_sha_raises(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + client = mock_client_cls.return_value + client.get_pr.return_value = {"head": {}} + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42"]) + assert result.exit_code != 0 + assert "SHA" in result.output diff --git a/tests/unit/test_pr_status.py b/tests/unit/test_pr_status.py new file mode 100644 index 0000000..6f2fe5b --- /dev/null +++ b/tests/unit/test_pr_status.py @@ -0,0 +1,279 @@ +"""Unit tests for devx.tools.pr_status.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest +from click.testing import CliRunner + +from devx.api_clients import GiteaClient +from devx.tools.pr_status import ( + _get_pr_sha, + _get_symbol, + cli, + print_status, + wait_for_completion, +) + + +class TestGetSymbol: + def test_success(self) -> None: + assert _get_symbol("success") == "[OK]" + + def test_failure(self) -> None: + assert _get_symbol("failure") == "[FAIL]" + + def test_pending(self) -> None: + assert _get_symbol("pending") == "[..]" + + def test_unknown(self) -> None: + assert _get_symbol("weird") == "[weird]" + + +class TestGetPrSha: + def test_returns_head_sha(self) -> None: + client = MagicMock(spec=GiteaClient) + client.get_pr.return_value = {"head": {"sha": "abc123"}} + assert _get_pr_sha(client, 42) == "abc123" + + def test_returns_empty_when_missing(self) -> None: + client = MagicMock(spec=GiteaClient) + client.get_pr.return_value = {"head": {}} + assert _get_pr_sha(client, 42) == "" + + +class TestPrintStatus: + def test_no_statuses(self, capsys: pytest.CaptureFixture) -> None: + client = MagicMock(spec=GiteaClient) + client.get_commit_status.return_value = [] + result = print_status(client, "abc123") + assert result == "none" + + def test_all_success(self, capsys: pytest.CaptureFixture) -> None: + client = MagicMock(spec=GiteaClient) + client.get_commit_status.return_value = [ + {"context": "CI / quality", "status": "success"}, + {"context": "CI / lint", "status": "success"}, + ] + result = print_status(client, "abc123") + assert result == "success" + + def test_has_failure(self, capsys: pytest.CaptureFixture) -> None: + client = MagicMock(spec=GiteaClient) + client.get_commit_status.return_value = [ + {"context": "CI / quality", "status": "success"}, + {"context": "CI / lint", "status": "failure"}, + ] + result = print_status(client, "abc123") + assert result == "failure" + + def test_pending(self, capsys: pytest.CaptureFixture) -> None: + client = MagicMock(spec=GiteaClient) + client.get_commit_status.return_value = [ + {"context": "CI / quality", "status": "pending"}, + ] + result = print_status(client, "abc123") + assert result == "pending" + + def test_skipped_still_success(self, capsys: pytest.CaptureFixture) -> None: + client = MagicMock(spec=GiteaClient) + client.get_commit_status.return_value = [ + {"context": "CI / quality", "status": "success"}, + {"context": "CI / molecule", "status": "skipped"}, + ] + result = print_status(client, "abc123") + assert result == "success" + + +class TestWaitForCompletion: + @patch("devx.tools.pr_status.time.sleep") + @patch("devx.tools.pr_status.time.time", side_effect=[0, 0, 100, 200]) + def test_success_after_pending( + self, mock_time: MagicMock, mock_sleep: MagicMock, capsys: pytest.CaptureFixture + ) -> None: + client = MagicMock(spec=GiteaClient) + client.get_commit_status.side_effect = [ + [{"context": "CI / quality", "status": "pending"}], + [{"context": "CI / quality", "status": "success"}], + ] + result = wait_for_completion(client, "abc", timeout=600, interval=1) + assert result == "success" + + @patch("devx.tools.pr_status.time.sleep") + @patch("devx.tools.pr_status.time.time", side_effect=[0, 0, 100, 200]) + def test_failure_after_pending(self, mock_time: MagicMock, mock_sleep: MagicMock) -> None: + client = MagicMock(spec=GiteaClient) + client.get_commit_status.side_effect = [ + [{"context": "CI / quality", "status": "pending"}], + [{"context": "CI / quality", "status": "failure"}], + ] + result = wait_for_completion(client, "abc", timeout=600, interval=1) + assert result == "failure" + + @patch("devx.tools.pr_status.time.sleep") + @patch("devx.tools.pr_status.time.time", side_effect=[0, 700]) + def test_timeout(self, mock_time: MagicMock, mock_sleep: MagicMock) -> None: + client = MagicMock(spec=GiteaClient) + client.get_commit_status.return_value = [{"context": "CI / quality", "status": "pending"}] + result = wait_for_completion(client, "abc", timeout=600, interval=1) + assert result == "pending" + + +class TestCli: + def test_no_token_raises(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("CI_GITEA_TOKEN", raising=False) + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42"]) + assert result.exit_code != 0 + assert "CI_GITEA_TOKEN" in result.output + + @patch("devx.tools.pr_status.REPO_OWNER", "") + @patch("devx.tools.pr_status.get_repo_name", side_effect=Exception("should not reach")) + def test_no_owner_raises(self, mock_repo_name: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42"]) + assert result.exit_code != 0 + assert "owner" in result.output.lower() + + @patch("devx.tools.pr_status.GiteaClient") + def test_check_pr_status(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + client = mock_client_cls.return_value + client.get_pr.return_value = {"head": {"sha": "abc123"}} + client.get_commit_status.return_value = [ + {"context": "CI / quality", "status": "success"}, + ] + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42"]) + assert result.exit_code == 0 + assert "[OK]" in result.output + + @patch("devx.tools.pr_status.GiteaClient") + def test_check_sha_directly(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + client = mock_client_cls.return_value + client.get_commit_status.return_value = [ + {"context": "CI / quality", "status": "success"}, + ] + runner = CliRunner() + result = runner.invoke(cli, ["--sha", "abc123"]) + assert result.exit_code == 0 + assert "[OK]" in result.output + + @patch("devx.tools.pr_status.GiteaClient") + def test_failure_raises_exception(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + client = mock_client_cls.return_value + client.get_pr.return_value = {"head": {"sha": "abc123"}} + client.get_commit_status.return_value = [ + {"context": "CI / quality", "status": "failure"}, + ] + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42"]) + assert result.exit_code != 0 + assert "failed" in result.output.lower() + + @patch("devx.tools.pr_status.GiteaClient") + @patch("devx.tools.pr_status.subprocess.run") + def test_auto_detect_branch( + self, mock_subprocess: MagicMock, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + mock_subprocess.return_value = MagicMock(returncode=0, stdout="feature-branch\n") + client = mock_client_cls.return_value + client.list_prs.return_value = [{"number": 42, "head": {"ref": "feature-branch"}}] + client.get_pr.return_value = {"head": {"sha": "abc123"}} + client.get_commit_status.return_value = [{"context": "CI / quality", "status": "success"}] + runner = CliRunner() + result = runner.invoke(cli, []) + assert result.exit_code == 0 + assert "PR #42" in result.output + + @patch("devx.tools.pr_status.GiteaClient") + @patch("devx.tools.pr_status.subprocess.run") + def test_auto_detect_no_pr_found( + self, mock_subprocess: MagicMock, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + mock_subprocess.return_value = MagicMock(returncode=0, stdout="feature-branch\n") + client = mock_client_cls.return_value + client.list_prs.return_value = [] + runner = CliRunner() + result = runner.invoke(cli, []) + assert result.exit_code != 0 + assert "No open PR" in result.output + + @patch("devx.tools.pr_status.GiteaClient") + @patch("devx.tools.pr_status.subprocess.run") + def test_auto_detect_branch_error( + self, mock_subprocess: MagicMock, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + mock_subprocess.return_value = MagicMock(returncode=1, stderr="git error\n") + runner = CliRunner() + result = runner.invoke(cli, []) + assert result.exit_code != 0 + assert "Could not detect" in result.output + + @patch("devx.tools.pr_status.GiteaClient") + def test_no_sha_raises(self, mock_client_cls: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + client = mock_client_cls.return_value + client.get_pr.return_value = {"head": {}} + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42"]) + assert result.exit_code != 0 + assert "SHA" in result.output + + @patch("devx.tools.pr_status.time.sleep") + @patch("devx.tools.pr_status.time.time", side_effect=[0, 0, 100, 200]) + @patch("devx.tools.pr_status.GiteaClient") + def test_wait_success( + self, mock_client_cls: MagicMock, mock_time: MagicMock, mock_sleep: MagicMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + client = mock_client_cls.return_value + client.get_pr.return_value = {"head": {"sha": "abc123"}} + client.get_commit_status.side_effect = [ + [{"context": "CI / quality", "status": "pending"}], + [{"context": "CI / quality", "status": "success"}], + ] + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42", "--wait", "--timeout", "600", "--interval", "1"]) + assert result.exit_code == 0 + assert "[OK]" in result.output + + @patch("devx.tools.pr_status.time.sleep") + @patch("devx.tools.pr_status.time.time", side_effect=[0, 700]) + @patch("devx.tools.pr_status.GiteaClient") + def test_wait_timeout( + self, mock_client_cls: MagicMock, mock_time: MagicMock, mock_sleep: MagicMock, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("CI_GITEA_TOKEN", "tok") + monkeypatch.setenv("DEVX_REPO_OWNER", "owner") + monkeypatch.setenv("DEVX_REPO_NAME", "repo") + client = mock_client_cls.return_value + client.get_pr.return_value = {"head": {"sha": "abc123"}} + client.get_commit_status.return_value = [{"context": "CI / quality", "status": "pending"}] + runner = CliRunner() + result = runner.invoke(cli, ["--pr", "42", "--wait", "--timeout", "600", "--interval", "1"]) + assert result.exit_code != 0 + assert "timeout" in result.output.lower() diff --git a/tests/unit/test_pre_push_check.py b/tests/unit/test_pre_push_check.py index 4ad3bb2..4d48deb 100644 --- a/tests/unit/test_pre_push_check.py +++ b/tests/unit/test_pre_push_check.py @@ -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: