Compare commits

...
20 Commits
Author SHA1 Message Date
devx-ci-bot 951ba7de7a release: v0.33.4 [skip ci] 2026-07-06 06:18:39 +00:00
emil e796b06a91 DEVX-117: refactor: remove project-specific references from devx
Post-merge / detect-type (push) Successful in 9s
Post-merge / validate-commit-msg (push) Successful in 9s
Post-merge / vikunja (push) Successful in 15s
Post-merge / configure-repo (push) Successful in 19s
Post-merge / release (push) Successful in 45s
Post-merge / sync-wiki (push) Successful in 46s
Post-merge / badges (push) Successful in 46s
Post-merge / publish (push) Successful in 17s
2026-07-06 06:17:52 +00:00
gitea-actions-bot 990f2fa612 chore: update badge URLs to commit 7802ce60 [skip ci] 2026-07-06 04:56:16 +00:00
devx-ci-bot a7f5f47564 release: v0.33.3 [skip ci] 2026-07-06 04:56:04 +00:00
emil d623a64344 DEVX-115: fix: make wiki sync resilient to API timeouts and stale page lists
Post-merge / detect-type (push) Successful in 12s
Post-merge / validate-commit-msg (push) Successful in 9s
Post-merge / vikunja (push) Successful in 16s
Post-merge / release (push) Successful in 39s
Post-merge / configure-repo (push) Successful in 17s
Post-merge / sync-wiki (push) Successful in 45s
Post-merge / badges (push) Successful in 47s
Post-merge / publish (push) Successful in 18s
2026-07-06 04:55:06 +00:00
gitea-actions-bot 268a4e7988 chore: update badge URLs to commit b07bea6f [skip ci] 2026-07-05 20:47:45 +00:00
emil 7daaf9e4a9 DEVX-114: ci: add testing-and-debugging skill for devx repo
Post-merge / detect-type (push) Successful in 13s
Post-merge / validate-commit-msg (push) Successful in 9s
Post-merge / release (push) Successful in 17s
Post-merge / publish (push) Has been skipped
Post-merge / vikunja (push) Successful in 17s
Post-merge / configure-repo (push) Successful in 20s
Post-merge / sync-wiki (push) Successful in 43s
Post-merge / badges (push) Successful in 54s
2026-07-05 20:46:21 +00:00
gitea-actions-bot b7c9334881 chore: update badge URLs to commit 83595808 [skip ci] 2026-07-05 19:18:21 +00:00
devx-ci-bot 3406639f13 release: v0.33.2 [skip ci] 2026-07-05 19:18:03 +00:00
emil 9f02ccb40d DEVX-113: fix: abort sync_wiki when list_wiki_pages fails
Post-merge / detect-type (push) Successful in 11s
Post-merge / validate-commit-msg (push) Successful in 10s
Post-merge / vikunja (push) Successful in 22s
Post-merge / configure-repo (push) Successful in 18s
Post-merge / release (push) Successful in 47s
Post-merge / badges (push) Successful in 54s
Post-merge / sync-wiki (push) Successful in 55s
Post-merge / publish (push) Successful in 31s
2026-07-05 19:17:10 +00:00
gitea-actions-bot 5206158603 chore: update badge URLs to commit 66fec9ab [skip ci] 2026-07-05 14:47:30 +00:00
devx-ci-bot 489cc8343a release: v0.33.1 [skip ci] 2026-07-05 14:47:16 +00:00
emil 20ea80135c DEVX-112: fix: build images after post-merge publish, not on push
Post-merge / detect-type (push) Successful in 10s
Post-merge / validate-commit-msg (push) Successful in 10s
Post-merge / sync-wiki (push) Successful in 28s
Post-merge / release (push) Successful in 32s
Post-merge / vikunja (push) Successful in 13s
Post-merge / badges (push) Successful in 36s
Post-merge / configure-repo (push) Successful in 14s
Post-merge / publish (push) Successful in 20s
2026-07-05 14:46:33 +00:00
gitea-actions-bot 53b1d300aa chore: update badge URLs to commit 546910d3 [skip ci] 2026-07-05 14:13:08 +00:00
devx-ci-bot 5b9e92f324 release: v0.33.0 [skip ci] 2026-07-05 14:12:51 +00:00
emil 2c0118111d DEVX-111: feat: add check_api_identity_checks, setup_ssh_key, and api utils
Post-merge / detect-type (push) Successful in 9s
Post-merge / validate-commit-msg (push) Successful in 9s
Build Images / detect-type (push) Successful in 42s
Post-merge / vikunja (push) Successful in 17s
Post-merge / release (push) Successful in 52s
Post-merge / configure-repo (push) Successful in 23s
Post-merge / badges (push) Successful in 55s
Post-merge / sync-wiki (push) Successful in 58s
Post-merge / publish (push) Successful in 21s
Build Images / build-and-push (push) Successful in 3m15s
Build Images / cleanup (push) Successful in 3m38s
2026-07-05 14:11:58 +00:00
gitea-actions-bot 333641f862 chore: update badge URLs to commit e8088b8e [skip ci] 2026-07-01 23:37:55 +00:00
devx-ci-bot f21b01dce2 release: v0.32.1 [skip ci] 2026-07-01 23:37:11 +00:00
emil ff80745eea DEVX-110: fix: add missing i18n translations for new tools
Post-merge / detect-type (push) Successful in 15s
Post-merge / validate-commit-msg (push) Successful in 14s
Build Images / detect-type (push) Successful in 56s
Post-merge / release (push) Successful in 1m2s
Post-merge / configure-repo (push) Successful in 28s
Post-merge / sync-wiki (push) Successful in 1m11s
Post-merge / badges (push) Successful in 1m19s
Post-merge / vikunja (push) Successful in 1m15s
Post-merge / publish (push) Successful in 33s
Build Images / build-and-push (push) Successful in 4m22s
Build Images / cleanup (push) Successful in 2m50s
2026-07-01 23:35:56 +00:00
gitea-actions-bot 319807f41c chore: update badge URLs to commit ce9bf024 [skip ci] 2026-07-01 23:01:36 +00:00
41 changed files with 1199 additions and 131 deletions
@@ -0,0 +1,98 @@
# testing-and-debugging
Make targets for testing, debugging, and CI investigation. **Use these
instead of raw `pytest`, `ruff`, or `actionlint` commands.**
## Why Make Targets
Make targets encapsulate the correct venv activation, PYTHONPATH, env
vars, and flags. Running raw commands bypasses venv activation and
produces false failures (missing dependencies, wrong Python version).
## Unit Tests
| Task | Command | Notes |
|------|---------|-------|
| Run all unit tests | `make test-unit` | Fast, no coverage |
| Run with coverage | `make pytest-cov` | **Required before push** — enforces 100% |
| Run single test | `make pytest-cov TEST=tests/test_foo.py::test_bar` | |
| Check test speed | `make check-test-speed` | Fails if tests > 10s total or > 0.5s each |
| Check test coverage | `make check-test-coverage` | Fails if source changed but tests didn't |
## Linting
| Task | Command | Notes |
|------|---------|-------|
| Full lint | `make lint-all` | ruff + workflow-lint + lint-dockerfiles |
| Ruff only | `make lint-ruff` | |
| Format check | `make lint-format` | |
| Type check | `make typecheck` | pyright |
| Bandit | `make lint-bandit` | Security linter |
| Workflow lint | `make workflow-check` | actionlint + act_runner dry-run |
| Dockerfile lint | `make lint-dockerfiles` | hadolint on all Dockerfiles |
| Check mutable globals | `make check-mutable-globals` | Detects module-level mutable state |
| Check dep docs | `make check-dep-docs` | Verifies pyproject.toml deps have comments |
## Pre-Push Verification
**Before pushing any branch:**
```bash
make pre-push
```
This runs `lint-all` + `pytest-cov`. The pre-push git hook only
validates the Vikunja task exists — it does NOT run tests. You must
run `make pre-push` manually.
## CI Failure Investigation
When investigating a CI failure:
1. **Fetch logs via MCP** — use `mcp_call_tool` with gitea server,
`actions_run_read` method, `download_job_log` tool
2. **Reproduce locally** — use `make pytest-cov` or `make lint-all`
depending on which CI job failed
3. **Never run raw pytest** — always use the make target
## Virtual Environment
All commands run inside `.venv`. `make` targets handle activation
automatically. For raw commands (rare), activate first:
```bash
source activate.sh # bash/zsh
source activate.fish # fish
source activate.zsh # zsh
```
If `.venv` doesn't exist, run `make setup` first.
## Common Pitfalls
### Coverage Verification Before Push
**Always run `make pytest-cov` before pushing** — CI enforces 100%
coverage and will fail the PR if any lines are uncovered. This is the
most common cause of CI quality job failures after code changes. The
pre-push git hook only validates Vikunja task existence, not tests.
### API Response Type Checking
Never use `is True`/`is False` identity checks on API response values.
Many APIs return boolean values as strings (`"true"`/`"false"`). Use
the `is_truthy()`/`is_falsy()` helpers from `devx.utils.api` or compare
against string values.
### Time Mocking in Tests
Always mock `time.sleep` and `time.monotonic` in unit tests using
`@patch` decorators. Real sleep calls make tests slow and exceed test
speed limits (10s total, 0.5s per test).
### Mutable Global State
The `check-mutable-globals` tool detects module-level mutable state
(lists, dicts, sets) that can cause test pollution. Avoid module-level
mutable defaults — use factory functions or `None` with initialization
inside functions.
+11 -7
View File
@@ -5,7 +5,9 @@ name: Build Images
# devx and all dependencies into the image.
#
# Triggers:
# - On push to master (after post-merge release completes)
# - After post-merge workflow completes successfully (workflow_run)
# This ensures images are only rebuilt AFTER the release is published
# to PyPI, so the image always has the latest released version.
# - Manually via workflow_dispatch
#
# The workflow builds 3 tier images in sequence:
@@ -15,12 +17,10 @@ name: Build Images
# After pushing, a cleanup job removes old versions (keeps last 2 + latest).
on:
push:
workflow_run:
workflows: ["Post-merge"]
types: [completed]
branches: [master]
paths:
- docker/**
- pyproject.toml
- src/devx/**
workflow_dispatch:
concurrency:
@@ -49,7 +49,11 @@ jobs:
build-and-push:
needs: [detect-type]
if: needs.detect-type.outputs.is-release == 'false'
if: >-
needs.detect-type.outputs.is-release == 'false' && (
github.event_name == 'workflow_dispatch' ||
(github.event_name == 'workflow_run' && github.event.workflow_run.conclusion == 'success')
)
runs-on: docker
timeout-minutes: 30
steps:
+4 -1
View File
@@ -169,7 +169,10 @@ jobs:
if: needs.detect-type.outputs.is-release == 'false'
runs-on: docker
container: git.oblachno.oblachno.fyi/oblachno-oss/runner-images/ci-base:latest
timeout-minutes: 10
timeout-minutes: 15
concurrency:
group: sync-wiki-${{ github.repository }}
cancel-in-progress: false
defaults:
run:
shell: bash
+38
View File
@@ -49,6 +49,44 @@ repos:
pass_filenames: false
stages: [pre-commit]
- id: checkmake
name: checkmake Makefile linter
entry: make checkmake
language: system
files: (Makefile|\.mak)$
pass_filenames: false
stages: [pre-commit]
- id: check-test-speed
name: unit test speed check
entry: .venv/bin/python -m devx.tools.check_test_speed --max-seconds 6 --max-single-seconds 0.5
language: system
types: [python]
pass_filenames: false
stages: [pre-commit]
- id: check-translations
name: translation completeness check
entry: env PYTHONPATH=src .venv/bin/python -m devx.ci.check_translations
language: system
files: ^src/devx/translations\.json$
pass_filenames: false
stages: [pre-commit]
- id: doc-coverage
name: documentation coverage check
entry: env PYTHONPATH=src .venv/bin/python -m devx.ci.doc_coverage --fail-on-missing
language: system
pass_filenames: false
stages: [pre-commit]
- id: lint-docs
name: documentation lint check
entry: env PYTHONPATH=src .venv/bin/python -m devx.ci.lint_docs --root .
language: system
pass_filenames: false
stages: [pre-commit]
- id: pytest-cov
name: pytest with 100% coverage
entry: make pytest-cov
+1 -1
View File
@@ -572,7 +572,7 @@ the user should not need to specify which profile to use.
### Available Profiles
**Global** (shared with infra and grm):
**Global** (shared across all projects):
| Profile | Location | Purpose |
|---------|----------|---------|
+36
View File
@@ -2,6 +2,42 @@
All notable changes to this project will be documented in this file.
## [0.33.4] - 2026-07-06
### Refactor
- Remove project-specific references from devx
## [0.33.3] - 2026-07-06
### Bug Fixes
- Make wiki sync resilient to API timeouts and stale page lists
## [0.33.2] - 2026-07-05
### Bug Fixes
- Abort sync_wiki when list_wiki_pages fails
## [0.33.1] - 2026-07-05
### Bug Fixes
- Build images after post-merge publish, not on push
## [0.33.0] - 2026-07-05
### Features
- Add check_api_identity_checks, setup_ssh_key, and api utils
## [0.32.1] - 2026-07-01
### Bug Fixes
- Add missing i18n translations for new tools
## [0.32.0] - 2026-07-01
### Features
+1 -4
View File
@@ -115,10 +115,7 @@ lint-all: lint workflow-lint lint-dockerfiles
# devx's own CI images may have an older devx.mak. Consumer repos can safely alias.
lint-dockerfiles:
@echo "[lint-dockerfiles] Linting Dockerfiles with hadolint..."
@if ! command -v hadolint >/dev/null 2>&1; then \
echo "[lint-dockerfiles] ERROR: hadolint not found. Install from https://github.com/hadolint/hadolint/releases" >&2; \
exit 1; \
fi
@command -v hadolint >/dev/null 2>&1 || { echo "hadolint not found" >&2; exit 1; }
@find docker -name 'Dockerfile*' -exec hadolint {} +
@echo "[lint-dockerfiles] All Dockerfiles passed."
+6 -6
View File
@@ -16,12 +16,12 @@ quality badges.
[![CI](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions/workflows/ci.yml/badge.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![License: GPL-3.0](https://img.shields.io/badge/license-GPL--3.0-blue)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/src/branch/master/LICENSE)
[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/de35661c34e1b942f36763b86b3289f9fb01e4d7/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/de35661c34e1b942f36763b86b3289f9fb01e4d7/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/de35661c34e1b942f36763b86b3289f9fb01e4d7/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/de35661c34e1b942f36763b86b3289f9fb01e4d7/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/de35661c34e1b942f36763b86b3289f9fb01e4d7/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/de35661c34e1b942f36763b86b3289f9fb01e4d7/python.svg)](https://www.python.org/downloads/)
[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/7802ce6061b3c947821bf4884d66a053d378a456/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/7802ce6061b3c947821bf4884d66a053d378a456/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/7802ce6061b3c947821bf4884d66a053d378a456/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/7802ce6061b3c947821bf4884d66a053d378a456/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/7802ce6061b3c947821bf4884d66a053d378a456/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/7802ce6061b3c947821bf4884d66a053d378a456/python.svg)](https://www.python.org/downloads/)
## Why devx?
+6 -6
View File
@@ -12,12 +12,12 @@ project to be reusable across all oblachno-oss repositories.
[![CI](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions/workflows/ci.yml/badge.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![License: GPL-3.0](https://img.shields.io/badge/license-GPL--3.0-blue)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/src/branch/master/LICENSE)
[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/de35661c34e1b942f36763b86b3289f9fb01e4d7/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/de35661c34e1b942f36763b86b3289f9fb01e4d7/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/de35661c34e1b942f36763b86b3289f9fb01e4d7/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/de35661c34e1b942f36763b86b3289f9fb01e4d7/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/de35661c34e1b942f36763b86b3289f9fb01e4d7/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/de35661c34e1b942f36763b86b3289f9fb01e4d7/python.svg)](https://www.python.org/downloads/)
[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/7802ce6061b3c947821bf4884d66a053d378a456/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/7802ce6061b3c947821bf4884d66a053d378a456/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/7802ce6061b3c947821bf4884d66a053d378a456/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/7802ce6061b3c947821bf4884d66a053d378a456/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/7802ce6061b3c947821bf4884d66a053d378a456/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/raw/commit/7802ce6061b3c947821bf4884d66a053d378a456/python.svg)](https://www.python.org/downloads/)
## Overview
+1 -1
View File
@@ -1,3 +1,3 @@
"""devx — reusable development and CI/CD tools for oblachno-oss projects."""
__version__ = "0.32.0"
__version__ = "0.33.4"
-1
View File
@@ -185,7 +185,6 @@ def main(translations: tuple[Path, ...], source_dir: str | None) -> None:
# Try common locations
candidates = [
root / "src" / "devx" / "translations.json",
root / "src" / "gitea_runner_manager" / "translations.json",
]
# Also search for any translations.json in src/
for match in root.glob("src/*/translations.json"):
+2 -2
View File
@@ -7,7 +7,7 @@ ordering, then assigned to *max_runners* groups using LPT (Longest
Processing Time first) scheduling.
Each item is a string (e.g. an Ansible ``--limit`` pattern like
``observability`` or ``infra-314-vm``). Optionally, items can be objects
``observability`` or ``customer-1-vm``). Optionally, items can be objects
with ``{"id": "...", "weight": N}`` to provide explicit weights.
The assigned group for *runner_index* is written to ``$GITHUB_ENV`` as
@@ -15,7 +15,7 @@ The assigned group for *runner_index* is written to ``$GITHUB_ENV`` as
Usage::
echo '["observability", "infra-314-vm"]' | \\
echo '["observability", "customer-1-vm"]' | \\
python3 -m devx.ci.distribute_items \\
--runner-index 1 --max-runners 3 \\
--github-env --skip-if-excess
+47 -18
View File
@@ -122,6 +122,9 @@ def sync_page(
"""Create or update a single wiki page.
Returns "created", "updated", or "skipped" (if dry-run).
If a create fails with HTTP 400 "already exists" (the page list was
stale), re-lists the wiki and falls back to an update.
"""
if dry_run:
click.echo(_("[dry-run] Would sync page: {title} ({chars} chars)", title=page_title, chars=len(content)))
@@ -144,16 +147,36 @@ def sync_page(
return "updated"
# Create new page via POST /wiki/new
client._request(
"POST",
"/wiki/new",
json={
"title": page_title,
"content_base64": content_b64,
"message": f"Sync from docs/ — create {page_title}",
},
)
return "created"
try:
client._request(
"POST",
"/wiki/new",
json={
"title": page_title,
"content_base64": content_b64,
"message": f"Sync from docs/ — create {page_title}",
},
)
return "created"
except APIError as e:
if e.status == 400 and "already exists" in e.message.lower():
# The page list was stale (e.g. after a timeout-retry returned
# incomplete data). Re-list and fall back to update.
click.echo(_(" Page '{title}' already exists (stale list). Re-listing and updating...", title=page_title))
fresh_pages = _list_wiki_pages_with_retry(client)
if page_title in fresh_pages:
sub_url = fresh_pages[page_title]
client._request(
"PATCH",
f"/wiki/page/{sub_url}",
json={
"title": page_title,
"content_base64": content_b64,
"message": f"Sync from docs/ — update {page_title} (create→update fallback)",
},
)
return "updated"
raise
def verify_wiki_page(
@@ -173,15 +196,15 @@ def verify_wiki_page(
def _list_wiki_pages_with_retry(client: GiteaClient) -> dict[str, str]:
"""List wiki pages with tenacity retry on APIError.
The Gitea API can be briefly unavailable right after a batch of wiki
page updates. Uses the same tenacity pattern as ``api_clients`` for
exponential backoff.
The Gitea wiki API can be slow (it renders pages on each request)
and may time out. Uses 5 attempts with exponential backoff to handle
transient slowness.
"""
_logger = logging.getLogger("sync_wiki")
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=2, min=2, max=8),
stop=stop_after_attempt(5),
wait=wait_exponential(multiplier=2, min=2, max=16),
retry=retry_if_exception_type(APIError),
before_sleep=before_sleep_log(_logger, logging.WARNING),
reraise=True,
@@ -291,9 +314,15 @@ def main(dry_run: bool, repo: str | None, verify: bool, strict: bool) -> None:
click.echo(_("Syncing {count} documentation pages to wiki...", count=len(mapping)))
try:
existing_pages = list_wiki_pages(client)
except APIError:
existing_pages = {}
existing_pages = _list_wiki_pages_with_retry(client)
except APIError as e:
raise click.ClickException(
_(
"Failed to list existing wiki pages after retries: {error}. "
"Aborting to avoid creating duplicate pages.",
error=e,
)
) from e
if existing_pages:
click.echo(_("Found {count} existing wiki pages.", count=len(existing_pages)))
+10 -1
View File
@@ -66,7 +66,7 @@ DEVX_PIP_INSTALL := if [ -z "$$CI_GITEA_TOKEN" ]; then . ./.env 2>/dev/null; fi;
# ── Virtual environment management ────────────────────────────────────────────
#
# These targets provide a single, consistent venv setup across all
# devx-integrated projects (infra, grm, devx). Each project includes
# devx-integrated projects. Each project includes
# devx.mak and aliases its local targets to these.
#
# The venv is a standard .venv directory (no pyenv virtualenv dependency).
@@ -110,6 +110,7 @@ devx-ensure-venv:
.PHONY: devx-lint-ruff devx-lint-format devx-typecheck devx-lint-bandit devx-lint-deps devx-lint
.PHONY: devx-clean devx-pre-push
.PHONY: devx-check-mutable-globals devx-check-dep-docs devx-check-test-coverage devx-check-docs devx-check-test-speed
.PHONY: devx-check-api-identity-checks devx-setup-ssh-key
.PHONY: devx-test-unit devx-pytest-cov
.PHONY: devx-setup-image devx-lint-dockerfiles
@@ -327,6 +328,14 @@ devx-check-docs:
devx-check-test-speed:
@$(DEVX_PYTHON) -m devx.tools.check_test_speed
# Scan integration tests for unsafe is True/is False identity checks
devx-check-api-identity-checks:
@$(DEVX_PYTHON) -m devx.tools.check_api_identity_checks
# Set up SSH private key from SSH_PRIVATE_KEY env var
devx-setup-ssh-key:
@$(DEVX_PYTHON) -m devx.tools.setup_ssh_key
# ── Pre-push validation ───────────────────────────────────────────────────────
# Run lint + tests before push (projects can override with project-specific targets)
+1 -1
View File
@@ -16,7 +16,7 @@ Outputs:
- (default): prints both as ``count=N`` and ``indices=[0,1,...]``
Usage:
python3 -m devx.molecule.discover_runners --owner oblachno-oss --repo grm
python3 -m devx.molecule.discover_runners --owner my-org --repo my-repo
python3 -m devx.molecule.discover_runners --indices
python3 -m devx.molecule.discover_runners --count
"""
+1 -1
View File
@@ -137,7 +137,7 @@ def build_multi_role_pairs(
# --- Molecule weight configuration ---
#
# Weights are loaded from ``[tool.devx.molecule.weights]`` in
# ``pyproject.toml``. Each project (infra, grm, …) contributes its own
# ``pyproject.toml``. Each project contributes its own
# weights calibrated from actual CI execution times.
#
# Two key formats are supported:
+2 -2
View File
@@ -15,9 +15,9 @@ exits early with code 1.
Usage::
# Single-role (grm-style)
# Single-role
python3 -m devx.molecule.molecule_ci_guard pair1 pair2 ...
# Multi-role (infra-style)
# Multi-role
python3 -m devx.molecule.molecule_ci_guard --roles-root ansible/roles pair1 pair2 ...
Environment variables:
+137
View File
@@ -0,0 +1,137 @@
#!/usr/bin/env python3
"""Scan integration tests for unsafe ``is True``/``is False`` identity checks.
Many APIs (e.g. Mattermost) return boolean values as strings (``"true"``,
``"false"``) rather than native JSON booleans. Using ``is True`` or
``is not False`` on such responses silently fails because ``"true" is True``
evaluates to ``False`` in Python.
This tool scans ``tests/integration/test_*.py`` files for identity checks
on API response values and reports them as errors.
Configuration (``[tool.devx.check_api_identity_checks]`` in pyproject.toml):
``scan_dirs`` list of directories to scan (default: ``["tests/integration"]``)
``skip_patterns`` list of filename patterns to skip (default: ``["test_*_helpers.py"]``)
``noqa_marker`` comment to suppress individual lines (default: ``# noqa``)
Usage::
python3 -m devx.tools.check_api_identity_checks
python3 -m devx.tools.check_api_identity_checks --scan-dir tests/integration
"""
from __future__ import annotations
import re
from pathlib import Path
import click
from devx.config import _load_pyproject_devx
from devx.i18n import _
DEFAULT_SCAN_DIRS = ["tests/integration"]
DEFAULT_SKIP_PATTERNS = ["test_*_helpers.py"]
DEFAULT_NOQA_MARKER = "# noqa"
# Matches: x is True, x is False, x is not True, x is not False
_IDENTITY_CHECK_RE = re.compile(r"\bis\s+(not\s+)?(True|False)\b")
def _load_config() -> tuple[list[str], list[str], str]:
"""Load configuration from pyproject.toml [tool.devx.check_api_identity_checks]."""
devx_cfg = _load_pyproject_devx()
cfg_raw = devx_cfg.get("check_api_identity_checks", {})
if not isinstance(cfg_raw, dict):
return DEFAULT_SCAN_DIRS, DEFAULT_SKIP_PATTERNS, DEFAULT_NOQA_MARKER
cfg: dict[str, object] = cfg_raw # type: ignore[assignment]
scan_dirs_raw = cfg.get("scan_dirs", DEFAULT_SCAN_DIRS)
scan_dirs: list[str] = [str(d) for d in scan_dirs_raw] if isinstance(scan_dirs_raw, list) else DEFAULT_SCAN_DIRS
skip_raw = cfg.get("skip_patterns", DEFAULT_SKIP_PATTERNS)
skip_patterns: list[str] = [str(p) for p in skip_raw] if isinstance(skip_raw, list) else DEFAULT_SKIP_PATTERNS
noqa_marker = str(cfg.get("noqa_marker", DEFAULT_NOQA_MARKER))
return scan_dirs, skip_patterns, noqa_marker
def _matches_skip_pattern(path: Path, skip_patterns: list[str]) -> bool:
"""Check if a file path matches any skip pattern."""
name = path.name
return any(Path(name).match(pattern) for pattern in skip_patterns)
def find_identity_checks(
file_path: Path,
repo_root: Path,
noqa_marker: str,
) -> list[str]:
"""Return a list of issue strings for unsafe identity checks in *file_path*."""
issues: list[str] = []
try:
source = file_path.read_text(encoding="utf-8")
except (OSError, UnicodeDecodeError):
return issues
rel = str(file_path.relative_to(repo_root))
for lineno, line in enumerate(source.splitlines(), 1):
if noqa_marker in line:
continue
match = _IDENTITY_CHECK_RE.search(line)
if match:
issues.append(
f"{rel}:{lineno}: unsafe identity check '{match.group()}' "
f"— APIs may return string 'true'/'false'. "
f"Use string comparison or _is_truthy()/_is_falsy() helpers."
)
return issues
@click.command()
@click.option(
"--scan-dir",
multiple=True,
help=_("Directory to scan (default: tests/integration). Can be repeated."),
)
def cli(scan_dir: tuple[str, ...]) -> None:
"""Scan integration tests for unsafe ``is True``/``is False`` identity checks."""
repo_root = Path.cwd()
config_scan_dirs, skip_patterns, noqa_marker = _load_config()
scan_dirs = list(scan_dir) if scan_dir else config_scan_dirs
all_issues: list[str] = []
for scan_dir_name in scan_dirs:
scan_path = repo_root / scan_dir_name
if not scan_path.exists():
continue
for py_file in scan_path.rglob("test_*.py"):
if _matches_skip_pattern(py_file, skip_patterns):
continue
all_issues.extend(find_identity_checks(py_file, repo_root, noqa_marker))
if all_issues:
click.echo(
_("Found {count} unsafe identity check(s) in integration tests.", count=len(all_issues)),
err=True,
)
for issue in all_issues:
click.echo(f" {issue}", err=True)
raise click.ClickException(
_(
"Use string comparison or _is_truthy()/_is_falsy() helpers instead. "
"Add '{marker}' to suppress individual lines.",
marker=noqa_marker,
)
)
click.echo(_("[check-api-identity-checks] Passed: no unsafe identity checks found"))
if __name__ == "__main__": # pragma: no cover
cli() # pragma: no cover
+5 -5
View File
@@ -80,7 +80,7 @@ def _check_python_version(venv_bin: Path) -> None:
@click.option("--checkmake-bin", default=None, help="Path to checkmake binary (fallback if not on PATH).")
def cli(venv: str, checkmake_bin: str | None) -> None:
"""Verify that required development tools are present."""
click.echo("[check-deps] Verifying tools...")
click.echo(_("[check-deps] Verifying tools..."))
all_required = True
for tool in REQUIRED_TOOLS:
@@ -90,15 +90,15 @@ def cli(venv: str, checkmake_bin: str | None) -> None:
for tool in OPTIONAL_TOOLS:
if not _check_tool(tool, optional=True):
if checkmake_bin and Path(checkmake_bin).exists():
click.echo(f" {tool}: found at {checkmake_bin}")
click.echo(_(" {tool}: found at {path}", tool=tool, path=checkmake_bin))
else:
click.echo(" Run 'make install-checkmake' to install the Makefile linter.")
click.echo(_(" Run 'make install-checkmake' to install the Makefile linter."))
_check_python_version(Path(venv) / "bin")
if not all_required:
raise click.ClickException("Required tools missing.")
click.echo("[check-deps] All core tools present.")
raise click.ClickException(_("Required tools missing."))
click.echo(_("[check-deps] All core tools present."))
if __name__ == "__main__": # pragma: no cover
+4 -1
View File
@@ -32,7 +32,10 @@ _TIMING_RE = re.compile(r"(\d+) passed.* in ([0-9.]+)s")
# Matches per-test duration lines from --durations=0:
# 0.51s call tests/test_foo.py::test_bar
_DURATION_LINE_RE = re.compile(r"^(\d+\.?\d*)s\s+(?:setup|call|teardown)\s+(.+)$")
# Only "call" duration is counted — "setup" includes import/collection
# overhead (coverage init, module imports) which is environment-dependent
# and not a test quality signal.
_DURATION_LINE_RE = re.compile(r"^(\d+\.?\d*)s\s+call\s+(.+)$")
def run_tests() -> tuple[str, str]:
+1 -1
View File
@@ -182,7 +182,7 @@ def main(repo: str | None, owner: str | None, branch: str, api_url: str | None)
if not repo:
raise click.ClickException(_("ERROR: Repository name not specified. Use --repo or set DEVX_REPO_NAME."))
# If DEVX_REPO_NAME contains a slash (e.g. "oblachno/infra"), split into owner/repo.
# If DEVX_REPO_NAME contains a slash (e.g. "my-org/my-repo"), split into owner/repo.
# This prevents 404s when workflows set DEVX_REPO_NAME to the full path.
if "/" in repo and owner is None:
parts = repo.split("/", 1)
+2 -2
View File
@@ -68,8 +68,8 @@ def detect_package_name(repo_root: Path) -> str | None:
Looks for the first subdirectory under ``src/`` that contains
an ``__init__.py`` file with ``__version__``.
Returns the package directory name (e.g., ``devx``,
``gitea_runner_manager``) or ``None`` if no package is found.
Returns the package directory name (e.g., ``devx``) or ``None`` if
no package is found.
"""
src_dir = repo_root / "src"
if not src_dir.is_dir():
+90
View File
@@ -0,0 +1,90 @@
#!/usr/bin/env python3
"""Set up SSH private key for CI jobs that need SSH access to remote hosts.
Writes the ``SSH_PRIVATE_KEY`` env var to ``~/.ssh/id_rsa``, starts
``ssh-agent``, and adds the key. Replaces the repeated inline shell
pattern in CI workflow files.
Usage::
python3 -m devx.tools.setup_ssh_key
Reads ``SSH_PRIVATE_KEY`` from the environment. Exits 0 on success,
1 on missing key.
"""
from __future__ import annotations
import os
import subprocess # nosec B404
import sys
from pathlib import Path
import click
from devx.i18n import _
def setup_ssh_key(private_key: str | None = None) -> bool:
"""Set up SSH private key and start ssh-agent.
Args:
private_key: The SSH private key content. If None, reads from
``SSH_PRIVATE_KEY`` environment variable.
Returns:
True if setup succeeded, False if key is missing.
"""
key = private_key or os.environ.get("SSH_PRIVATE_KEY", "")
if not key:
click.echo(_("SSH_PRIVATE_KEY not set — skipping SSH key setup"), err=True)
return False
ssh_dir = Path.home() / ".ssh"
ssh_dir.mkdir(parents=True, exist_ok=True)
key_path = ssh_dir / "id_rsa"
key_path.write_text(f"{key}\n", encoding="utf-8")
key_path.chmod(0o600)
# Start ssh-agent and add the key
agent_result = subprocess.run( # nosec B603, B607
["ssh-agent", "-s"],
capture_output=True,
text=True,
check=False,
)
if agent_result.returncode != 0:
click.echo(_("Failed to start ssh-agent: {error}", error=agent_result.stderr), err=True)
return False
# Parse ssh-agent output to set env vars
for raw_line in agent_result.stdout.splitlines():
stripped = raw_line.strip()
if "=" in stripped and ";" in stripped:
var, val = stripped.split("=", 1)
val = val.rstrip(";")
os.environ[var] = val
# Add the key (non-fatal if it fails — key may already be loaded)
subprocess.run( # nosec B603, B607
["ssh-add", str(key_path)],
capture_output=True,
text=True,
check=False,
)
return True
@click.command()
def cli() -> None:
"""Set up SSH private key from SSH_PRIVATE_KEY env var."""
if setup_ssh_key():
click.echo(_("SSH key set up successfully"))
sys.exit(0)
click.echo(_("SSH key setup skipped (no key provided)"), err=True)
sys.exit(1)
if __name__ == "__main__": # pragma: no cover
cli() # pragma: no cover
+5 -5
View File
@@ -33,7 +33,7 @@ DEFAULT_VALIDATE_DIRS = [
def _run_tofu(cmd: list[str], cwd: Path) -> None:
"""Run a tofu command in the given directory, raising on failure."""
click.echo(f" -> {cwd}")
click.echo(_(" -> {dir}", dir=cwd))
result = subprocess.run( # nosec B603, B607
cmd,
cwd=str(cwd),
@@ -60,9 +60,9 @@ def tofu_init(env: str, root: str = ".", extra_dirs: list[str] | None = None) ->
for dir_pattern in dirs:
dir_path = root_path / dir_pattern
if dir_path.is_dir():
click.echo(f"[tofu-init] Initializing {dir_path}...")
click.echo(_("[tofu-init] Initializing {dir}...", dir=dir_path))
_run_tofu(["tofu", "init"], dir_path)
click.echo("[tofu-init] Done.")
click.echo(_("[tofu-init] Done."))
def tofu_validate(
@@ -83,7 +83,7 @@ def tofu_validate(
root_path = Path(root)
target_dirs = dirs or DEFAULT_VALIDATE_DIRS
mode = "ci" if ci else "validate"
click.echo(f"[tofu-{mode}] Validating OpenTofu configurations...")
click.echo(_("[tofu-{mode}] Validating OpenTofu configurations...", mode=mode))
for dir_rel in target_dirs:
dir_path = root_path / dir_rel
if not dir_path.is_dir():
@@ -91,7 +91,7 @@ def tofu_validate(
if ci:
_run_tofu(["tofu", "init", "-backend=false", "-input=false"], dir_path)
_run_tofu(["tofu", "validate"], dir_path)
click.echo(f"[tofu-{mode}] All configurations valid.")
click.echo(_("[tofu-{mode}] All configurations valid.", mode=mode))
@click.group()
+160
View File
@@ -3198,5 +3198,165 @@
"pl": "OSTRZEŻENIE: .venv ma Python {version}, ale wymagane jest >={req}.",
"ru": "ПРЕДУПРЕЖДЕНИЕ: в .venv установлен Python {version}, но требуется >={req}.",
"zh": "警告: .venv 的 Python 版本为 {version},但要求 >={req}。"
},
" -> {dir}": {
"en": " -> {dir}",
"bg": " -> {dir}",
"de": " -> {dir}",
"pl": " -> {dir}",
"ru": " -> {dir}",
"zh": " -> {dir}"
},
"[tofu-init] Initializing {dir}...": {
"en": "[tofu-init] Initializing {dir}...",
"bg": "[tofu-init] Инициализиране на {dir}...",
"de": "[tofu-init] Initialisiere {dir}...",
"pl": "[tofu-init] Inicjalizacja {dir}...",
"ru": "[tofu-init] Инициализация {dir}...",
"zh": "[tofu-init] 正在初始化 {dir}..."
},
"[tofu-init] Done.": {
"en": "[tofu-init] Done.",
"bg": "[tofu-init] Готово.",
"de": "[tofu-init] Fertig.",
"pl": "[tofu-init] Gotowe.",
"ru": "[tofu-init] Готово.",
"zh": "[tofu-init] 完成。"
},
"[tofu-{mode}] Validating OpenTofu configurations...": {
"en": "[tofu-{mode}] Validating OpenTofu configurations...",
"bg": "[tofu-{mode}] Проверка на OpenTofu конфигурациите...",
"de": "[tofu-{mode}] Validiere OpenTofu-Konfigurationen...",
"pl": "[tofu-{mode}] Sprawdzanie konfiguracji OpenTofu...",
"ru": "[tofu-{mode}] Проверка конфигураций OpenTofu...",
"zh": "[tofu-{mode}] 正在验证 OpenTofu 配置..."
},
"[tofu-{mode}] All configurations valid.": {
"en": "[tofu-{mode}] All configurations valid.",
"bg": "[tofu-{mode}] Всички конфигурации са валидни.",
"de": "[tofu-{mode}] Alle Konfigurationen gültig.",
"pl": "[tofu-{mode}] Wszystkie konfiguracje są poprawne.",
"ru": "[tofu-{mode}] Все конфигурации валидны.",
"zh": "[tofu-{mode}] 所有配置有效。"
},
"[check-deps] Verifying tools...": {
"en": "[check-deps] Verifying tools...",
"bg": "[check-deps] Проверка на инструментите...",
"de": "[check-deps] Werkzeuge werden überprüft...",
"pl": "[check-deps] Sprawdzanie narzędzi...",
"ru": "[check-deps] Проверка инструментов...",
"zh": "[check-deps] 正在验证工具..."
},
" {tool}: found at {path}": {
"en": " {tool}: found at {path}",
"bg": " {tool}: намерен на {path}",
"de": " {tool}: gefunden unter {path}",
"pl": " {tool}: znaleziono w {path}",
"ru": " {tool}: найден в {path}",
"zh": " {tool}: 在 {path} 找到"
},
" Run 'make install-checkmake' to install the Makefile linter.": {
"en": " Run 'make install-checkmake' to install the Makefile linter.",
"bg": " Изпълнете 'make install-checkmake' за инсталиране на Makefile линтера.",
"de": " Führen Sie 'make install-checkmake' aus, um den Makefile-Linter zu installieren.",
"pl": " Uruchom 'make install-checkmake', aby zainstalować linter Makefile.",
"ru": " Выполните 'make install-checkmake' для установки линтера Makefile.",
"zh": " 运行 'make install-checkmake' 来安装 Makefile 检查器。"
},
"Required tools missing.": {
"en": "Required tools missing.",
"bg": "Липсват задължителни инструменти.",
"de": "Erforderliche Werkzeuge fehlen.",
"pl": "Brak wymaganych narzędzi.",
"ru": "Отсутствуют обязательные инструменты.",
"zh": "缺少必需的工具。"
},
"[check-deps] All core tools present.": {
"en": "[check-deps] All core tools present.",
"bg": "[check-deps] Всички основни инструменти са налични.",
"de": "[check-deps] Alle Kernwerkzeuge vorhanden.",
"pl": "[check-deps] Wszystkie podstawowe narzędzia są dostępne.",
"ru": "[check-deps] Все основные инструменты доступны.",
"zh": "[check-deps] 所有核心工具均已就绪。"
},
"SSH_PRIVATE_KEY not set — skipping SSH key setup": {
"en": "SSH_PRIVATE_KEY not set — skipping SSH key setup",
"bg": "SSH_PRIVATE_KEY не е зададен — пропускане на SSH ключ настройката",
"de": "SSH_PRIVATE_KEY nicht gesetzt — SSH-Schlüssel-Setup übersprungen",
"pl": "SSH_PRIVATE_KEY nie ustawione — pomijanie konfiguracji klucza SSH",
"ru": "SSH_PRIVATE_KEY не задан — пропуск настройки SSH-ключа",
"zh": "SSH_PRIVATE_KEY 未设置 — 跳过 SSH 密钥设置"
},
"Failed to start ssh-agent: {error}": {
"en": "Failed to start ssh-agent: {error}",
"bg": "Неуспешно стартиране на ssh-agent: {error}",
"de": "Starten von ssh-agent fehlgeschlagen: {error}",
"pl": "Nie udało się uruchomić ssh-agent: {error}",
"ru": "Не удалось запустить ssh-agent: {error}",
"zh": "启动 ssh-agent 失败: {error}"
},
"SSH key set up successfully": {
"en": "SSH key set up successfully",
"bg": "SSH ключът е настроен успешно",
"de": "SSH-Schlüssel erfolgreich eingerichtet",
"pl": "Klucz SSH skonfigurowany pomyślnie",
"ru": "SSH-ключ успешно настроен",
"zh": "SSH 密钥设置成功"
},
"SSH key setup skipped (no key provided)": {
"en": "SSH key setup skipped (no key provided)",
"bg": "Настройката на SSH ключ е пропусната (не е предоставен ключ)",
"de": "SSH-Schlüssel-Setup übersprungen (kein Schlüssel bereitgestellt)",
"pl": "Pominięto konfigurację klucza SSH (brak klucza)",
"ru": "Настройка SSH-ключа пропущена (ключ не предоставлен)",
"zh": "SSH 密钥设置已跳过(未提供密钥)"
},
"Found {count} unsafe identity check(s) in integration tests.": {
"en": "Found {count} unsafe identity check(s) in integration tests.",
"bg": "Намерени са {count} небрежни проверки за идентичност в интеграционните тестове.",
"de": "{count} unsichere Identitätsprüfung(en) in Integrationstests gefunden.",
"pl": "Znaleziono {count} niebezpiecznych sprawdzeń tożsamości w testach integracyjnych.",
"ru": "Найдено {count} небезопасных проверок идентичности в интеграционных тестах.",
"zh": "在集成测试中发现 {count} 个不安全的身份检查。"
},
"Use string comparison or _is_truthy()/_is_falsy() helpers instead. Add '{marker}' to suppress individual lines.": {
"en": "Use string comparison or _is_truthy()/_is_falsy() helpers instead. Add '{marker}' to suppress individual lines.",
"bg": "Използвайте сравнение на низове или _is_truthy()/_is_falsy() помощници. Добавете '{marker}' за потискане на отделни редове.",
"de": "Verwenden Sie String-Vergleich oder _is_truthy()/_is_falsy() Hilfsfunktionen. Fügen Sie '{marker}' hinzu, um einzelne Zeilen zu unterdrücken.",
"pl": "Użyj porównania ciągów lub pomocników _is_truthy()/_is_falsy(). Dodaj '{marker}', aby pominąć pojedyncze linie.",
"ru": "Используйте строковое сравнение или помощники _is_truthy()/_is_falsy(). Добавьте '{marker}' для подавления отдельных строк.",
"zh": "使用字符串比较或 _is_truthy()/_is_falsy() 辅助函数。添加 '{marker}' 以抑制个别行。"
},
"[check-api-identity-checks] Passed: no unsafe identity checks found": {
"en": "[check-api-identity-checks] Passed: no unsafe identity checks found",
"bg": "[check-api-identity-checks] Мина: не са намерени небрежни проверки за идентичност",
"de": "[check-api-identity-checks] Bestanden: keine unsicheren Identitätsprüfungen gefunden",
"pl": "[check-api-identity-checks] Passed: nie znaleziono niebezpiecznych sprawdzeń tożsamości",
"ru": "[check-api-identity-checks] Пройдено: небезопасных проверок идентичности не найдено",
"zh": "[check-api-identity-checks] 通过:未发现不安全的身份检查"
},
"Directory to scan (default: tests/integration). Can be repeated.": {
"en": "Directory to scan (default: tests/integration). Can be repeated.",
"bg": "Директория за сканиране (по подразбиране: tests/integration). Може да се повтаря.",
"de": "Zu scannendes Verzeichnis (Standard: tests/integration). Kann wiederholt werden.",
"pl": "Katalog do skanowania (domyślnie: tests/integration). Można powtarzać.",
"ru": "Директория для сканирования (по умолчанию: tests/integration). Можно повторять.",
"zh": "要扫描的目录(默认:tests/integration)。可重复。"
},
"Failed to list existing wiki pages after retries: {error}. Aborting to avoid creating duplicate pages.": {
"bg": "Неуспешно извличане на съществуващи wiki страници след повторни опити: {error}. Прекратяване, за да се избегне създаване на дублирани страници.",
"de": "Abrufen bestehender Wiki-Seiten nach Wiederholungen fehlgeschlagen: {error}. Abbruch, um doppelte Seiten zu vermeiden.",
"en": "Failed to list existing wiki pages after retries: {error}. Aborting to avoid creating duplicate pages.",
"pl": "Nie udało się wylistować istniejących stron wiki po ponownych próbach: {error}. Przerywanie, aby uniknąć tworzenia zduplikowanych stron.",
"ru": "Не удалось получить список существующих wiki-страниц после повторных попыток: {error}. Прерывание, чтобы избежать создания дубликатов страниц.",
"zh": "重试后列出现有 wiki 页面失败:{error}。正在中止以避免创建重复页面。"
},
" Page '{title}' already exists (stale list). Re-listing and updating...": {
"bg": " Страницата '{title}' вече съществува (остарял списък). Пресписване и обновяване...",
"de": " Seite '{title}' existiert bereits (veraltete Liste). Neu auflisten und aktualisieren...",
"en": " Page '{title}' already exists (stale list). Re-listing and updating...",
"pl": " Strona '{title}' już istnieje (nieaktualna lista). Ponowne listowanie i aktualizacja...",
"ru": " Страница '{title}' уже существует (устаревший список). Повторное получение списка и обновление...",
"zh": " 页面 '{title}' 已存在(列表过期)。重新列出并更新..."
}
}
+3
View File
@@ -0,0 +1,3 @@
"""Shared utility functions for devx and consumer projects."""
from __future__ import annotations
+51
View File
@@ -0,0 +1,51 @@
#!/usr/bin/env python3
"""Utilities for handling API response values.
Many APIs return boolean values as strings (``"true"``, ``"false"``)
rather than native JSON booleans. The Mattermost ``/api/v4/config/client``
endpoint is a notable example. These helpers handle both string and
boolean responses safely.
Usage::
from devx.utils.api import is_truthy, is_falsy
if not is_truthy(config.get("EnableOpenServer")):
raise ValueError("EnableOpenServer not enabled")
"""
from __future__ import annotations
def is_truthy(value: str | bool | None) -> bool:
"""Check if an API config value is truthy.
The API may return strings (``"true"``/``"false"``) or native
booleans. This helper handles both.
Args:
value: The value to check (string, bool, or None).
Returns:
True if the value represents a truthy boolean.
"""
if isinstance(value, bool):
return value
return str(value).lower() == "true"
def is_falsy(value: str | bool | None) -> bool:
"""Check if an API config value is falsy.
The API may return strings (``"true"``/``"false"``) or native
booleans. This helper handles both.
Args:
value: The value to check (string, bool, or None).
Returns:
True if the value represents a falsy boolean.
"""
if isinstance(value, bool):
return not value
return str(value).lower() == "false"
+54
View File
@@ -0,0 +1,54 @@
"""Unit tests for devx.utils.api."""
from __future__ import annotations
from devx.utils.api import is_falsy, is_truthy
class TestIsTruthy:
def test_string_true(self) -> None:
assert is_truthy("true") is True
def test_string_true_uppercase(self) -> None:
assert is_truthy("True") is True
def test_boolean_true(self) -> None:
assert is_truthy(True) is True
def test_string_false(self) -> None:
assert is_truthy("false") is False
def test_boolean_false(self) -> None:
assert is_truthy(False) is False
def test_none(self) -> None:
assert is_truthy(None) is False
def test_empty_string(self) -> None:
assert is_truthy("") is False
def test_random_string(self) -> None:
assert is_truthy("random") is False
class TestIsFalsy:
def test_string_false(self) -> None:
assert is_falsy("false") is True
def test_string_false_uppercase(self) -> None:
assert is_falsy("False") is True
def test_boolean_false(self) -> None:
assert is_falsy(False) is True
def test_string_true(self) -> None:
assert is_falsy("true") is False
def test_boolean_true(self) -> None:
assert is_falsy(True) is False
def test_none(self) -> None:
assert is_falsy(None) is False
def test_empty_string(self) -> None:
assert is_falsy("") is False
@@ -0,0 +1,176 @@
"""Unit tests for devx.tools.check_api_identity_checks."""
from __future__ import annotations
from pathlib import Path
from unittest.mock import patch
from click.testing import CliRunner
from devx.tools.check_api_identity_checks import (
DEFAULT_NOQA_MARKER,
DEFAULT_SCAN_DIRS,
DEFAULT_SKIP_PATTERNS,
_load_config,
_matches_skip_pattern,
cli,
find_identity_checks,
)
class TestFindIdentityChecks:
def test_detects_is_true(self, tmp_path: Path) -> None:
f = tmp_path / "test_foo.py"
f.write_text("assert config.get('x') is True\n")
issues = find_identity_checks(f, tmp_path, DEFAULT_NOQA_MARKER)
assert len(issues) == 1
assert "is True" in issues[0]
def test_detects_is_false(self, tmp_path: Path) -> None:
f = tmp_path / "test_foo.py"
f.write_text("if config.get('x') is False:\n pass\n")
issues = find_identity_checks(f, tmp_path, DEFAULT_NOQA_MARKER)
assert len(issues) == 1
assert "is False" in issues[0]
def test_detects_is_not_true(self, tmp_path: Path) -> None:
f = tmp_path / "test_foo.py"
f.write_text("if config.get('x') is not True:\n fail()\n")
issues = find_identity_checks(f, tmp_path, DEFAULT_NOQA_MARKER)
assert len(issues) == 1
assert "is not True" in issues[0]
def test_detects_is_not_false(self, tmp_path: Path) -> None:
f = tmp_path / "test_foo.py"
f.write_text("if config.get('x') is not False:\n fail()\n")
issues = find_identity_checks(f, tmp_path, DEFAULT_NOQA_MARKER)
assert len(issues) == 1
assert "is not False" in issues[0]
def test_noqa_suppresses(self, tmp_path: Path) -> None:
f = tmp_path / "test_foo.py"
f.write_text("assert config.get('x') is True # noqa\n")
issues = find_identity_checks(f, tmp_path, DEFAULT_NOQA_MARKER)
assert len(issues) == 0
def test_no_false_positives(self, tmp_path: Path) -> None:
f = tmp_path / "test_foo.py"
f.write_text("assert config.get('x') == 'true'\nassert config.get('y') == True\nx = True\nif x:\n pass\n")
issues = find_identity_checks(f, tmp_path, DEFAULT_NOQA_MARKER)
assert len(issues) == 0
def test_multiple_issues(self, tmp_path: Path) -> None:
f = tmp_path / "test_foo.py"
f.write_text("if config.get('a') is True:\n pass\nif config.get('b') is not False:\n pass\n")
issues = find_identity_checks(f, tmp_path, DEFAULT_NOQA_MARKER)
assert len(issues) == 2
def test_file_not_found(self, tmp_path: Path) -> None:
f = tmp_path / "nonexistent.py"
issues = find_identity_checks(f, tmp_path, DEFAULT_NOQA_MARKER)
assert issues == []
class TestMatchesSkipPattern:
def test_matches_helpers(self) -> None:
assert _matches_skip_pattern(Path("test_mattermost_helpers.py"), DEFAULT_SKIP_PATTERNS)
def test_does_not_match_regular(self) -> None:
assert not _matches_skip_pattern(Path("test_mattermost.py"), DEFAULT_SKIP_PATTERNS)
def test_empty_patterns(self) -> None:
assert not _matches_skip_pattern(Path("test_anything.py"), [])
class TestLoadConfig:
def test_defaults(self) -> None:
with patch("devx.tools.check_api_identity_checks._load_pyproject_devx") as mock:
mock.return_value = {}
scan_dirs, skip_patterns, noqa = _load_config()
assert scan_dirs == DEFAULT_SCAN_DIRS
assert skip_patterns == DEFAULT_SKIP_PATTERNS
assert noqa == DEFAULT_NOQA_MARKER
def test_custom_config(self) -> None:
with patch("devx.tools.check_api_identity_checks._load_pyproject_devx") as mock:
mock.return_value = {
"check_api_identity_checks": {
"scan_dirs": ["tests/api"],
"skip_patterns": ["test_*_unit.py"],
"noqa_marker": "# allow",
}
}
scan_dirs, skip_patterns, noqa = _load_config()
assert scan_dirs == ["tests/api"]
assert skip_patterns == ["test_*_unit.py"]
assert noqa == "# allow"
def test_invalid_config_returns_defaults(self) -> None:
with patch("devx.tools.check_api_identity_checks._load_pyproject_devx") as mock:
mock.return_value = {"check_api_identity_checks": "not a dict"}
scan_dirs, _, _ = _load_config()
assert scan_dirs == DEFAULT_SCAN_DIRS
class TestCli:
def test_no_issues(self, tmp_path: Path) -> None:
runner = CliRunner()
with (
patch("devx.tools.check_api_identity_checks._load_config") as mock_cfg,
patch("devx.tools.check_api_identity_checks.Path.cwd", return_value=tmp_path),
):
mock_cfg.return_value = (["tests/integration"], DEFAULT_SKIP_PATTERNS, DEFAULT_NOQA_MARKER)
(tmp_path / "tests" / "integration").mkdir(parents=True)
(tmp_path / "tests" / "integration" / "test_foo.py").write_text("assert config.get('x') == 'true'\n")
result = runner.invoke(cli, [])
assert result.exit_code == 0
assert "Passed" in result.output
def test_with_issues(self, tmp_path: Path) -> None:
runner = CliRunner()
with (
patch("devx.tools.check_api_identity_checks._load_config") as mock_cfg,
patch("devx.tools.check_api_identity_checks.Path.cwd", return_value=tmp_path),
):
mock_cfg.return_value = (["tests/integration"], DEFAULT_SKIP_PATTERNS, DEFAULT_NOQA_MARKER)
(tmp_path / "tests" / "integration").mkdir(parents=True)
(tmp_path / "tests" / "integration" / "test_foo.py").write_text(
"if config.get('x') is not True:\n fail()\n"
)
result = runner.invoke(cli, [])
assert result.exit_code != 0
assert "is not True" in result.output
def test_skips_helpers(self, tmp_path: Path) -> None:
runner = CliRunner()
with (
patch("devx.tools.check_api_identity_checks._load_config") as mock_cfg,
patch("devx.tools.check_api_identity_checks.Path.cwd", return_value=tmp_path),
):
mock_cfg.return_value = (["tests/integration"], DEFAULT_SKIP_PATTERNS, DEFAULT_NOQA_MARKER)
(tmp_path / "tests" / "integration").mkdir(parents=True)
(tmp_path / "tests" / "integration" / "test_foo_helpers.py").write_text("assert x is True\n")
result = runner.invoke(cli, [])
assert result.exit_code == 0
def test_nonexistent_dir(self, tmp_path: Path) -> None:
runner = CliRunner()
with (
patch("devx.tools.check_api_identity_checks._load_config") as mock_cfg,
patch("devx.tools.check_api_identity_checks.Path.cwd", return_value=tmp_path),
):
mock_cfg.return_value = (["nonexistent"], DEFAULT_SKIP_PATTERNS, DEFAULT_NOQA_MARKER)
result = runner.invoke(cli, [])
assert result.exit_code == 0
def test_custom_scan_dir(self, tmp_path: Path) -> None:
runner = CliRunner()
with (
patch("devx.tools.check_api_identity_checks._load_config") as mock_cfg,
patch("devx.tools.check_api_identity_checks.Path.cwd", return_value=tmp_path),
):
mock_cfg.return_value = (["other"], DEFAULT_SKIP_PATTERNS, DEFAULT_NOQA_MARKER)
(tmp_path / "custom").mkdir()
(tmp_path / "custom" / "test_foo.py").write_text("if x is True:\n pass\n")
result = runner.invoke(cli, ["--scan-dir", "custom"])
assert result.exit_code != 0
+5 -5
View File
@@ -69,16 +69,16 @@ class TestParsePerTestDurations:
assert len(durations) == 1
assert durations[0] == ("tests/test_foo.py::test_bar", 0.01)
def test_parses_setup_and_teardown(self) -> None:
def test_ignores_setup_and_teardown(self) -> None:
"""Only 'call' durations are counted — setup includes import overhead."""
output = (
"0.02s setup tests/test_foo.py::test_bar\n"
"0.68s setup tests/test_foo.py::test_bar\n"
"0.01s call tests/test_foo.py::test_bar\n"
"0.00s teardown tests/test_foo.py::test_bar\n"
)
durations = parse_per_test_durations(output)
assert len(durations) == 3
names = [d[0] for d in durations]
assert "tests/test_foo.py::test_bar" in names
assert len(durations) == 1
assert durations[0] == ("tests/test_foo.py::test_bar", 0.01)
def test_sorted_slowest_first(self) -> None:
output = "0.01s call tests/test_a.py::test_slow\n0.50s call tests/test_b.py::test_fast\n"
+9 -9
View File
@@ -185,7 +185,7 @@ class TestMain:
args = mock_client.ensure_branch_protection.call_args
assert args[0][0] == "develop"
@patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok", "DEVX_REPO_NAME": "oblachno/infra"}, clear=True)
@patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok", "DEVX_REPO_NAME": "my-org/my-repo"}, clear=True)
@patch("devx.tools.configure_repo.GiteaClient")
def test_main_parses_owner_repo_from_env(self, mock_client_cls: MagicMock) -> None:
"""DEVX_REPO_NAME with 'owner/repo' format should be split."""
@@ -197,15 +197,15 @@ class TestMain:
assert result.exit_code == 0
# Verify GiteaClient was constructed with parsed owner and repo (positional)
call_args = mock_client_cls.call_args
assert call_args[0][2] == "oblachno" # owner is 3rd positional arg
assert call_args[0][3] == "infra" # repo is 4th positional arg
assert call_args[0][2] == "my-org" # owner is 3rd positional arg
assert call_args[0][3] == "my-repo" # repo is 4th positional arg
@patch.dict(
"os.environ",
{"CI_GITEA_TOKEN": "tok", "DEVX_REPO_NAME": "infra", "DEVX_REPO_OWNER": "oblachno"},
{"CI_GITEA_TOKEN": "tok", "DEVX_REPO_NAME": "my-repo", "DEVX_REPO_OWNER": "my-org"},
clear=True,
)
@patch("devx.tools.configure_repo.REPO_OWNER", "oblachno")
@patch("devx.tools.configure_repo.REPO_OWNER", "my-org")
@patch("devx.tools.configure_repo.GiteaClient")
def test_main_no_slash_when_owner_set_separately(self, mock_client_cls: MagicMock) -> None:
"""When DEVX_REPO_OWNER is set, DEVX_REPO_NAME should not be split."""
@@ -216,12 +216,12 @@ class TestMain:
result = runner.invoke(main, [])
assert result.exit_code == 0
call_args = mock_client_cls.call_args
assert call_args[0][2] == "oblachno" # owner
assert call_args[0][3] == "infra" # repo
assert call_args[0][2] == "my-org" # owner
assert call_args[0][3] == "my-repo" # 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.REPO_OWNER", "my-org")
@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."""
@@ -232,5 +232,5 @@ class TestMain:
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][2] == "my-org" # owner
assert call_args[0][3] == "devx" # repo
+6 -6
View File
@@ -25,14 +25,14 @@ class TestExtractTaskId:
class TestGetRepoName:
@patch.dict("os.environ", {"DEVX_REPO_NAME": "infra"})
@patch.dict("os.environ", {"DEVX_REPO_NAME": "my-repo"})
def test_from_env(self) -> None:
assert get_repo_name() == "infra"
assert get_repo_name() == "my-repo"
@patch("devx.tools.create_pr.REPO_NAME", "devx")
@patch.dict("os.environ", {"GITHUB_REPOSITORY": "oblachno/infra"}, clear=True)
@patch.dict("os.environ", {"GITHUB_REPOSITORY": "my-org/my-repo"}, clear=True)
def test_env_overrides_pyproject(self) -> None:
assert get_repo_name() == "infra"
assert get_repo_name() == "my-repo"
@patch("devx.tools.create_pr.REPO_NAME", "devx")
@patch.dict("os.environ", {}, clear=True)
@@ -40,9 +40,9 @@ class TestGetRepoName:
assert get_repo_name() == "devx"
@patch("devx.tools.create_pr.REPO_NAME", "")
@patch.dict("os.environ", {"GITHUB_REPOSITORY": "oblachno/infra"}, clear=True)
@patch.dict("os.environ", {"GITHUB_REPOSITORY": "my-org/my-repo"}, clear=True)
def test_from_github(self) -> None:
assert get_repo_name() == "infra"
assert get_repo_name() == "my-repo"
@patch("devx.tools.create_pr.REPO_NAME", "")
@patch.dict("os.environ", {}, clear=True)
+1 -1
View File
@@ -592,7 +592,7 @@ class TestLptDistribute:
def test_load_balance_with_varying_weights(self) -> None:
"""LPT should produce better load balance than round-robin."""
items = list(range(7))
# Simulate infra-like weights: 2 heavy, 2 medium, 3 light
# Simulate multi-role-like weights: 2 heavy, 2 medium, 3 light
weights = [10, 10, 7, 7, 3, 3, 3]
groups = _lpt_distribute(items, weights, 3)
loads = [sum(weights[i] for i in g) for g in groups]
+3 -3
View File
@@ -105,7 +105,7 @@ class TestCli:
"RUN_ID": "123",
"JOB_NAME": "integration-tests",
"MATRIX_INDEX": "0",
"GITEA_REPOSITORY": "oblachno-oss/infra",
"GITEA_REPOSITORY": "my-org/my-repo",
"PATH": os.environ.get("PATH", ""),
},
clear=True,
@@ -152,7 +152,7 @@ class TestCli:
"RUN_ID": "123",
"JOB_NAME": "integration-tests",
"MATRIX_INDEX": "0",
"GITEA_REPOSITORY": "oblachno-oss/infra",
"GITEA_REPOSITORY": "my-org/my-repo",
"PATH": os.environ.get("PATH", ""),
},
clear=True,
@@ -197,7 +197,7 @@ class TestCli:
"RUN_ID": "123",
"JOB_NAME": "integration-tests",
"MATRIX_INDEX": "0",
"GITEA_REPOSITORY": "oblachno-oss/infra",
"GITEA_REPOSITORY": "my-org/my-repo",
"PATH": os.environ.get("PATH", ""),
},
clear=True,
+4 -4
View File
@@ -224,7 +224,7 @@ class TestCli:
"RUN_ID": "123",
"JOB_NAME": "molecule-tests",
"MATRIX_INDEX": "0",
"GITEA_REPOSITORY": "oblachno-oss/grm",
"GITEA_REPOSITORY": "my-org/my-repo",
"PATH": os.environ.get("PATH", ""),
},
clear=True,
@@ -292,7 +292,7 @@ class TestCli:
"RUN_ID": "123",
"JOB_NAME": "molecule-tests",
"MATRIX_INDEX": "0",
"GITEA_REPOSITORY": "oblachno-oss/grm",
"GITEA_REPOSITORY": "my-org/my-repo",
"PATH": os.environ.get("PATH", ""),
},
clear=True,
@@ -375,7 +375,7 @@ class TestCli:
"RUN_ID": "123",
"JOB_NAME": "molecule-tests",
"MATRIX_INDEX": "0",
"GITEA_REPOSITORY": "oblachno-oss/grm",
"GITEA_REPOSITORY": "my-org/my-repo",
"PATH": os.environ.get("PATH", ""),
},
clear=True,
@@ -422,7 +422,7 @@ class TestCli:
"RUN_ID": "123",
"JOB_NAME": "molecule-tests",
"MATRIX_INDEX": "0",
"GITEA_REPOSITORY": "oblachno-oss/grm",
"GITEA_REPOSITORY": "my-org/my-repo",
"PATH": os.environ.get("PATH", ""),
},
clear=True,
+5 -5
View File
@@ -713,7 +713,7 @@ class TestMain:
def test_dry_run_does_not_post(self, mock_client_class: MagicMock, mock_run: MagicMock) -> None:
mock_run.return_value = ReviewResult()
runner = CliRunner()
result = runner.invoke(main, ["42", "oblachno-oss/grm", "--dry-run"], env={"CI_GITEA_TOKEN": "fake"})
result = runner.invoke(main, ["42", "my-org/my-repo", "--dry-run"], env={"CI_GITEA_TOKEN": "fake"})
assert result.exit_code == 0
assert "[dry-run]" in result.output
mock_client_class.return_value.create_review.assert_not_called()
@@ -724,7 +724,7 @@ class TestMain:
mock_run.return_value = ReviewResult()
mock_client_class.return_value.create_review.return_value = {"id": 123}
runner = CliRunner()
result = runner.invoke(main, ["42", "oblachno-oss/grm"], env={"CI_GITEA_TOKEN": "fake"})
result = runner.invoke(main, ["42", "my-org/my-repo"], env={"CI_GITEA_TOKEN": "fake"})
assert result.exit_code == 0
assert "Review #123" in result.output
mock_client_class.return_value.create_review.assert_called_once()
@@ -740,7 +740,7 @@ class TestMain:
{"id": 124},
]
runner = CliRunner()
result = runner.invoke(main, ["42", "oblachno-oss/grm"], env={"CI_GITEA_TOKEN": "fake"})
result = runner.invoke(main, ["42", "my-org/my-repo"], env={"CI_GITEA_TOKEN": "fake"})
assert result.exit_code == 0
assert "Review #124" in result.output
assert client.create_review.call_count == 2
@@ -753,12 +753,12 @@ class TestMain:
client = mock_client_class.return_value
client.create_review.side_effect = APIError(500, "Internal server error")
runner = CliRunner()
result = runner.invoke(main, ["42", "oblachno-oss/grm"], env={"CI_GITEA_TOKEN": "fake"})
result = runner.invoke(main, ["42", "my-org/my-repo"], env={"CI_GITEA_TOKEN": "fake"})
assert result.exit_code != 0
def test_no_token_raises(self) -> None:
runner = CliRunner()
result = runner.invoke(main, ["42", "oblachno-oss/grm"], env={"CI_GITEA_TOKEN": ""})
result = runner.invoke(main, ["42", "my-org/my-repo"], env={"CI_GITEA_TOKEN": ""})
assert result.exit_code != 0
assert "CI_GITEA_TOKEN" in result.output
+7 -7
View File
@@ -84,7 +84,7 @@ class TestPushToBadgesBranch:
class TestUpdateBadgeUrls:
def test_replaces_branch_url(self) -> None:
content = "[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/branch/badges/tests.svg)]"
content = "[![Tests](https://git.oblachno.oblachno.fyi/my-org/my-repo/raw/branch/badges/tests.svg)]"
result = push_badges.update_badge_urls(content, "abc123def456")
assert "raw/commit/abc123def456/tests.svg" in result
assert "raw/branch/badges" not in result
@@ -93,7 +93,7 @@ class TestUpdateBadgeUrls:
"""Old commit SHA URLs should be replaced with the new one."""
old_sha = "aabb123456789012345678901234567890123456" # 40 hex chars
new_sha = "ccdd123456789012345678901234567890123456" # 40 hex chars
content = f"[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/{old_sha}/tests.svg)]"
content = f"[![Tests](https://git.oblachno.oblachno.fyi/my-org/my-repo/raw/commit/{old_sha}/tests.svg)]"
result = push_badges.update_badge_urls(content, new_sha)
assert f"raw/commit/{new_sha}/tests.svg" in result
assert old_sha not in result
@@ -105,16 +105,16 @@ class TestUpdateBadgeUrls:
def test_multiple_badges(self) -> None:
content = (
"[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/branch/badges/coverage.svg)]\n"
"[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/branch/badges/tests.svg)]\n"
"[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/branch/badges/version.svg)]"
"[![Coverage](https://git.oblachno.oblachno.fyi/my-org/my-repo/raw/branch/badges/coverage.svg)]\n"
"[![Tests](https://git.oblachno.oblachno.fyi/my-org/my-repo/raw/branch/badges/tests.svg)]\n"
"[![Version](https://git.oblachno.oblachno.fyi/my-org/my-repo/raw/branch/badges/version.svg)]"
)
result = push_badges.update_badge_urls(content, "abc123def456")
assert result.count("raw/commit/abc123def456/") == 3
assert "raw/branch/badges" not in result
def test_preserves_non_badge_urls(self) -> None:
content = "[![CI](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/actions/workflows/ci.yml/badge.svg)]"
content = "[![CI](https://git.oblachno.oblachno.fyi/my-org/my-repo/actions/workflows/ci.yml/badge.svg)]"
result = push_badges.update_badge_urls(content, "abc123")
assert result == content
@@ -122,7 +122,7 @@ class TestUpdateBadgeUrls:
class TestUpdateReadmeWithBadgeSha:
def test_updates_readme(self, tmp_path: Path) -> None:
readme = tmp_path / "README.md"
readme.write_text("[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/branch/badges/tests.svg)]")
readme.write_text("[![Tests](https://git.oblachno.oblachno.fyi/my-org/my-repo/raw/branch/badges/tests.svg)]")
with patch("subprocess.run"):
push_badges.update_readme_with_badge_sha("abc123def456", repo_root=tmp_path)
content = readme.read_text()
+1 -1
View File
@@ -1147,7 +1147,7 @@ class TestMain:
mock_ft: MagicMock,
mock_vtc: MagicMock,
) -> None:
"""Release is skipped when only workflow/infra files changed."""
"""Release is skipped when only workflow/infrastructure files changed."""
mock_run_cmd.return_value = MagicMock(returncode=0, stdout="master\n", stderr="")
runner = CliRunner()
result = runner.invoke(main, [])
+153
View File
@@ -0,0 +1,153 @@
"""Unit tests for devx.tools.setup_ssh_key."""
from __future__ import annotations
import os
from unittest.mock import MagicMock, patch
from click.testing import CliRunner
from devx.tools.setup_ssh_key import cli, setup_ssh_key
class TestSetupSshKey:
def test_success(self, tmp_path, monkeypatch) -> None:
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("SSH_PRIVATE_KEY", "-----BEGIN KEY-----\nfake\n-----END KEY-----")
with (
patch("subprocess.run") as mock_run,
patch("pathlib.Path.chmod"),
):
agent_result = MagicMock()
agent_result.returncode = 0
agent_result.stdout = "SSH_AUTH_SOCK=/tmp/agent.sock;\nSSH_AGENT_PID=12345;\n"
agent_result.stderr = ""
add_result = MagicMock()
add_result.returncode = 0
add_result.stdout = ""
add_result.stderr = ""
mock_run.side_effect = [agent_result, add_result]
assert setup_ssh_key() is True
assert mock_run.call_count == 2
def test_missing_key(self, monkeypatch) -> None:
monkeypatch.delenv("SSH_PRIVATE_KEY", raising=False)
assert setup_ssh_key() is False
def test_empty_key(self, monkeypatch) -> None:
monkeypatch.setenv("SSH_PRIVATE_KEY", "")
assert setup_ssh_key() is False
def test_explicit_key_param(self, tmp_path, monkeypatch) -> None:
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.delenv("SSH_PRIVATE_KEY", raising=False)
with (
patch("subprocess.run") as mock_run,
patch("pathlib.Path.chmod"),
):
agent_result = MagicMock()
agent_result.returncode = 0
agent_result.stdout = "SSH_AUTH_SOCK=/tmp/agent.sock;\n"
agent_result.stderr = ""
add_result = MagicMock()
add_result.returncode = 0
add_result.stdout = ""
add_result.stderr = ""
mock_run.side_effect = [agent_result, add_result]
assert setup_ssh_key("-----BEGIN KEY-----\nfake\n-----END KEY-----") is True
def test_ssh_agent_failure(self, tmp_path, monkeypatch) -> None:
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("SSH_PRIVATE_KEY", "fake-key")
with (
patch("subprocess.run") as mock_run,
patch("pathlib.Path.chmod"),
):
agent_result = MagicMock()
agent_result.returncode = 1
agent_result.stdout = ""
agent_result.stderr = "ssh-agent failed"
mock_run.return_value = agent_result
assert setup_ssh_key() is False
def test_key_file_written(self, tmp_path, monkeypatch) -> None:
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("SSH_PRIVATE_KEY", "my-secret-key")
with (
patch("subprocess.run") as mock_run,
patch("pathlib.Path.chmod") as mock_chmod,
):
agent_result = MagicMock()
agent_result.returncode = 0
agent_result.stdout = "SSH_AUTH_SOCK=/tmp/agent.sock;\n"
agent_result.stderr = ""
add_result = MagicMock()
add_result.returncode = 0
add_result.stdout = ""
add_result.stderr = ""
mock_run.side_effect = [agent_result, add_result]
setup_ssh_key()
key_file = tmp_path / ".ssh" / "id_rsa"
assert key_file.exists()
assert "my-secret-key" in key_file.read_text()
mock_chmod.assert_called_with(0o600)
def test_env_vars_set_from_agent(self, tmp_path, monkeypatch) -> None:
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("SSH_PRIVATE_KEY", "fake-key")
with (
patch("subprocess.run") as mock_run,
patch("pathlib.Path.chmod"),
):
agent_result = MagicMock()
agent_result.returncode = 0
agent_result.stdout = "SSH_AUTH_SOCK=/tmp/agent.sock;\nSSH_AGENT_PID=999;\n"
agent_result.stderr = ""
add_result = MagicMock()
add_result.returncode = 0
add_result.stdout = ""
add_result.stderr = ""
mock_run.side_effect = [agent_result, add_result]
setup_ssh_key()
assert os.environ.get("SSH_AUTH_SOCK") == "/tmp/agent.sock"
assert os.environ.get("SSH_AGENT_PID") == "999"
def test_agent_output_without_env_vars(self, tmp_path, monkeypatch) -> None:
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("SSH_PRIVATE_KEY", "fake-key")
monkeypatch.delenv("SSH_AUTH_SOCK", raising=False)
with (
patch("subprocess.run") as mock_run,
patch("pathlib.Path.chmod"),
):
agent_result = MagicMock()
agent_result.returncode = 0
agent_result.stdout = "Agent started\nsome message without equals\n"
agent_result.stderr = ""
add_result = MagicMock()
add_result.returncode = 0
add_result.stdout = ""
add_result.stderr = ""
mock_run.side_effect = [agent_result, add_result]
assert setup_ssh_key() is True
assert os.environ.get("SSH_AUTH_SOCK") is None
class TestCli:
def test_success(self, tmp_path, monkeypatch) -> None:
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("SSH_PRIVATE_KEY", "fake-key")
runner = CliRunner()
with patch("devx.tools.setup_ssh_key.setup_ssh_key") as mock_setup:
mock_setup.return_value = True
result = runner.invoke(cli, [])
assert result.exit_code == 0
assert "successfully" in result.output
def test_no_key(self, monkeypatch) -> None:
monkeypatch.delenv("SSH_PRIVATE_KEY", raising=False)
runner = CliRunner()
with patch("devx.tools.setup_ssh_key.setup_ssh_key") as mock_setup:
mock_setup.return_value = False
result = runner.invoke(cli, [])
assert result.exit_code == 1
+40 -12
View File
@@ -171,6 +171,35 @@ class TestSyncPage:
assert "content" not in payload
assert base64.b64decode(payload["content_base64"]).decode("utf-8") == "# Updated"
def test_create_falls_back_to_update_on_already_exists(self) -> None:
"""When create fails with 400 'already exists', re-list and update."""
client = MagicMock()
# First call: POST /wiki/new → 400 already exists
# Second call: PATCH /wiki/page/{sub_url} → success
create_error = APIError(400, "wiki page already exists [title: Test-Page]")
client._request.side_effect = [create_error, MagicMock()]
with patch("devx.ci.sync_wiki._list_wiki_pages_with_retry", return_value={"Test-Page": "Test-Page.-"}):
result = sync_page(client, "Test-Page", "# Content", {}, dry_run=False)
assert result == "updated"
# Verify PATCH was called (second call)
patch_call = client._request.call_args_list[1]
assert patch_call.args[0] == "PATCH"
assert "/wiki/page/Test-Page.-" in patch_call.args[1]
def test_create_raises_non_400_error(self) -> None:
"""Non-400 errors from create should propagate, not trigger fallback."""
client = MagicMock()
client._request.side_effect = APIError(500, "server error")
with pytest.raises(APIError):
sync_page(client, "Test-Page", "# Content", {}, dry_run=False)
def test_create_raises_400_not_already_exists(self) -> None:
"""400 errors that don't mention 'already exists' should propagate."""
client = MagicMock()
client._request.side_effect = APIError(400, "invalid title")
with pytest.raises(APIError):
sync_page(client, "Test-Page", "# Content", {}, dry_run=False)
class TestVerifyWikiPage:
def test_verifies_matching_content(self) -> None:
@@ -537,20 +566,21 @@ class TestMain:
@patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True)
@patch("devx.ci.sync_wiki.GiteaClient")
def test_initial_list_api_error_treated_as_empty(self, mock_client_cls: MagicMock) -> None:
"""When the initial page list fails, sync proceeds treating wiki as empty."""
def test_initial_list_api_error_aborts(self, mock_client_cls: MagicMock) -> None:
"""When the initial page list fails after retries, sync aborts to avoid duplicate pages."""
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
with patch("devx.ci.sync_wiki.MAPPING_FILE") as mock_mapping:
mock_mapping.exists.return_value = True
with patch("devx.ci.sync_wiki.load_mapping", return_value={"index.md": "Home"}):
with patch("devx.ci.sync_wiki.read_doc_content", return_value="# Home"):
with patch("devx.ci.sync_wiki.list_wiki_pages", side_effect=APIError(0, "timeout")):
with patch("devx.ci.sync_wiki._list_wiki_pages_with_retry", side_effect=APIError(0, "timeout")):
with patch("devx.ci.sync_wiki.sync_page", return_value="created"):
runner = CliRunner()
result = runner.invoke(main, ["--repo", "owner/repo"])
assert result.exit_code == 0
assert "Created: Home" in result.output
assert result.exit_code != 0
assert "Failed to list existing wiki pages" in result.output
assert "Aborting" in result.output
@patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True)
@patch("devx.ci.sync_wiki.GiteaClient")
@@ -558,17 +588,15 @@ class TestMain:
"""When --verify re-fetch fails after retries, verification is skipped gracefully."""
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
# Initial list succeeds, but verify re-fetch fails
list_side_effect = [{"Home": "Home"}, APIError(0, "timeout")]
with patch("devx.ci.sync_wiki.MAPPING_FILE") as mock_mapping:
mock_mapping.exists.return_value = True
with patch("devx.ci.sync_wiki.load_mapping", return_value={"index.md": "Home"}):
with patch("devx.ci.sync_wiki.read_doc_content", return_value="# Home"):
with patch("devx.ci.sync_wiki.list_wiki_pages", return_value={"Home": "Home"}):
with patch("devx.ci.sync_wiki._list_wiki_pages_with_retry", side_effect=list_side_effect):
with patch("devx.ci.sync_wiki.sync_page", return_value="updated"):
with patch(
"devx.ci.sync_wiki._list_wiki_pages_with_retry",
side_effect=APIError(0, "timeout"),
):
runner = CliRunner()
result = runner.invoke(main, ["--repo", "owner/repo", "--verify"])
runner = CliRunner()
result = runner.invoke(main, ["--repo", "owner/repo", "--verify"])
assert result.exit_code == 0
assert "Skipping content verification" in result.output
+12 -12
View File
@@ -166,9 +166,9 @@ class TestCustomPrefix:
f.write(content)
return path
@patch.dict("os.environ", {"DEVX_TASK_PREFIX": "GRM"})
def test_master_accepts_grm_prefix(self) -> None:
"""Master branch accepts GRM-N: prefix when DEVX_TASK_PREFIX=GRM."""
@patch.dict("os.environ", {"DEVX_TASK_PREFIX": "PROJ"})
def test_master_accepts_proj_prefix(self) -> None:
"""Master branch accepts PROJ-N: prefix when DEVX_TASK_PREFIX=GRM."""
import importlib
import devx.ci.validate_commit_msg as vcm
@@ -177,7 +177,7 @@ class TestCustomPrefix:
importlib.reload(devx.config)
importlib.reload(vcm)
try:
msg_path = self._write_msg("GRM-66: fix: add scripts/** to infrastructure")
msg_path = self._write_msg("PROJ-66: fix: add scripts/** to infrastructure")
with patch("devx.ci.validate_commit_msg.get_branch", return_value="master"):
runner = CliRunner()
result = runner.invoke(vcm.main, [msg_path])
@@ -188,8 +188,8 @@ class TestCustomPrefix:
importlib.reload(devx.config)
importlib.reload(vcm)
@patch.dict("os.environ", {"DEVX_TASK_PREFIX": "GRM"})
def test_master_rejects_devx_prefix_when_grm_configured(self) -> None:
@patch.dict("os.environ", {"DEVX_TASK_PREFIX": "PROJ"})
def test_master_rejects_devx_prefix_when_proj_configured(self) -> None:
"""Master branch rejects DEVX-N: prefix when DEVX_TASK_PREFIX=GRM."""
import importlib
@@ -204,16 +204,16 @@ class TestCustomPrefix:
runner = CliRunner()
result = runner.invoke(vcm.main, [msg_path])
assert result.exit_code == 1
assert "GRM-N" in result.output
assert "PROJ-N" in result.output
os.unlink(msg_path)
finally:
os.environ.pop("DEVX_TASK_PREFIX", None)
importlib.reload(devx.config)
importlib.reload(vcm)
@patch.dict("os.environ", {"DEVX_TASK_PREFIX": "GRM"})
def test_feature_branch_rejects_grm_prefix(self) -> None:
"""Feature branch rejects GRM-N: prefix when DEVX_TASK_PREFIX=GRM."""
@patch.dict("os.environ", {"DEVX_TASK_PREFIX": "PROJ"})
def test_feature_branch_rejects_proj_prefix(self) -> None:
"""Feature branch rejects PROJ-N: prefix when DEVX_TASK_PREFIX=GRM."""
import importlib
import devx.ci.validate_commit_msg as vcm
@@ -222,8 +222,8 @@ class TestCustomPrefix:
importlib.reload(devx.config)
importlib.reload(vcm)
try:
msg_path = self._write_msg("GRM-66: fix: should not have prefix on branch")
with patch("devx.ci.validate_commit_msg.get_branch", return_value="GRM-66-fix"):
msg_path = self._write_msg("PROJ-66: fix: should not have prefix on branch")
with patch("devx.ci.validate_commit_msg.get_branch", return_value="PROJ-66-fix"):
runner = CliRunner()
result = runner.invoke(vcm.main, [msg_path])
assert result.exit_code == 1