Public Access
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 devx-ci-bot
|
9e604ea2c7 | ||
 emil
|
4bb50bed58 | ||
|
gitea-actions-bot
|
5bb9dce530 | ||
|
emil
|
73662a3bf0 | ||
|
gitea-actions-bot
|
a1e87b1905 | ||
|
devx-ci-bot
|
d8d0ad04a2 | ||
|
emil
|
e836c09088 | ||
|
gitea-actions-bot
|
507436b134 | ||
|
devx-ci-bot
|
32cec2c5ad | ||
|
emil
|
55c530eb00 | ||
|
gitea-actions-bot
|
3928de4507 |
@@ -160,6 +160,21 @@ jobs:
|
||||
token: ${{ secrets.CI_GITEA_TOKEN }}
|
||||
- name: Set up environment
|
||||
run: make setup-image
|
||||
- name: Post approval review
|
||||
env:
|
||||
CI_GITEA_TOKEN: ${{ secrets.REVIEW_GITEA_TOKEN }}
|
||||
PR_NUMBER: ${{ github.event.number }}
|
||||
REPOSITORY: ${{ github.repository }}
|
||||
PYTHONPATH: src
|
||||
run: |
|
||||
. .venv/bin/activate
|
||||
python3 -m devx.ci.pr_review \
|
||||
"$PR_NUMBER" \
|
||||
"$REPOSITORY" \
|
||||
--event APPROVE \
|
||||
--checklist-confirmed \
|
||||
--checklist-categories 1,2,3,4,5,6,7,8,9,10,11,12,13 \
|
||||
--body "Auto-approved: all CI checks passed (quality, pr-review, release-dry-run)."
|
||||
- name: Squash merge with task ID
|
||||
env:
|
||||
CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
|
||||
|
||||
@@ -2,6 +2,24 @@
|
||||
|
||||
All notable changes to this project will be documented in this file.
|
||||
|
||||
## [0.26.3] - 2026-06-28
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
- Pin all dependencies to exact versions for reproducibility
|
||||
|
||||
## [0.26.2] - 2026-06-28
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
- Block admin merge override and auto-approve with review token
|
||||
|
||||
## [0.26.1] - 2026-06-28
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
- Force pip upgrade in setup-image to install new dependencies
|
||||
|
||||
## [0.26.0] - 2026-06-28
|
||||
|
||||
### Features
|
||||
|
||||
@@ -35,7 +35,7 @@ setup-release: $(VENV)/bin/activate .env
|
||||
# an older devx.mak that doesn't yet define devx-setup-image. Consumer repos
|
||||
# (grm, infra) can safely alias to devx-setup-image since they install devx from PyPI.
|
||||
setup-image:
|
||||
@if [ -d /opt/venv ]; then ln -sf /opt/venv .venv; . .venv/bin/activate && pip install -e . --no-deps 2>/dev/null; \
|
||||
@if [ -d /opt/venv ]; then ln -sf /opt/venv .venv; . .venv/bin/activate && pip install --no-cache-dir -e . 2>/dev/null; \
|
||||
else echo "[setup-image] /opt/venv not found — falling back to setup-ci"; $(MAKE) setup-ci; fi
|
||||
|
||||
.env:
|
||||
|
||||
@@ -16,12 +16,12 @@ quality badges.
|
||||
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/src/branch/master/LICENSE)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
|
||||
[](https://www.python.org/downloads/)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
|
||||
[](https://www.python.org/downloads/)
|
||||
|
||||
## Why devx?
|
||||
|
||||
|
||||
+6
-6
@@ -12,12 +12,12 @@ project to be reusable across all oblachno-oss repositories.
|
||||
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/src/branch/master/LICENSE)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
|
||||
[](https://www.python.org/downloads/)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
|
||||
[](https://www.python.org/downloads/)
|
||||
|
||||
## Overview
|
||||
|
||||
|
||||
+30
-25
@@ -13,10 +13,13 @@ classifiers = [
|
||||
"Programming Language :: Python :: 3",
|
||||
"License :: OSI Approved :: GNU General Public License v3 (GPLv3)",
|
||||
]
|
||||
# All dependencies are pinned to exact versions for full reproducibility.
|
||||
# Update pinned versions in a dedicated PR with verification.
|
||||
dependencies = [
|
||||
"requests>=2.34.2",
|
||||
"python-dotenv>=1.2.2",
|
||||
"click>=8.4.1",
|
||||
"requests==2.34.2",
|
||||
"python-dotenv==1.2.2",
|
||||
"click==8.4.2",
|
||||
"tenacity==9.1.4", # retry logic for GiteaClient/VikunjaClient
|
||||
]
|
||||
|
||||
[project.scripts]
|
||||
@@ -28,44 +31,44 @@ version = {attr = "devx.__version__"}
|
||||
[project.optional-dependencies]
|
||||
# Test runners (pytest + coverage + parallel execution)
|
||||
ci = [
|
||||
"pytest>=9.1.0",
|
||||
"pytest-cov>=7.1.0",
|
||||
"pytest-xdist>=3.8",
|
||||
"pytest==9.1.1",
|
||||
"pytest-cov==7.1.0",
|
||||
"pytest-xdist==3.8.0",
|
||||
]
|
||||
# Lint and type-checking tools (quality job, badge generation)
|
||||
lint = [
|
||||
"ruff>=0.15.17",
|
||||
"pyright>=1.1.410",
|
||||
"bandit>=1.8.2",
|
||||
"pip-audit>=2.10",
|
||||
"pre-commit>=4.6.0",
|
||||
"ruff==0.15.20",
|
||||
"pyright==1.1.411",
|
||||
"bandit==1.9.4",
|
||||
"pip-audit==2.10.1",
|
||||
"pre-commit==4.6.0",
|
||||
]
|
||||
# Release tools (build + publish to PyPI/Gitea registry)
|
||||
release = [
|
||||
"build>=1.5.0",
|
||||
"twine>=6.2.0",
|
||||
"build==1.5.0",
|
||||
"twine==6.2.0",
|
||||
]
|
||||
# Molecule testing (for projects with Ansible roles)
|
||||
molecule = [
|
||||
"molecule>=26.4.0",
|
||||
"molecule-docker>=2.1.0",
|
||||
"ansible-lint>=26.4.0",
|
||||
"ansible-core>=2.15,<2.17",
|
||||
"molecule==26.4.0",
|
||||
"molecule-docker==2.1.0",
|
||||
"ansible-lint==26.4.0",
|
||||
"ansible-core==2.21.1",
|
||||
]
|
||||
# Deploy tools (for infra staging/production deployments)
|
||||
deploy = [
|
||||
"ansible-core>=2.15,<2.17",
|
||||
"boto3>=1.34",
|
||||
"docker>=7.0",
|
||||
"jinja2>=3.1",
|
||||
"pyyaml>=6.0",
|
||||
"cryptography>=41.0",
|
||||
"ansible-core==2.21.1",
|
||||
"boto3==1.43.36",
|
||||
"docker==7.1.0",
|
||||
"jinja2==3.1.6",
|
||||
"pyyaml==6.0.3",
|
||||
"cryptography==49.0.0",
|
||||
]
|
||||
# Full dev environment (local development)
|
||||
dev = [
|
||||
"devx[ci,lint,release,molecule]",
|
||||
"build>=1.3.0",
|
||||
"twine>=6.2.0",
|
||||
"build==1.5.0",
|
||||
"twine==6.2.0",
|
||||
]
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
@@ -97,6 +100,8 @@ indent-style = "space"
|
||||
[tool.pyright]
|
||||
include = ["src"]
|
||||
pythonVersion = "3.12"
|
||||
venvPath = "."
|
||||
venv = ".venv"
|
||||
strict = ["src/devx/config.py", "src/devx/exceptions.py", "src/devx/i18n.py", "src/devx/api_clients.py", "src/devx/gitea_cli.py"]
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -1,3 +1,3 @@
|
||||
"""devx — reusable development and CI/CD tools for oblachno-oss projects."""
|
||||
|
||||
__version__ = "0.26.0"
|
||||
__version__ = "0.26.3"
|
||||
|
||||
+72
-93
@@ -4,10 +4,16 @@ from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
import requests
|
||||
from tenacity import (
|
||||
before_sleep_log,
|
||||
retry,
|
||||
retry_if_exception_type,
|
||||
stop_after_attempt,
|
||||
wait_exponential,
|
||||
)
|
||||
|
||||
from devx.config import DEFAULT_TIMEOUT, MAX_RETRIES, RETRY_BACKOFF_BASE, RETRY_STATUS_CODES
|
||||
from devx.exceptions import APIError
|
||||
@@ -27,14 +33,69 @@ def _parse_error(e: requests.HTTPError) -> tuple[int, str]:
|
||||
return status, message
|
||||
|
||||
|
||||
def _is_retryable(e: Exception) -> bool:
|
||||
"""Check if an exception is a transient error worth retrying."""
|
||||
if isinstance(e, requests.ConnectionError):
|
||||
return True
|
||||
if isinstance(e, requests.HTTPError):
|
||||
status, _ = _parse_error(e)
|
||||
return status in RETRY_STATUS_CODES
|
||||
return isinstance(e, requests.Timeout)
|
||||
class _TransientHTTPError(requests.HTTPError):
|
||||
"""HTTP error with a retryable status code (wrapped for tenacity)."""
|
||||
|
||||
|
||||
class _RetryableRequestError(Exception):
|
||||
"""Connection/timeout error wrapped for tenacity retry."""
|
||||
|
||||
|
||||
def _execute_request(
|
||||
session: requests.Session,
|
||||
method: str,
|
||||
url: str,
|
||||
**kwargs: Any,
|
||||
) -> requests.Response:
|
||||
"""Execute a single HTTP request, wrapping transient errors for tenacity.
|
||||
|
||||
Non-retryable HTTP errors (4xx except 429) raise :class:`APIError` directly.
|
||||
Retryable errors (429, 5xx, connection, timeout) raise exceptions that
|
||||
tenacity will retry.
|
||||
"""
|
||||
try:
|
||||
response = session.request(method, url, timeout=DEFAULT_TIMEOUT, **kwargs)
|
||||
response.raise_for_status()
|
||||
return response
|
||||
except requests.HTTPError as e:
|
||||
status, message = _parse_error(e)
|
||||
if status in RETRY_STATUS_CODES:
|
||||
# Wrap in _TransientHTTPError so tenacity retries it
|
||||
raise _TransientHTTPError(message, response=e.response) from e
|
||||
raise APIError(status, message) from e
|
||||
except (requests.ConnectionError, requests.Timeout) as e:
|
||||
raise _RetryableRequestError(str(e)) from e
|
||||
|
||||
|
||||
# Tenacity retry decorator shared by both clients.
|
||||
# Retries on transient HTTP errors (429, 5xx) and connection/timeout errors.
|
||||
_retry_decorator = retry(
|
||||
stop=stop_after_attempt(MAX_RETRIES),
|
||||
wait=wait_exponential(multiplier=RETRY_BACKOFF_BASE, min=RETRY_BACKOFF_BASE, max=RETRY_BACKOFF_BASE**MAX_RETRIES),
|
||||
retry=retry_if_exception_type((_TransientHTTPError, _RetryableRequestError)),
|
||||
before_sleep=before_sleep_log(logger, logging.WARNING),
|
||||
reraise=True,
|
||||
)
|
||||
|
||||
|
||||
def _request_with_retry(
|
||||
session: requests.Session,
|
||||
url: str,
|
||||
method: str,
|
||||
**kwargs: Any,
|
||||
) -> requests.Response:
|
||||
"""Execute an HTTP request with tenacity-managed retry logic.
|
||||
|
||||
On exhaustion, the last exception is translated to :class:`APIError`.
|
||||
"""
|
||||
try:
|
||||
return _retry_decorator(_execute_request)(session, method, url, **kwargs)
|
||||
except _TransientHTTPError as e:
|
||||
response = getattr(e, "response", None)
|
||||
status = response.status_code if response is not None else 0
|
||||
raise APIError(status, str(e)) from e
|
||||
except _RetryableRequestError as e:
|
||||
raise APIError(0, str(e)) from e
|
||||
|
||||
|
||||
class GiteaClient:
|
||||
@@ -56,49 +117,7 @@ class GiteaClient:
|
||||
return f"{self._base_url}/repos/{self._owner}/{self._repo}{path}"
|
||||
|
||||
def _request(self, method: str, path: str, **kwargs: Any) -> requests.Response:
|
||||
url = self._url(path)
|
||||
last_exc: Exception | None = None
|
||||
for attempt in range(MAX_RETRIES):
|
||||
try:
|
||||
response = self._session.request(method, url, timeout=DEFAULT_TIMEOUT, **kwargs)
|
||||
response.raise_for_status()
|
||||
return response
|
||||
except requests.HTTPError as e:
|
||||
status, message = _parse_error(e)
|
||||
if _is_retryable(e) and attempt < MAX_RETRIES - 1:
|
||||
wait = RETRY_BACKOFF_BASE ** (attempt + 1)
|
||||
logger.warning(
|
||||
"Transient HTTP %d on %s %s, retrying in %ds (attempt %d/%d)",
|
||||
status,
|
||||
method,
|
||||
path,
|
||||
wait,
|
||||
attempt + 1,
|
||||
MAX_RETRIES,
|
||||
)
|
||||
time.sleep(wait)
|
||||
last_exc = e
|
||||
continue
|
||||
raise APIError(status, message) from e
|
||||
except (requests.ConnectionError, requests.Timeout) as e:
|
||||
if attempt < MAX_RETRIES - 1:
|
||||
wait = RETRY_BACKOFF_BASE ** (attempt + 1)
|
||||
logger.warning(
|
||||
"Connection error on %s %s, retrying in %ds (attempt %d/%d)",
|
||||
method,
|
||||
path,
|
||||
wait,
|
||||
attempt + 1,
|
||||
MAX_RETRIES,
|
||||
)
|
||||
time.sleep(wait)
|
||||
last_exc = e
|
||||
continue
|
||||
raise APIError(0, str(e)) from e
|
||||
# Should not reach here, but just in case
|
||||
if last_exc: # pragma: no cover
|
||||
raise APIError(0, str(last_exc)) from last_exc
|
||||
raise APIError(0, "Max retries exceeded") # pragma: no cover
|
||||
return _request_with_retry(self._session, self._url(path), method, **kwargs)
|
||||
|
||||
# -- repo settings --
|
||||
|
||||
@@ -351,47 +370,7 @@ class VikunjaClient:
|
||||
|
||||
def _request(self, method: str, path: str, **kwargs: Any) -> requests.Response:
|
||||
url = f"{self._base_url}{path}"
|
||||
last_exc: Exception | None = None
|
||||
for attempt in range(MAX_RETRIES):
|
||||
try:
|
||||
response = self._session.request(method, url, timeout=DEFAULT_TIMEOUT, **kwargs)
|
||||
response.raise_for_status()
|
||||
return response
|
||||
except requests.HTTPError as e:
|
||||
status, message = _parse_error(e)
|
||||
if _is_retryable(e) and attempt < MAX_RETRIES - 1:
|
||||
wait = RETRY_BACKOFF_BASE ** (attempt + 1)
|
||||
logger.warning(
|
||||
"Transient HTTP %d on %s %s, retrying in %ds (attempt %d/%d)",
|
||||
status,
|
||||
method,
|
||||
path,
|
||||
wait,
|
||||
attempt + 1,
|
||||
MAX_RETRIES,
|
||||
)
|
||||
time.sleep(wait)
|
||||
last_exc = e
|
||||
continue
|
||||
raise APIError(status, message) from e
|
||||
except (requests.ConnectionError, requests.Timeout) as e:
|
||||
if attempt < MAX_RETRIES - 1:
|
||||
wait = RETRY_BACKOFF_BASE ** (attempt + 1)
|
||||
logger.warning(
|
||||
"Connection error on %s %s, retrying in %ds (attempt %d/%d)",
|
||||
method,
|
||||
path,
|
||||
wait,
|
||||
attempt + 1,
|
||||
MAX_RETRIES,
|
||||
)
|
||||
time.sleep(wait)
|
||||
last_exc = e
|
||||
continue
|
||||
raise APIError(0, str(e)) from e
|
||||
if last_exc: # pragma: no cover
|
||||
raise APIError(0, str(last_exc)) from last_exc
|
||||
raise APIError(0, "Max retries exceeded") # pragma: no cover
|
||||
return _request_with_retry(self._session, url, method, **kwargs)
|
||||
|
||||
def list_tasks(self, **params: Any) -> list[dict[str, Any]]:
|
||||
r = self._request("GET", "/tasks", params=params)
|
||||
|
||||
@@ -2,8 +2,14 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess # nosec B404
|
||||
|
||||
import click
|
||||
|
||||
from devx.config import TASK_ID_RE
|
||||
from devx.i18n import _
|
||||
|
||||
|
||||
def get_latest_tag() -> str:
|
||||
"""Get the latest git tag, or empty string if none exists."""
|
||||
@@ -16,3 +22,97 @@ def get_latest_tag() -> str:
|
||||
if result.returncode != 0:
|
||||
return ""
|
||||
return result.stdout.strip()
|
||||
|
||||
|
||||
def run_cmd(
|
||||
args: list[str],
|
||||
check: bool = True,
|
||||
capture: bool = True,
|
||||
) -> subprocess.CompletedProcess[str]:
|
||||
"""Run a command and return the completed process.
|
||||
|
||||
Args:
|
||||
args: Command and arguments as a list.
|
||||
check: If True, raise :class:`click.ClickException` on non-zero exit.
|
||||
capture: If True, capture stdout/stderr. If False, inherit parent's.
|
||||
"""
|
||||
result = subprocess.run( # nosec B603
|
||||
args,
|
||||
capture_output=capture,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
if check and result.returncode != 0:
|
||||
raise click.ClickException(
|
||||
_(
|
||||
"Command failed ({cmd}): {stderr}",
|
||||
cmd=" ".join(args),
|
||||
stderr=result.stderr.strip() if result.stderr else result.stdout.strip(),
|
||||
)
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def extract_task_id(text: str) -> str:
|
||||
"""Extract the ``{PREFIX}-N`` task identifier from *text*.
|
||||
|
||||
Returns the matched string (e.g. ``DEVX-42``) or an empty string if
|
||||
no task ID is found.
|
||||
"""
|
||||
match = TASK_ID_RE.search(text)
|
||||
return match.group(0) if match else ""
|
||||
|
||||
|
||||
def write_github_env(key: str, value: str) -> None:
|
||||
"""Append a key=value line to the ``$GITHUB_ENV`` file.
|
||||
|
||||
Multi-line values use the heredoc syntax required by Gitea Actions.
|
||||
Raises :class:`click.ClickException` if ``GITHUB_ENV`` is not set.
|
||||
"""
|
||||
gh_env = os.environ.get("GITHUB_ENV")
|
||||
if not gh_env:
|
||||
raise click.ClickException("GITHUB_ENV environment variable is not set")
|
||||
with open(gh_env, "a", encoding="utf-8") as f: # noqa: PTH123
|
||||
if "\n" in value:
|
||||
delimiter = "EOF"
|
||||
f.write(f"{key}<<{delimiter}\n{value}\n{delimiter}\n")
|
||||
else:
|
||||
f.write(f"{key}={value}\n")
|
||||
|
||||
|
||||
def write_github_output(key: str, value: str) -> None:
|
||||
"""Append a key=value line to the ``$GITHUB_OUTPUT`` file.
|
||||
|
||||
Raises :class:`click.ClickException` if ``GITHUB_OUTPUT`` is not set.
|
||||
"""
|
||||
gh_output = os.environ.get("GITHUB_OUTPUT")
|
||||
if not gh_output:
|
||||
raise click.ClickException("GITHUB_OUTPUT environment variable is not set")
|
||||
with open(gh_output, "a", encoding="utf-8") as f: # noqa: PTH123
|
||||
f.write(f"{key}={value}\n")
|
||||
|
||||
|
||||
def lpt_distribute[T](items: list[T], weights: list[int], max_runners: int) -> list[list[T]]:
|
||||
"""Distribute *items* across *max_runners* using LPT scheduling.
|
||||
|
||||
Sorts items by weight (descending), then assigns each to the runner
|
||||
with the least total weight. This produces a more balanced distribution
|
||||
than naive round-robin when items have varying costs.
|
||||
|
||||
Args:
|
||||
items: Items to distribute.
|
||||
weights: Parallel list of integer weights (higher = heavier).
|
||||
max_runners: Number of runner groups to create.
|
||||
|
||||
Returns:
|
||||
A list of ``max_runners`` lists, each containing the items assigned
|
||||
to that runner.
|
||||
"""
|
||||
groups: list[list[T]] = [[] for _ in range(max_runners)]
|
||||
loads = [0] * max_runners
|
||||
indexed = sorted(enumerate(items), key=lambda x: (-weights[x[0]], x[0]))
|
||||
for orig_idx, item in indexed:
|
||||
min_runner = min(range(max_runners), key=lambda r: loads[r])
|
||||
groups[min_runner].append(item)
|
||||
loads[min_runner] += weights[orig_idx]
|
||||
return groups
|
||||
|
||||
@@ -22,7 +22,6 @@ Usage:
|
||||
|
||||
import os
|
||||
import re
|
||||
import subprocess # nosec B404
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
@@ -30,11 +29,11 @@ import click
|
||||
from dotenv import load_dotenv # pyright: ignore[reportMissingImports,reportUnknownVariableType]
|
||||
|
||||
from devx.api_clients import GiteaClient, VikunjaClient
|
||||
from devx.ci._shared import extract_task_id as _extract_task_id
|
||||
from devx.config import (
|
||||
CONVENTIONAL_RE,
|
||||
DEFAULT_PER_PAGE,
|
||||
GITEA_API_URL,
|
||||
TASK_ID_RE,
|
||||
TASK_PREFIX,
|
||||
VIKUNJA_API_URL,
|
||||
VIKUNJA_PROJECT_ID,
|
||||
@@ -48,20 +47,6 @@ PR_TITLE_RE = re.compile(rf"^{TASK_PREFIX}-\d+:\s+.+")
|
||||
load_dotenv()
|
||||
|
||||
|
||||
def run_cmd(args: list[str], check: bool = True) -> subprocess.CompletedProcess[str]:
|
||||
"""Run a command and return the completed process."""
|
||||
result = subprocess.run(args, capture_output=True, text=True, check=False) # nosec B603
|
||||
if check and result.returncode != 0:
|
||||
raise click.ClickException(
|
||||
_(
|
||||
"Command failed ({cmd}): {stderr}",
|
||||
cmd=" ".join(args),
|
||||
stderr=result.stderr.strip() or result.stdout.strip(),
|
||||
)
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def read_taskid(branch: str) -> str:
|
||||
"""Read task ID from branch name.
|
||||
|
||||
@@ -92,9 +77,8 @@ def read_taskid(branch: str) -> str:
|
||||
|
||||
|
||||
def extract_task_id(branch: str) -> str:
|
||||
"""Extract DEVX-N task identifier from branch name (legacy fallback)."""
|
||||
match = TASK_ID_RE.search(branch)
|
||||
return match.group(0) if match else ""
|
||||
"""Extract task identifier from branch name (delegates to shared utility)."""
|
||||
return _extract_task_id(branch)
|
||||
|
||||
|
||||
def validate_pr_title(pr_title: str, task_id: str) -> None:
|
||||
|
||||
@@ -53,6 +53,7 @@ from devx.config import (
|
||||
VIKUNJA_API_URL,
|
||||
VIKUNJA_PROJECT_ID,
|
||||
)
|
||||
from devx.exceptions import APIError
|
||||
from devx.i18n import _
|
||||
|
||||
load_dotenv()
|
||||
@@ -108,7 +109,7 @@ def get_pr_title_from_gitea(repo: str, pr_number: int) -> str | None:
|
||||
try:
|
||||
pr = client.get_pr(pr_number)
|
||||
return str(pr.get("title", ""))
|
||||
except Exception:
|
||||
except APIError:
|
||||
return None
|
||||
|
||||
|
||||
|
||||
@@ -138,7 +138,7 @@ from typing import Any
|
||||
|
||||
import click
|
||||
|
||||
from devx.ci._shared import get_latest_tag
|
||||
from devx.ci._shared import get_latest_tag, write_github_output
|
||||
from devx.i18n import _
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -584,15 +584,8 @@ def has_user_facing_changes(
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _write_github_output(key: str, value: str) -> None:
|
||||
"""Append a key=value line to the $GITHUB_OUTPUT file."""
|
||||
gh_output = os.environ.get("GITHUB_OUTPUT")
|
||||
if not gh_output:
|
||||
raise click.ClickException("GITHUB_OUTPUT environment variable is not set")
|
||||
with open(gh_output, "a") as f: # noqa: PTH123
|
||||
f.write(f"{key}={value}\n")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Classification logic
|
||||
# ---------------------------------------------------------------------------
|
||||
# CLI
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -633,9 +626,9 @@ def main(base: str | None, head: str, quiet: bool, check: str, github_output: bo
|
||||
force = True
|
||||
|
||||
if force and github_output:
|
||||
_write_github_output("user-facing-changed", "true")
|
||||
write_github_output("user-facing-changed", "true")
|
||||
for tag in available_tags:
|
||||
_write_github_output(f"{tag}-changed", "true")
|
||||
write_github_output(f"{tag}-changed", "true")
|
||||
click.echo("Forced user-facing-changed=true via --force flag.")
|
||||
return
|
||||
|
||||
@@ -643,9 +636,9 @@ def main(base: str | None, head: str, quiet: bool, check: str, github_output: bo
|
||||
base = get_latest_tag()
|
||||
if not base:
|
||||
if github_output:
|
||||
_write_github_output("user-facing-changed", "true")
|
||||
write_github_output("user-facing-changed", "true")
|
||||
for tag in available_tags:
|
||||
_write_github_output(f"{tag}-changed", "true")
|
||||
write_github_output(f"{tag}-changed", "true")
|
||||
click.echo("No tags found — treating all changes as user-facing.")
|
||||
return
|
||||
if quiet:
|
||||
@@ -657,9 +650,9 @@ def main(base: str | None, head: str, quiet: bool, check: str, github_output: bo
|
||||
files = get_changed_files(base, head)
|
||||
if not files:
|
||||
if github_output:
|
||||
_write_github_output("user-facing-changed", "false")
|
||||
write_github_output("user-facing-changed", "false")
|
||||
for tag in available_tags:
|
||||
_write_github_output(f"{tag}-changed", "false")
|
||||
write_github_output(f"{tag}-changed", "false")
|
||||
click.echo(f"No changes between {base} and {head}.")
|
||||
return
|
||||
if quiet:
|
||||
@@ -671,9 +664,9 @@ def main(base: str | None, head: str, quiet: bool, check: str, github_output: bo
|
||||
result = classifier.classify(files)
|
||||
|
||||
if github_output:
|
||||
_write_github_output("user-facing-changed", "true" if result.has_user_facing else "false")
|
||||
write_github_output("user-facing-changed", "true" if result.has_user_facing else "false")
|
||||
for tag in available_tags:
|
||||
_write_github_output(f"{tag}-changed", "true" if result.has_tag(tag) else "false")
|
||||
write_github_output(f"{tag}-changed", "true" if result.has_tag(tag) else "false")
|
||||
click.echo(f"User-facing files changed: {result.has_user_facing}")
|
||||
for tag in available_tags:
|
||||
click.echo(f"{tag.capitalize()} files changed: {result.has_tag(tag)}")
|
||||
|
||||
@@ -12,12 +12,13 @@ Usage::
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
import subprocess # nosec B404
|
||||
|
||||
import click
|
||||
|
||||
from devx.ci._shared import write_github_output
|
||||
|
||||
RELEASE_RE = re.compile(r"^release: v\d+\.\d+\.\d+")
|
||||
|
||||
|
||||
@@ -39,15 +40,6 @@ def is_release_commit(message: str) -> bool:
|
||||
return bool(RELEASE_RE.match(message))
|
||||
|
||||
|
||||
def write_github_output(key: str, value: str) -> None:
|
||||
"""Append a key=value line to the $GITHUB_OUTPUT file."""
|
||||
gh_output = os.environ.get("GITHUB_OUTPUT")
|
||||
if not gh_output:
|
||||
raise click.ClickException("GITHUB_OUTPUT environment variable is not set")
|
||||
with open(gh_output, "a") as f: # noqa: PTH123
|
||||
f.write(f"{key}={value}\n")
|
||||
|
||||
|
||||
@click.command()
|
||||
def main() -> None:
|
||||
"""Detect if the latest commit is a release commit and set GITHUB_OUTPUT."""
|
||||
|
||||
@@ -162,7 +162,7 @@ def main(
|
||||
gh_output = os.environ.get("GITHUB_OUTPUT")
|
||||
if not gh_output:
|
||||
raise click.ClickException("GITHUB_OUTPUT environment variable is not set")
|
||||
with open(gh_output, "a") as f: # noqa: PTH123
|
||||
with open(gh_output, "a", encoding="utf-8") as f: # noqa: PTH123
|
||||
f.write(f"runner-count={count}\n")
|
||||
f.write(f"runner-indices={json.dumps(indices)}\n")
|
||||
click.echo(f"Runner count: {count}")
|
||||
|
||||
@@ -26,6 +26,7 @@ import os
|
||||
|
||||
import click
|
||||
|
||||
from devx.ci._shared import lpt_distribute, write_github_env
|
||||
from devx.i18n import _
|
||||
|
||||
DEFAULT_MAX_RUNNERS = 3
|
||||
@@ -54,15 +55,7 @@ def distribute(files: list[str], max_runners: int) -> list[list[str]]:
|
||||
the runner with the least total weight.
|
||||
"""
|
||||
weights = [_file_weight(f) for f in files]
|
||||
groups: list[list[str]] = [[] for _ in range(max_runners)]
|
||||
loads = [0] * max_runners
|
||||
# Sort by weight descending, preserving original order for ties
|
||||
indexed = sorted(enumerate(files), key=lambda x: (-weights[x[0]], x[0]))
|
||||
for orig_idx, f in indexed:
|
||||
min_runner = min(range(max_runners), key=lambda r: loads[r])
|
||||
groups[min_runner].append(f)
|
||||
loads[min_runner] += weights[orig_idx]
|
||||
return groups
|
||||
return lpt_distribute(files, weights, max_runners)
|
||||
|
||||
|
||||
def files_for_runner(files: list[str], runner_index: int, max_runners: int) -> list[str]:
|
||||
@@ -75,19 +68,6 @@ def files_for_runner(files: list[str], runner_index: int, max_runners: int) -> l
|
||||
return groups[runner_index]
|
||||
|
||||
|
||||
def _write_github_env(key: str, value: str) -> None:
|
||||
gh_env = os.environ.get("GITHUB_ENV")
|
||||
if not gh_env:
|
||||
raise click.ClickException("GITHUB_ENV environment variable is not set")
|
||||
with open(gh_env, "a") as f: # noqa: PTH123
|
||||
if "\n" in value:
|
||||
# Multi-line values require the heredoc syntax in $GITHUB_ENV.
|
||||
delimiter = "EOF"
|
||||
f.write(f"{key}<<{delimiter}\n{value}\n{delimiter}\n")
|
||||
else:
|
||||
f.write(f"{key}={value}\n")
|
||||
|
||||
|
||||
@click.command()
|
||||
@click.option("--pattern", required=True, help="Glob pattern for files to distribute.")
|
||||
@click.option(
|
||||
@@ -127,8 +107,8 @@ def main(pattern: str, runner_index: int | None, max_runners: int, github_env: b
|
||||
|
||||
if skip_if_excess and github_env and runner_index > max_runners:
|
||||
click.echo(f"Skipping — runner index {runner_index} > max runners {max_runners}")
|
||||
_write_github_env("ASSIGNED_FILES", "")
|
||||
_write_github_env("SKIP", "true")
|
||||
write_github_env("ASSIGNED_FILES", "")
|
||||
write_github_env("SKIP", "true")
|
||||
return
|
||||
|
||||
if runner_index < 1:
|
||||
@@ -139,8 +119,8 @@ def main(pattern: str, runner_index: int | None, max_runners: int, github_env: b
|
||||
encoded = "\n".join(assigned)
|
||||
|
||||
if github_env:
|
||||
_write_github_env("ASSIGNED_FILES", encoded)
|
||||
_write_github_env("SKIP", "false")
|
||||
write_github_env("ASSIGNED_FILES", encoded)
|
||||
write_github_env("SKIP", "false")
|
||||
click.echo(f"Assigned {len(assigned)} files to runner {runner_index}")
|
||||
return
|
||||
|
||||
|
||||
@@ -29,11 +29,11 @@ Usage::
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
import click
|
||||
|
||||
from devx.ci._shared import lpt_distribute, write_github_env
|
||||
from devx.i18n import _
|
||||
|
||||
DEFAULT_MAX_RUNNERS = 3
|
||||
@@ -93,14 +93,7 @@ def distribute(items: list[str], weights: list[int], max_runners: int) -> list[l
|
||||
Items are sorted by weight (descending), then assigned to the runner
|
||||
with the least total weight.
|
||||
"""
|
||||
groups: list[list[str]] = [[] for _ in range(max_runners)]
|
||||
loads = [0] * max_runners
|
||||
indexed = sorted(enumerate(items), key=lambda x: (-weights[x[0]], x[0]))
|
||||
for orig_idx, item in indexed:
|
||||
min_runner = min(range(max_runners), key=lambda r: loads[r])
|
||||
groups[min_runner].append(item)
|
||||
loads[min_runner] += weights[orig_idx]
|
||||
return groups
|
||||
return lpt_distribute(items, weights, max_runners)
|
||||
|
||||
|
||||
def items_for_runner(items: list[str], weights: list[int], runner_index: int, max_runners: int) -> list[str]:
|
||||
@@ -113,18 +106,6 @@ def items_for_runner(items: list[str], weights: list[int], runner_index: int, ma
|
||||
return groups[runner_index]
|
||||
|
||||
|
||||
def _write_github_env(key: str, value: str) -> None:
|
||||
gh_env = os.environ.get("GITHUB_ENV")
|
||||
if not gh_env:
|
||||
raise click.ClickException("GITHUB_ENV environment variable is not set")
|
||||
with open(gh_env, "a") as f: # noqa: PTH123
|
||||
if "\n" in value:
|
||||
delimiter = "EOF"
|
||||
f.write(f"{key}<<{delimiter}\n{value}\n{delimiter}\n")
|
||||
else:
|
||||
f.write(f"{key}={value}\n")
|
||||
|
||||
|
||||
@click.command()
|
||||
@click.option(
|
||||
"--items-file",
|
||||
@@ -166,7 +147,7 @@ def main(
|
||||
) -> None:
|
||||
# Read items from file or stdin
|
||||
if items_file is not None:
|
||||
with open(items_file) as f: # noqa: PTH123
|
||||
with open(items_file, encoding="utf-8") as f: # noqa: PTH123
|
||||
raw = f.read()
|
||||
else:
|
||||
raw = sys.stdin.read()
|
||||
@@ -186,8 +167,8 @@ def main(
|
||||
|
||||
if skip_if_excess and github_env and runner_index > max_runners:
|
||||
click.echo(f"Skipping — runner index {runner_index} > max runners {max_runners}")
|
||||
_write_github_env("ASSIGNED_ITEMS", "")
|
||||
_write_github_env("SKIP", "true")
|
||||
write_github_env("ASSIGNED_ITEMS", "")
|
||||
write_github_env("SKIP", "true")
|
||||
return
|
||||
|
||||
if runner_index < 1:
|
||||
@@ -198,8 +179,8 @@ def main(
|
||||
encoded = " ".join(assigned)
|
||||
|
||||
if github_env:
|
||||
_write_github_env("ASSIGNED_ITEMS", encoded)
|
||||
_write_github_env("SKIP", "false")
|
||||
write_github_env("ASSIGNED_ITEMS", encoded)
|
||||
write_github_env("SKIP", "false")
|
||||
click.echo(f"Assigned {len(assigned)} items to runner {runner_index}: {encoded}")
|
||||
return
|
||||
|
||||
|
||||
@@ -13,7 +13,8 @@ import click
|
||||
from dotenv import load_dotenv # pyright: ignore[reportMissingImports,reportUnknownVariableType]
|
||||
|
||||
from devx.api_clients import VikunjaClient
|
||||
from devx.config import DEFAULT_PER_PAGE, TASK_ID_RE, TASK_PREFIX, VIKUNJA_API_URL, VIKUNJA_PROJECT_ID
|
||||
from devx.ci._shared import extract_task_id as _extract_task_id
|
||||
from devx.config import DEFAULT_PER_PAGE, TASK_PREFIX, VIKUNJA_API_URL, VIKUNJA_PROJECT_ID
|
||||
from devx.exceptions import APIError
|
||||
from devx.i18n import _
|
||||
|
||||
@@ -47,10 +48,9 @@ def _get_git_commit_sha() -> str:
|
||||
|
||||
|
||||
def extract_task_id(commit_msg: str) -> str:
|
||||
"""Extract DEVX-N task identifier from the first line of commit message."""
|
||||
"""Extract task identifier from the first line of commit message (delegates to shared utility)."""
|
||||
first_line = commit_msg.split("\n")[0]
|
||||
match = TASK_ID_RE.search(first_line)
|
||||
return match.group(0) if match else ""
|
||||
return _extract_task_id(first_line)
|
||||
|
||||
|
||||
def extract_conventional_msg(commit_msg: str) -> str:
|
||||
@@ -61,7 +61,7 @@ def extract_conventional_msg(commit_msg: str) -> str:
|
||||
- ``DEVX-N <message>`` (current, space-separated)
|
||||
"""
|
||||
first_line = commit_msg.split("\n")[0]
|
||||
return re.sub(r"^DEVX-\d+[:\s]\s*", "", first_line)
|
||||
return re.sub(rf"^{TASK_PREFIX}-\d+[:\s]\s*", "", first_line)
|
||||
|
||||
|
||||
def resolve_task_id(client: VikunjaClient, task_id: str) -> int:
|
||||
|
||||
+14
-36
@@ -37,13 +37,12 @@ from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
import subprocess # nosec B404
|
||||
import sys
|
||||
|
||||
import click
|
||||
from dotenv import load_dotenv # pyright: ignore[reportMissingImports,reportUnknownVariableType]
|
||||
|
||||
from devx.ci._shared import get_latest_tag
|
||||
from devx.ci._shared import get_latest_tag, run_cmd, write_github_output
|
||||
from devx.ci.classify_changes import has_user_facing_changes # cross-CI import, needs PYTHONPATH=.
|
||||
from devx.i18n import _
|
||||
|
||||
@@ -54,25 +53,6 @@ CHANGELOG_FILE = "CHANGELOG.md"
|
||||
CLIFF_CONFIG = "cliff.toml"
|
||||
|
||||
|
||||
def run_cmd(args: list[str], check: bool = True, capture: bool = True) -> subprocess.CompletedProcess[str]:
|
||||
"""Run a command and return the completed process."""
|
||||
result = subprocess.run( # nosec B603
|
||||
args,
|
||||
capture_output=capture,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
if check and result.returncode != 0:
|
||||
raise click.ClickException(
|
||||
_(
|
||||
"Command failed ({cmd}): {stderr}",
|
||||
cmd=" ".join(args),
|
||||
stderr=result.stderr.strip() if result.stderr else result.stdout.strip(),
|
||||
)
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def tag_exists(tag: str) -> bool:
|
||||
"""Check if a git tag already exists."""
|
||||
result = run_cmd(["git", "tag", "-l", tag], check=False)
|
||||
@@ -217,7 +197,7 @@ def has_unreleased_changes(bumped_version: str | None = None) -> bool:
|
||||
|
||||
def update_init_version(new_version: str) -> None:
|
||||
"""Update __version__ in __init__.py."""
|
||||
with open(INIT_FILE) as f:
|
||||
with open(INIT_FILE, encoding="utf-8") as f:
|
||||
content = f.read()
|
||||
if not re.search(r'^__version__\s*=\s*"[^"]*"', content, flags=re.MULTILINE):
|
||||
raise click.ClickException(_("Could not find __version__ in {file}", file=INIT_FILE))
|
||||
@@ -228,7 +208,7 @@ def update_init_version(new_version: str) -> None:
|
||||
count=1,
|
||||
flags=re.MULTILINE,
|
||||
)
|
||||
with open(INIT_FILE, "w") as f:
|
||||
with open(INIT_FILE, "w", encoding="utf-8") as f:
|
||||
f.write(updated)
|
||||
|
||||
|
||||
@@ -245,10 +225,10 @@ def update_changelog(changelog: str) -> None:
|
||||
changelog = changelog[section_match.start() :]
|
||||
|
||||
try:
|
||||
with open(CHANGELOG_FILE) as f:
|
||||
with open(CHANGELOG_FILE, encoding="utf-8") as f:
|
||||
existing = f.read()
|
||||
except FileNotFoundError:
|
||||
with open(CHANGELOG_FILE, "w") as f:
|
||||
with open(CHANGELOG_FILE, "w", encoding="utf-8") as f:
|
||||
f.write(changelog + "\n")
|
||||
return
|
||||
|
||||
@@ -261,7 +241,7 @@ def update_changelog(changelog: str) -> None:
|
||||
else:
|
||||
# No version sections found — append
|
||||
updated = existing.rstrip() + "\n\n" + changelog + "\n"
|
||||
with open(CHANGELOG_FILE, "w") as f:
|
||||
with open(CHANGELOG_FILE, "w", encoding="utf-8") as f:
|
||||
f.write(updated)
|
||||
|
||||
|
||||
@@ -317,18 +297,16 @@ def run_tests() -> None:
|
||||
click.echo(_("Tests passed."))
|
||||
|
||||
|
||||
def _write_github_output(tag: str) -> None:
|
||||
def _write_release_tag(tag: str) -> None:
|
||||
"""Write the release tag to GITHUB_OUTPUT for downstream jobs.
|
||||
|
||||
This allows a publish job (needs: release) to read the tag via
|
||||
``${{ needs.release.outputs.tag }}`` instead of relying on
|
||||
tag-push event triggering a separate workflow.
|
||||
"""
|
||||
github_output = os.environ.get("GITHUB_OUTPUT")
|
||||
if not github_output:
|
||||
if not os.environ.get("GITHUB_OUTPUT"):
|
||||
return
|
||||
with open(github_output, "a") as f: # noqa: PTH123
|
||||
f.write(f"tag={tag}\n")
|
||||
write_github_output("tag", tag)
|
||||
click.echo(_("Wrote tag {tag} to GITHUB_OUTPUT.", tag=tag))
|
||||
|
||||
|
||||
@@ -360,7 +338,7 @@ def create_and_push_tag(new_version: str, changelog: str, dry_run: bool) -> bool
|
||||
if not dry_run:
|
||||
# Ensure the existing tag is pushed
|
||||
run_cmd(["git", "push", "origin", f"refs/tags/{tag}"], check=False)
|
||||
_write_github_output(tag)
|
||||
_write_release_tag(tag)
|
||||
return False
|
||||
tag_msg = f"Release v{new_version}\n\n{changelog}"
|
||||
if dry_run:
|
||||
@@ -368,7 +346,7 @@ def create_and_push_tag(new_version: str, changelog: str, dry_run: bool) -> bool
|
||||
return True
|
||||
run_cmd(["git", "tag", "-a", tag, "-m", tag_msg])
|
||||
run_cmd(["git", "push", "origin", f"refs/tags/{tag}"])
|
||||
_write_github_output(tag)
|
||||
_write_release_tag(tag)
|
||||
return True
|
||||
|
||||
|
||||
@@ -380,7 +358,7 @@ def create_and_push_tag(new_version: str, changelog: str, dry_run: bool) -> bool
|
||||
def get_init_version() -> str | None:
|
||||
"""Read __version__ from the version file."""
|
||||
try:
|
||||
with open(INIT_FILE) as f:
|
||||
with open(INIT_FILE, encoding="utf-8") as f:
|
||||
content = f.read()
|
||||
match = re.search(r'^__version__\s*=\s*"([^"]*)"', content, flags=re.MULTILINE)
|
||||
return match.group(1) if match else None
|
||||
@@ -391,7 +369,7 @@ def get_init_version() -> str | None:
|
||||
def get_changelog_versions() -> list[str]:
|
||||
"""Extract version numbers from CHANGELOG.md headers, in order."""
|
||||
try:
|
||||
with open(CHANGELOG_FILE) as f:
|
||||
with open(CHANGELOG_FILE, encoding="utf-8") as f:
|
||||
content = f.read()
|
||||
return re.findall(r"^## \[(\d+\.\d+\.\d+)\]", content, flags=re.MULTILINE)
|
||||
except FileNotFoundError:
|
||||
@@ -634,7 +612,7 @@ def main(dry_run: bool, skip_tests: bool, verify: bool) -> None:
|
||||
tag=release_tag,
|
||||
)
|
||||
)
|
||||
_write_github_output(release_tag)
|
||||
_write_release_tag(release_tag)
|
||||
return
|
||||
# Tag is missing — recover by creating and pushing it
|
||||
click.echo(
|
||||
|
||||
@@ -49,7 +49,7 @@ def load_mapping() -> dict[str, str]:
|
||||
|
||||
Validates that the mapping is a dict of string-to-string pairs.
|
||||
"""
|
||||
with open(MAPPING_FILE) as f:
|
||||
with open(MAPPING_FILE, encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
if not isinstance(data, dict):
|
||||
raise click.ClickException(
|
||||
@@ -64,7 +64,7 @@ def load_mapping() -> dict[str, str]:
|
||||
def read_doc_content(file_path: str) -> str:
|
||||
"""Read markdown content from a docs file."""
|
||||
full_path = DOCS_DIR / file_path
|
||||
with open(full_path) as f:
|
||||
with open(full_path, encoding="utf-8") as f:
|
||||
return f.read()
|
||||
|
||||
|
||||
|
||||
@@ -69,7 +69,7 @@ def main(commit_msg_file: str | None, branch: str | None, from_git: bool) -> Non
|
||||
if commit_msg_file == "-":
|
||||
msg = sys.stdin.read().strip()
|
||||
else:
|
||||
with open(commit_msg_file) as f:
|
||||
with open(commit_msg_file, encoding="utf-8") as f:
|
||||
msg = f.read().strip()
|
||||
else:
|
||||
raise click.ClickException(_("Provide a commit message file or use --git."))
|
||||
|
||||
@@ -330,7 +330,7 @@ devx-setup-image:
|
||||
@if [ -d /opt/venv ]; then ln -sf /opt/venv $(DEVX_VENV); . $(DEVX_BIN)/activate; \
|
||||
_U="$${CI_GITEA_USERNAME:-emil}"; \
|
||||
if [ -n "$$CI_GITEA_TOKEN" ]; then export PIP_EXTRA_INDEX_URL="https://$$_U:$$CI_GITEA_TOKEN@$(DEVX_GITEA_PYPI_HOST)/api/packages/$(DEVX_GITEA_PYPI_ORG)/pypi/simple/"; fi; \
|
||||
pip install -e .$(if $(EXTRAS),[$(EXTRAS)],); \
|
||||
pip install --no-cache-dir -e .$(if $(EXTRAS),[$(EXTRAS)],); \
|
||||
echo "[devx-setup-image] Linked /opt/venv$(if $(EXTRAS), with [$(EXTRAS)],)."; \
|
||||
else echo "[devx-setup-image] /opt/venv not found — falling back to setup-ci"; $(MAKE) setup-ci; fi
|
||||
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Molecule testing helpers for Ansible projects."""
|
||||
|
||||
@@ -156,7 +156,7 @@ def main(
|
||||
gh_output = os.environ.get("GITHUB_OUTPUT")
|
||||
if not gh_output:
|
||||
raise click.ClickException("GITHUB_OUTPUT environment variable is not set")
|
||||
with open(gh_output, "a") as f: # noqa: PTH123
|
||||
with open(gh_output, "a", encoding="utf-8") as f: # noqa: PTH123
|
||||
f.write(f"runner-count={count}\n")
|
||||
f.write(f"runner-indices={json.dumps(indices)}\n")
|
||||
click.echo(f"Runner count: {count}")
|
||||
|
||||
@@ -25,6 +25,7 @@ from pathlib import Path
|
||||
|
||||
import click
|
||||
|
||||
from devx.ci._shared import lpt_distribute, write_github_env
|
||||
from devx.i18n import _
|
||||
from devx.molecule.platforms import PLATFORMS, load_platforms
|
||||
|
||||
@@ -216,22 +217,8 @@ def _scenario_weight(scenario: str, role: str | None = None) -> int:
|
||||
|
||||
|
||||
def _lpt_distribute[T](items: list[T], weights: list[int], max_runners: int) -> list[list[T]]:
|
||||
"""Distribute *items* across *max_runners* using LPT (Longest Processing Time first).
|
||||
|
||||
Sorts items by weight (descending), then assigns each to the runner
|
||||
with the least total weight. This produces a more balanced distribution
|
||||
than naive round-robin when items have varying costs.
|
||||
"""
|
||||
groups: list[list[T]] = [[] for _ in range(max_runners)]
|
||||
loads = [0] * max_runners
|
||||
# Sort by weight descending, preserving original order for ties
|
||||
indexed = sorted(enumerate(items), key=lambda x: (-weights[x[0]], x[0]))
|
||||
for orig_idx, item in indexed:
|
||||
# Find the runner with the minimum load
|
||||
min_runner = min(range(max_runners), key=lambda r: loads[r])
|
||||
groups[min_runner].append(item)
|
||||
loads[min_runner] += weights[orig_idx]
|
||||
return groups
|
||||
"""Distribute *items* across *max_runners* using LPT (delegates to shared utility)."""
|
||||
return lpt_distribute(items, weights, max_runners)
|
||||
|
||||
|
||||
def distribute_multi_role(pairs: list[MultiRoleTestPair], max_runners: int) -> list[list[MultiRoleTestPair]]:
|
||||
@@ -283,14 +270,8 @@ def pairs_for_runner(pairs: list[TestPair], runner_index: int, max_runners: int)
|
||||
|
||||
|
||||
def _write_github_env(key: str, value: str) -> None:
|
||||
"""Append a key=value line to the $GITHUB_ENV file."""
|
||||
import os
|
||||
|
||||
gh_env = os.environ.get("GITHUB_ENV")
|
||||
if not gh_env:
|
||||
raise click.ClickException("GITHUB_ENV environment variable is not set")
|
||||
with open(gh_env, "a") as f: # noqa: PTH123
|
||||
f.write(f"{key}={value}\n")
|
||||
"""Append a key=value line to the $GITHUB_ENV file (delegates to shared utility)."""
|
||||
write_github_env(key, value)
|
||||
|
||||
|
||||
@click.command()
|
||||
|
||||
@@ -41,7 +41,7 @@ def load_platforms(platforms_file: str | Path | None = None) -> list[dict[str, s
|
||||
path = Path(platforms_file)
|
||||
if not path.is_file():
|
||||
return PLATFORMS
|
||||
with path.open() as f:
|
||||
with path.open(encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
if not isinstance(data, list) or not data:
|
||||
return PLATFORMS
|
||||
|
||||
@@ -168,7 +168,7 @@ def start_docker_daemon(timeout: int = DEFAULT_TIMEOUT) -> bool:
|
||||
click.echo(_("Docker daemon failed to start"))
|
||||
click.echo("--- dockerd log ---")
|
||||
try:
|
||||
with open(log_file.name) as f:
|
||||
with open(log_file.name, encoding="utf-8") as f:
|
||||
log_content = f.read()
|
||||
click.echo(log_content[-3000:] if len(log_content) > 3000 else log_content)
|
||||
except OSError as e:
|
||||
@@ -191,7 +191,7 @@ def main(timeout: int) -> None:
|
||||
# Export DOCKER_HOST to GITHUB_ENV for subsequent CI steps
|
||||
github_env = os.environ.get("GITHUB_ENV")
|
||||
if github_env and os.environ.get("DOCKER_HOST"):
|
||||
with open(github_env, "a") as f:
|
||||
with open(github_env, "a", encoding="utf-8") as f:
|
||||
f.write(f"DOCKER_HOST={os.environ['DOCKER_HOST']}\n")
|
||||
click.echo(f"Exported DOCKER_HOST={os.environ['DOCKER_HOST']} to GITHUB_ENV")
|
||||
sys.exit(0)
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
"""Shared utilities for tools modules."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import platform
|
||||
|
||||
import click
|
||||
|
||||
|
||||
def arch_string() -> str:
|
||||
"""Return the architecture string used by release assets.
|
||||
|
||||
Maps ``platform.machine()`` to the common release asset naming:
|
||||
``amd64`` for x86_64, ``arm64`` for aarch64.
|
||||
|
||||
Raises:
|
||||
click.ClickException: If the architecture is not supported.
|
||||
"""
|
||||
machine = platform.machine().lower()
|
||||
if machine in {"x86_64", "amd64"}:
|
||||
return "amd64"
|
||||
if machine in {"aarch64", "arm64"}:
|
||||
return "arm64"
|
||||
raise click.ClickException(f"Unsupported architecture: {machine}")
|
||||
@@ -88,7 +88,7 @@ def load_manifest(path: str | Path) -> list[ImageSpec]:
|
||||
p = Path(path)
|
||||
if not p.is_file():
|
||||
raise click.ClickException(_("Manifest file not found: {path}", path=p))
|
||||
with p.open() as f: # noqa: PTH123
|
||||
with p.open(encoding="utf-8") as f: # noqa: PTH123
|
||||
data = json.load(f)
|
||||
if not isinstance(data, list):
|
||||
raise click.ClickException(_("Manifest must be a JSON list"))
|
||||
|
||||
@@ -50,11 +50,17 @@ def _default_branch_protection_config() -> dict[str, Any]:
|
||||
"push_whitelist_usernames": [],
|
||||
"enable_status_check": True,
|
||||
"status_check_contexts": _default_status_checks(),
|
||||
"required_approvals": 0,
|
||||
"required_approvals": 1,
|
||||
"dismiss_stale_approvals": True,
|
||||
"block_on_outdated_branch": True,
|
||||
"block_on_rejected_reviews": True,
|
||||
"block_on_official_review_requests": True,
|
||||
# Prevent admins from force-merging PRs that don't meet branch
|
||||
# protection requirements (e.g. missing approvals). Without this,
|
||||
# an admin token can bypass the approval gate via force_merge=true,
|
||||
# allowing merges that failed the auto-merge CI job to reach master
|
||||
# and trigger the post-merge release pipeline.
|
||||
"block_admin_merge_override": True,
|
||||
}
|
||||
|
||||
|
||||
@@ -123,6 +129,7 @@ def configure_repo(
|
||||
click.echo(_(" - Dismiss stale approvals: yes"))
|
||||
click.echo(_(" - Block outdated branches: yes"))
|
||||
click.echo(_(" - Block rejected reviews: yes"))
|
||||
click.echo(_(" - Block admin merge override: yes"))
|
||||
checks = ", ".join(cast(list[str], bp_config["status_check_contexts"]))
|
||||
click.echo(_(" - Required status checks: {checks}", checks=checks))
|
||||
|
||||
|
||||
@@ -348,7 +348,7 @@ def collect_quality(repo_root: Path) -> dict[str, str | int]:
|
||||
([sys.executable, "-m", "pyright"], "pyright"),
|
||||
([sys.executable, "-m", "bandit", "-r", "src/"], "bandit"),
|
||||
]:
|
||||
rc, _, stderr = run_command(cmd, cwd=repo_root)
|
||||
rc, _stdout, stderr = run_command(cmd, cwd=repo_root)
|
||||
if rc == 0:
|
||||
results.append(True)
|
||||
tool_names.append(f"{name}: pass")
|
||||
|
||||
@@ -7,7 +7,6 @@ pre-built Linux binary from the official GitHub releases.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import platform
|
||||
import shutil
|
||||
import subprocess # nosec B404
|
||||
import urllib.request
|
||||
@@ -15,6 +14,8 @@ from pathlib import Path
|
||||
|
||||
import click
|
||||
|
||||
from devx.tools._shared import arch_string
|
||||
|
||||
CHECKMAKE_VERSION = "0.3.2"
|
||||
RELEASE_URL_TEMPLATE = (
|
||||
"https://github.com/checkmake/checkmake/releases/download/"
|
||||
@@ -23,18 +24,11 @@ RELEASE_URL_TEMPLATE = (
|
||||
TARGET_PATH = Path("/usr/local/bin/checkmake")
|
||||
|
||||
|
||||
def _arch() -> str:
|
||||
"""Return the architecture string used by checkmake releases."""
|
||||
machine = platform.machine().lower()
|
||||
if machine in {"x86_64", "amd64"}:
|
||||
return "amd64"
|
||||
if machine in {"aarch64", "arm64"}:
|
||||
return "arm64"
|
||||
raise click.ClickException(f"Unsupported architecture: {machine}")
|
||||
|
||||
|
||||
def _install_with_go() -> bool:
|
||||
"""Install checkmake using go install if Go is available."""
|
||||
"""Install checkmake using go install if Go is available.
|
||||
|
||||
Returns True if the installation succeeded, False if Go is not installed.
|
||||
"""
|
||||
go_bin = shutil.which("go")
|
||||
if go_bin is None:
|
||||
return False
|
||||
@@ -51,7 +45,7 @@ def _install_with_go() -> bool:
|
||||
|
||||
def _download_binary() -> None:
|
||||
"""Download the prebuilt checkmake binary for the current architecture."""
|
||||
url = RELEASE_URL_TEMPLATE.format(arch=_arch())
|
||||
url = RELEASE_URL_TEMPLATE.format(arch=arch_string())
|
||||
urllib.request.urlretrieve(url, TARGET_PATH) # nosec B310
|
||||
TARGET_PATH.chmod(0o755)
|
||||
|
||||
|
||||
@@ -44,13 +44,10 @@ HADOLINT_VERSION = "2.12.0"
|
||||
|
||||
|
||||
def _arch() -> str:
|
||||
"""Return the architecture string used by release assets."""
|
||||
machine = platform.machine().lower()
|
||||
if machine in {"x86_64", "amd64"}:
|
||||
return "amd64"
|
||||
if machine in {"aarch64", "arm64"}:
|
||||
return "arm64"
|
||||
raise click.ClickException(f"Unsupported architecture: {machine}")
|
||||
"""Return the architecture string used by release assets (delegates to shared utility)."""
|
||||
from devx.tools._shared import arch_string
|
||||
|
||||
return arch_string()
|
||||
|
||||
|
||||
def _ensure_target_dir() -> Path:
|
||||
|
||||
@@ -2270,5 +2270,13 @@
|
||||
"pl": "Review body must be at least 50 characters.",
|
||||
"ru": "Review body must be at least 50 characters.",
|
||||
"zh": "Review body must be at least 50 characters."
|
||||
},
|
||||
" - Block admin merge override: yes": {
|
||||
"bg": " - Блокиране на admin merge override: да",
|
||||
"de": " - Admin-Merge-Override blockieren: ja",
|
||||
"en": " - Block admin merge override: yes",
|
||||
"pl": " - Blokuj admin merge override: tak",
|
||||
"ru": " - Блокировать admin merge override: да",
|
||||
"zh": " - 阻止管理员合并覆盖：是"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -6,7 +6,7 @@ from unittest.mock import MagicMock, patch
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from devx.api_clients import GiteaClient, VikunjaClient, _is_retryable, _parse_error
|
||||
from devx.api_clients import GiteaClient, VikunjaClient, _parse_error
|
||||
from devx.config import (
|
||||
DEFAULT_PER_PAGE,
|
||||
DEFAULT_TIMEOUT,
|
||||
@@ -507,7 +507,7 @@ class TestGiteaClient:
|
||||
assert result["id"] == 1
|
||||
assert client._session.request.call_count == 2
|
||||
|
||||
@patch("devx.api_clients.time.sleep")
|
||||
@patch("time.sleep")
|
||||
def test_request_retries_on_429(self, mock_sleep: MagicMock) -> None:
|
||||
"""Should retry on 429 rate limit with exponential backoff."""
|
||||
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
||||
@@ -520,7 +520,7 @@ class TestGiteaClient:
|
||||
assert client._session.request.call_count == 3
|
||||
assert mock_sleep.call_count == 2
|
||||
|
||||
@patch("devx.api_clients.time.sleep")
|
||||
@patch("time.sleep")
|
||||
def test_request_retries_on_503(self, mock_sleep: MagicMock) -> None:
|
||||
"""Should retry on 503 service unavailable."""
|
||||
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
||||
@@ -532,7 +532,7 @@ class TestGiteaClient:
|
||||
assert result.json() == {"ok": True}
|
||||
assert client._session.request.call_count == 2
|
||||
|
||||
@patch("devx.api_clients.time.sleep")
|
||||
@patch("time.sleep")
|
||||
def test_request_no_retry_on_404(self, mock_sleep: MagicMock) -> None:
|
||||
"""Should NOT retry on 404 — it's not a transient error."""
|
||||
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
||||
@@ -545,7 +545,7 @@ class TestGiteaClient:
|
||||
assert client._session.request.call_count == 1
|
||||
mock_sleep.assert_not_called()
|
||||
|
||||
@patch("devx.api_clients.time.sleep")
|
||||
@patch("time.sleep")
|
||||
def test_request_retries_on_connection_error(self, mock_sleep: MagicMock) -> None:
|
||||
"""Should retry on connection errors."""
|
||||
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
||||
@@ -555,7 +555,7 @@ class TestGiteaClient:
|
||||
assert result.json() == {"ok": True}
|
||||
assert client._session.request.call_count == 2
|
||||
|
||||
@patch("devx.api_clients.time.sleep")
|
||||
@patch("time.sleep")
|
||||
def test_request_max_retries_exhausted(self, mock_sleep: MagicMock) -> None:
|
||||
"""Should raise APIError after max retries on persistent 503."""
|
||||
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
||||
@@ -567,7 +567,7 @@ class TestGiteaClient:
|
||||
assert exc_info.value.status == 503
|
||||
assert client._session.request.call_count == 3 # MAX_RETRIES
|
||||
|
||||
@patch("devx.api_clients.time.sleep")
|
||||
@patch("time.sleep")
|
||||
def test_request_connection_error_exhausted(self, mock_sleep: MagicMock) -> None:
|
||||
"""Should raise APIError after max retries on persistent connection errors."""
|
||||
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
|
||||
@@ -691,7 +691,7 @@ class TestVikunjaClient:
|
||||
json={"id": 42, "title": "My task", "done": True},
|
||||
)
|
||||
|
||||
@patch("devx.api_clients.time.sleep")
|
||||
@patch("time.sleep")
|
||||
def test_http_error_raises_api_error(self, mock_sleep: MagicMock) -> None:
|
||||
client = VikunjaClient("https://work.example.com", "tok")
|
||||
mock_resp = MagicMock()
|
||||
@@ -713,7 +713,7 @@ class TestVikunjaClient:
|
||||
client.list_tasks()
|
||||
assert "connection failed" in str(exc_info.value)
|
||||
|
||||
@patch("devx.api_clients.time.sleep")
|
||||
@patch("time.sleep")
|
||||
def test_vikunja_retries_on_503(self, mock_sleep: MagicMock) -> None:
|
||||
"""VikunjaClient should also retry on 503."""
|
||||
client = VikunjaClient("https://work.example.com", "tok")
|
||||
@@ -725,7 +725,7 @@ class TestVikunjaClient:
|
||||
assert len(result) == 1
|
||||
assert client._session.request.call_count == 2
|
||||
|
||||
@patch("devx.api_clients.time.sleep")
|
||||
@patch("time.sleep")
|
||||
def test_vikunja_retries_on_connection_error(self, mock_sleep: MagicMock) -> None:
|
||||
"""VikunjaClient should retry on connection errors."""
|
||||
client = VikunjaClient("https://work.example.com", "tok")
|
||||
@@ -735,7 +735,7 @@ class TestVikunjaClient:
|
||||
assert len(result) == 1
|
||||
assert client._session.request.call_count == 2
|
||||
|
||||
@patch("devx.api_clients.time.sleep")
|
||||
@patch("time.sleep")
|
||||
def test_vikunja_max_retries_exhausted(self, mock_sleep: MagicMock) -> None:
|
||||
"""VikunjaClient should raise APIError after max retries on persistent 503."""
|
||||
client = VikunjaClient("https://work.example.com", "tok")
|
||||
@@ -747,7 +747,7 @@ class TestVikunjaClient:
|
||||
assert exc_info.value.status == 503
|
||||
assert client._session.request.call_count == 3 # MAX_RETRIES
|
||||
|
||||
@patch("devx.api_clients.time.sleep")
|
||||
@patch("time.sleep")
|
||||
def test_vikunja_connection_error_exhausted(self, mock_sleep: MagicMock) -> None:
|
||||
"""VikunjaClient should raise APIError after max retries on persistent connection errors."""
|
||||
client = VikunjaClient("https://work.example.com", "tok")
|
||||
@@ -818,22 +818,6 @@ class TestVikunjaClient:
|
||||
assert result["title"] == "Found on page 2"
|
||||
|
||||
|
||||
class TestIsRetryable:
|
||||
def test_connection_error_is_retryable(self) -> None:
|
||||
assert _is_retryable(requests.ConnectionError("refused")) is True
|
||||
|
||||
def test_timeout_is_retryable(self) -> None:
|
||||
assert _is_retryable(requests.Timeout("timed out")) is True
|
||||
|
||||
def test_429_is_retryable(self) -> None:
|
||||
err = _mock_http_error(429, "rate limited")
|
||||
assert _is_retryable(err) is True
|
||||
|
||||
def test_404_is_not_retryable(self) -> None:
|
||||
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")
|
||||
@@ -913,8 +897,3 @@ class TestGiteaClientActions:
|
||||
"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
|
||||
|
||||
@@ -6,12 +6,12 @@ import click
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
from devx.ci._shared import run_cmd
|
||||
from devx.ci.auto_merge import (
|
||||
extract_conventional_msg,
|
||||
extract_task_id,
|
||||
main,
|
||||
read_taskid,
|
||||
run_cmd,
|
||||
validate_pr_title,
|
||||
validate_pr_title_matches_vikunja,
|
||||
)
|
||||
@@ -399,7 +399,7 @@ class TestMain:
|
||||
mock_client.merge_pr.side_effect = APIError(405, "HEAD branch is behind master")
|
||||
mock_client_cls.return_value = mock_client
|
||||
|
||||
with patch("devx.ci.auto_merge.run_cmd") as mock_run:
|
||||
with patch("devx.ci._shared.run_cmd") as mock_run:
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(
|
||||
main,
|
||||
|
||||
@@ -92,8 +92,10 @@ class TestGetPrTitleFromGitea:
|
||||
|
||||
@patch("devx.ci.check_auto_merge_ready.GiteaClient")
|
||||
def test_returns_none_on_exception(self, mock_client_cls: MagicMock) -> None:
|
||||
from devx.exceptions import APIError
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_client.get_pr.side_effect = Exception("API error")
|
||||
mock_client.get_pr.side_effect = APIError(500, "API error")
|
||||
mock_client_cls.return_value = mock_client
|
||||
with patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True):
|
||||
result = get_pr_title_from_gitea("owner/repo", 1)
|
||||
|
||||
@@ -32,9 +32,10 @@ class TestDefaultConfigs:
|
||||
assert config["branch_name"] == "master"
|
||||
assert config["enable_push"] is True
|
||||
assert config["enable_push_whitelist"] is False
|
||||
assert config["required_approvals"] == 0
|
||||
assert config["required_approvals"] == 1
|
||||
assert isinstance(config["status_check_contexts"], list)
|
||||
assert "CI / quality (pull_request)" in config["status_check_contexts"]
|
||||
assert config["block_admin_merge_override"] is True
|
||||
|
||||
def test_default_repo_settings_config(self) -> None:
|
||||
config = _default_repo_settings_config()
|
||||
|
||||
@@ -8,21 +8,22 @@ import pytest
|
||||
from click import ClickException
|
||||
|
||||
import devx.tools.install_checkmake as install_checkmake
|
||||
from devx.tools._shared import arch_string
|
||||
|
||||
|
||||
class TestArch:
|
||||
class TestArchString:
|
||||
def test_amd64(self) -> None:
|
||||
with patch.object(platform, "machine", return_value="x86_64"):
|
||||
assert install_checkmake._arch() == "amd64"
|
||||
assert arch_string() == "amd64"
|
||||
|
||||
def test_arm64(self) -> None:
|
||||
with patch.object(platform, "machine", return_value="aarch64"):
|
||||
assert install_checkmake._arch() == "arm64"
|
||||
assert arch_string() == "arm64"
|
||||
|
||||
def test_unsupported(self) -> None:
|
||||
with patch.object(platform, "machine", return_value="riscv64"):
|
||||
with pytest.raises(ClickException):
|
||||
install_checkmake._arch()
|
||||
arch_string()
|
||||
|
||||
|
||||
class TestInstallWithGo:
|
||||
|
||||
@@ -34,20 +34,20 @@ from devx.ci.release import (
|
||||
|
||||
|
||||
class TestRunCmd:
|
||||
@patch("devx.ci.release.subprocess.run")
|
||||
@patch("devx.ci._shared.subprocess.run")
|
||||
def test_success(self, mock_run: MagicMock) -> None:
|
||||
mock_run.return_value = MagicMock(returncode=0, stderr="", stdout="")
|
||||
result = run_cmd(["echo", "hi"])
|
||||
assert result.returncode == 0
|
||||
mock_run.assert_called_once()
|
||||
|
||||
@patch("devx.ci.release.subprocess.run")
|
||||
@patch("devx.ci._shared.subprocess.run")
|
||||
def test_failure_raises(self, mock_run: MagicMock) -> None:
|
||||
mock_run.return_value = MagicMock(returncode=1, stderr="err", stdout="")
|
||||
with pytest.raises(click.ClickException):
|
||||
run_cmd(["false"])
|
||||
|
||||
@patch("devx.ci.release.subprocess.run")
|
||||
@patch("devx.ci._shared.subprocess.run")
|
||||
def test_check_false_no_raise(self, mock_run: MagicMock) -> None:
|
||||
mock_run.return_value = MagicMock(returncode=1, stderr="err", stdout="")
|
||||
result = run_cmd(["false"], check=False)
|
||||
|
||||
@@ -272,18 +272,16 @@ class TestMain:
|
||||
assert result.exit_code == 0
|
||||
mock_start.assert_called_once_with(60)
|
||||
|
||||
@patch("devx.molecule.start_docker.os.environ.get")
|
||||
@patch("devx.molecule.start_docker.start_docker_daemon", return_value=True)
|
||||
def test_exports_github_env(self, mock_start: MagicMock, mock_get: MagicMock) -> None:
|
||||
def test_exports_github_env(self, mock_start: MagicMock) -> None:
|
||||
"""Should write DOCKER_HOST to GITHUB_ENV when available."""
|
||||
mock_get.side_effect = lambda key, default="": (
|
||||
"/tmp/github_env" if key == "GITHUB_ENV" else f"unix://{DOCKER_SOCK}" if key == "DOCKER_HOST" else default
|
||||
)
|
||||
with patch("builtins.open", mock_open()) as mock_file:
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, [])
|
||||
assert result.exit_code == 0
|
||||
mock_file.assert_called_with("/tmp/github_env", "a")
|
||||
env = {"GITHUB_ENV": "/tmp/github_env", "DOCKER_HOST": f"unix://{DOCKER_SOCK}"}
|
||||
with patch.dict("os.environ", env, clear=True):
|
||||
with patch("builtins.open", mock_open()) as mock_file:
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, [])
|
||||
assert result.exit_code == 0
|
||||
mock_file.assert_called_with("/tmp/github_env", "a", encoding="utf-8")
|
||||
|
||||
@patch("devx.molecule.start_docker.os.environ.get", return_value="")
|
||||
@patch("devx.molecule.start_docker.start_docker_daemon", return_value=True)
|
||||
|
||||
Reference in New Issue
Block a user