Compare commits

...
16 Commits
Author SHA1 Message Date
grm-ci-bot a4d5ba6b70 release: v0.12.0 [skip ci] 2026-06-28 13:15:08 +00:00
emil d9ce4e240f GRM-113: feat: upgrade all dependencies, add trigger-workflow command
Post-merge / vikunja (push) Successful in 1m13s
Post-merge / configure-repo (push) Successful in 1m13s
Post-merge / badges (push) Successful in 1m33s
Post-merge / detect-type (push) Successful in 55s
Post-merge / release (push) Successful in 1m19s
Post-merge / validate-commit-msg (push) Successful in 1m19s
Post-merge / publish (push) Successful in 55s
Post-merge / sync-wiki (push) Successful in 1m54s
2026-06-28 13:13:06 +00:00
gitea-actions-bot c1f68f115a chore: update badge URLs to commit dfe10dfc [skip ci] 2026-06-28 11:41:38 +00:00
grm-ci-bot fdf1293c85 release: v0.11.1 [skip ci] 2026-06-28 11:41:20 +00:00
emil 01b3f594f7 GRM-112: fix: Makefile HOST/NAME requirement errors, add restart and list targets
Post-merge / detect-type (push) Successful in 48s
Post-merge / release (push) Successful in 1m0s
Post-merge / validate-commit-msg (push) Successful in 1m3s
Post-merge / vikunja (push) Successful in 1m3s
Post-merge / badges (push) Successful in 1m11s
Post-merge / sync-wiki (push) Successful in 1m35s
Post-merge / configure-repo (push) Successful in 58s
Post-merge / publish (push) Successful in 57s
2026-06-28 11:39:37 +00:00
gitea-actions-bot 9ffa7a3671 chore: update badge URLs to commit cc0e4b3f [skip ci] 2026-06-28 11:23:32 +00:00
grm-ci-bot f04be9c39b release: v0.11.0 [skip ci] 2026-06-28 11:23:13 +00:00
emil f67dff8458 GRM-111: feat: unified --become-password-file, --verbose, --no-status, labels fix
Post-merge / detect-type (push) Successful in 51s
Post-merge / release (push) Successful in 1m14s
Post-merge / validate-commit-msg (push) Successful in 1m14s
Post-merge / vikunja (push) Successful in 1m14s
Post-merge / badges (push) Successful in 1m25s
Post-merge / sync-wiki (push) Successful in 1m53s
Post-merge / publish (push) Successful in 1m21s
Post-merge / configure-repo (push) Successful in 1m20s
2026-06-28 11:21:11 +00:00
gitea-actions-bot 763f7640af chore: update badge URLs to commit b95f5ede [skip ci] 2026-06-28 02:15:53 +00:00
emil f8eeea611f GRM-110: chore: bump devx dependency to 0.26.0
Post-merge / detect-type (push) Successful in 53s
Post-merge / release (push) Successful in 1m12s
Post-merge / validate-commit-msg (push) Successful in 1m14s
Post-merge / publish (push) Has been skipped
Post-merge / vikunja (push) Successful in 1m22s
Post-merge / badges (push) Successful in 1m27s
Post-merge / configure-repo (push) Successful in 1m9s
Post-merge / sync-wiki (push) Successful in 1m51s
2026-06-28 02:13:28 +00:00
gitea-actions-bot 774bf479ab chore: update badge URLs to commit 63396d99 [skip ci] 2026-06-28 00:25:55 +00:00
emil 844171ee93 GRM-109: chore: bump devx to >=0.25.0, fix pr_review docs
Post-merge / detect-type (push) Successful in 1m0s
Post-merge / release (push) Successful in 59s
Post-merge / validate-commit-msg (push) Successful in 1m8s
Post-merge / sync-wiki (push) Successful in 1m45s
Post-merge / publish (push) Has been skipped
Post-merge / badges (push) Successful in 1m19s
Post-merge / vikunja (push) Successful in 1m8s
Post-merge / configure-repo (push) Successful in 1m7s
2026-06-28 00:22:45 +00:00
gitea-actions-bot 9a5be879a2 chore: update badge URLs to commit e7d92aa8 [skip ci] 2026-06-27 22:30:14 +00:00
emil f24ed4c963 GRM-108: ci: bump devx>=0.23.4 for classification fix and --auto-login on notify_failure
Post-merge / detect-type (push) Successful in 49s
Post-merge / release (push) Successful in 1m9s
Post-merge / validate-commit-msg (push) Successful in 1m13s
Post-merge / publish (push) Has been skipped
Post-merge / vikunja (push) Successful in 1m12s
Post-merge / badges (push) Successful in 1m20s
Post-merge / sync-wiki (push) Successful in 1m42s
Post-merge / configure-repo (push) Successful in 59s
2026-06-27 22:27:57 +00:00
gitea-actions-bot d1e1d05be5 chore: update badge URLs to commit e8fef3da [skip ci] 2026-06-27 20:28:52 +00:00
gitea-actions-bot dbb9bd7108 chore: update badge URLs to commit b316c4a8 [skip ci] 2026-06-27 20:26:55 +00:00
37 changed files with 1423 additions and 1244 deletions
+14 -1
View File
@@ -24,6 +24,9 @@ GITEA_REGISTRATION_TOKEN=your-registration-token
# Default SSH user for remote hosts (optional, overrides --user)
# GITEA_RUNNER_USER=ubuntu
# Repository for grm trigger-workflow (optional, default: oblachno-oss/grm)
# GRM_REPO=oblachno-oss/grm
# Default SSH private key path (optional, overrides --key)
# GITEA_RUNNER_KEY=~/.ssh/id_ed25519
@@ -34,13 +37,23 @@ GITEA_REGISTRATION_TOKEN=your-registration-token
# GITEA_RUNNER_LABELS=docker:docker://gitea/runner-images:ubuntu-latest
# UI language for GRM console messages (optional, default: en)
# Supported: en, bg, de, ru, zh
# Supported: en, bg, de, ru, zh, pl
# GRM_LANG=en
# Sudo password file for Ansible become operations (optional)
# When set, GRM reads the sudo password from this file instead of prompting.
# Priority: --become-password-file CLI flag > GRM_BECOME_PASSWORD_FILE > ANSIBLE_BECOME_PASSWORD_FILE
# GRM_BECOME_PASSWORD_FILE=~/.grm-sudo-pass
# ANSIBLE_BECOME_PASSWORD_FILE=~/.grm-sudo-pass
# Gitea PyPI registry username (for private package access)
# Used by PIP_INSTALL to configure PIP_EXTRA_INDEX_URL
CI_GITEA_USERNAME=emil
# Vikunja API token (required for `make create-task` dev workflow)
# Generate at: Vikunja → Settings → API Tokens
# VIKUNJA_TOKEN=your-vikunja-api-token
# devx configuration (GRM-specific overrides)
# Task prefix for Vikunja task IDs
DEVX_TASK_PREFIX=GRM
+1
View File
@@ -188,6 +188,7 @@ jobs:
CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
CI_GITEA_USERNAME: ${{ vars.CI_GITEA_USERNAME }}
RUN_ID: ${{ github.run_id }}
ANSIBLE_INJECT_INVOCATION: "1"
JOB_NAME: ${{ github.job }}
MATRIX_INDEX: ${{ matrix.runner-index }}
GITEA_REPOSITORY: ${{ github.repository }}
+1
View File
@@ -29,6 +29,7 @@ name: Post-merge
on:
push:
branches: [master]
workflow_dispatch:
env:
CI_GITEA_TOKEN: ${{ secrets.CI_GITEA_TOKEN }}
-7
View File
@@ -64,10 +64,3 @@ repos:
types: [python]
pass_filenames: false
stages: [pre-push]
- id: commit-msg
name: validate commit message
entry: env PYTHONPATH=src .venv/bin/python -m devx.ci.validate_commit_msg
language: system
stages: [commit-msg]
pass_filenames: true
+9 -11
View File
@@ -98,8 +98,7 @@ docs: update README
### 6. Review the PR (Mandatory — Before Adding ready-to-merge Label)
**Review checklist:** Every PR is reviewed against
[REVIEW_CHECKLIST.md](REVIEW_CHECKLIST.md) — 13 categories covering
**Review checklist:** Every PR is reviewed against 13 categories covering
architecture, code quality, security, i18n, testing, performance,
UX, documentation, workflow compliance, maintainability, resource
management, backwards compatibility, and logging.
@@ -120,11 +119,11 @@ the **[auto]** items in the checklist:
- Commit conventions (conventional commit format on PR commits)
The automated review posts inline comments on specific lines and
includes a link to the full checklist. The agent **must** address all
includes a summary of the checklist categories. The agent **must** address all
`REQUEST_CHANGES` issues before proceeding.
**Manual review (agent):** After the automated review passes, the agent
must go through **every category** in `REVIEW_CHECKLIST.md` and verify
must go through **every category** listed above and verify
the **[manual]** items by reviewing the full diff
(`git diff master...HEAD`).
@@ -132,8 +131,7 @@ Post review comments using `devx.ci.pr_review` (run as `python -m devx.ci.pr_rev
```bash
CI_GITEA_TOKEN=<token> python -m devx.ci.pr_review <pr_number> <owner/repo> \
--event REQUEST_CHANGES \
--body "Review summary" \
--comments-json comments.json
--body "Review summary"
```
### 7. Address Review Comments
@@ -146,7 +144,7 @@ an approval review with `--checklist-confirmed` and `--checklist-categories`:
CI_GITEA_TOKEN=<token> python -m devx.ci.pr_review <pr_number> <owner/repo> \
--event APPROVE --checklist-confirmed \
--checklist-categories 1,2,3,4,5,6,7,8,9,10,11,12,13 \
--body "All 13 REVIEW_CHECKLIST.md categories verified. Architecture: <summary>. Security: <summary>. Tests: <summary>. Docs: <summary>."
--body "All 13 checklist categories verified. Architecture: <summary>. Security: <summary>. Tests: <summary>. Docs: <summary>."
```
The `--checklist-confirmed` flag is **required** for APPROVE events —
@@ -253,16 +251,16 @@ via `[tool.devx.classify]` in `pyproject.toml`.
- `scripts/**` — Dev tools and CI/CD automation (not part of installed package)
- `docs/**` — Documentation
- `tests/**` — Test files
- `AGENTS.md`, `README.md`, `CHANGELOG.md`, `TROUBLESHOOTING.md`, `CONTRIBUTING.md`, `CODE_OF_CONDUCT.md`, `REVIEW_CHECKLIST.md` — Project docs
- `AGENTS.md`, `README.md`, `CHANGELOG.md`, `TROUBLESHOOTING.md`, `CONTRIBUTING.md` — Project docs
- `Makefile`, `cliff.toml`, `uv.lock` — Build tooling
- `.pre-commit-config.yaml`, `.ruff.toml`, `.ansible-lint`, `.checkmake.ini`, `.editorconfig` — Lint config
- `.env.example`, `.gitignore`, `.gitattributes` — Config
- `.pre-commit-config.yaml`, `.ansible-lint`, `.checkmake.ini` — Lint config (ruff config is in `pyproject.toml`)
- `.env.example`, `.gitignore` — Config
- `.devin/**` — Agent/CI tooling config
- `hooks/**` — Git hooks
- `activate.sh`, `activate.fish`, `activate.zsh` — Generated venv scripts
**User-facing paths** (tool changes → release needed) — everything else:
- `src/gitea_runner_manager/**` — Python CLI source (except `__init__.py` and `api_clients.py`)
- `src/gitea_runner_manager/**` — Python CLI source (except `__init__.py`)
- `ansible/**` — Ansible role
- `pyproject.toml` — Package metadata
- Any new file type not in the allowlist
+18
View File
@@ -2,6 +2,24 @@
All notable changes to this project will be documented in this file.
## [0.12.0] - 2026-06-28
### Features
- Upgrade all dependencies, add trigger-workflow command
## [0.11.1] - 2026-06-28
### Bug Fixes
- Makefile HOST/NAME requirement errors, add restart and list targets
## [0.11.0] - 2026-06-28
### Features
- Unified --become-password-file, --verbose, --no-status, labels fix
## [0.10.3] - 2026-06-27
### Bug Fixes
+14 -7
View File
@@ -109,28 +109,35 @@ update:
$(BIN)/grm update $(HOST) $(if $(USER),--user $(USER),) $(if $(KEY),--key $(KEY),) $(if $(VERSION),--version $(VERSION),) $(if $(ASK_BECOME_PASS),--ask-become-pass,)
start:
@if [ -z "$(HOST)" ]; then echo "HOST is required. Example: make start HOST=192.168.1.10"; exit 1; fi
@if [ -z "$(NAME)" ]; then echo "NAME is required. Example: make start NAME=runner1"; exit 1; fi
$(BIN)/grm start $(NAME) $(if $(HOST),--host $(HOST),) $(if $(USER),--user $(USER),) $(if $(ASK_BECOME_PASS),--ask-become-pass,)
stop:
@if [ -z "$(HOST)" ]; then echo "HOST is required. Example: make stop HOST=192.168.1.10"; exit 1; fi
@if [ -z "$(NAME)" ]; then echo "NAME is required. Example: make stop NAME=runner1"; exit 1; fi
$(BIN)/grm stop $(NAME) $(if $(HOST),--host $(HOST),) $(if $(USER),--user $(USER),) $(if $(ASK_BECOME_PASS),--ask-become-pass,)
restart:
@if [ -z "$(NAME)" ]; then echo "NAME is required. Example: make restart NAME=runner1"; exit 1; fi
$(BIN)/grm restart $(NAME) $(if $(HOST),--host $(HOST),) $(if $(USER),--user $(USER),) $(if $(ASK_BECOME_PASS),--ask-become-pass,)
enable:
@if [ -z "$(HOST)" ]; then echo "HOST is required. Example: make enable HOST=192.168.1.10"; exit 1; fi
@if [ -z "$(NAME)" ]; then echo "NAME is required. Example: make enable NAME=runner1"; exit 1; fi
$(BIN)/grm enable $(NAME) $(if $(HOST),--host $(HOST),) $(if $(USER),--user $(USER),) $(if $(ASK_BECOME_PASS),--ask-become-pass,)
disable:
@if [ -z "$(HOST)" ]; then echo "HOST is required. Example: make disable HOST=192.168.1.10"; exit 1; fi
@if [ -z "$(NAME)" ]; then echo "NAME is required. Example: make disable NAME=runner1"; exit 1; fi
$(BIN)/grm disable $(NAME) $(if $(HOST),--host $(HOST),) $(if $(USER),--user $(USER),) $(if $(TOKEN),--token $(TOKEN),) $(if $(ASK_BECOME_PASS),--ask-become-pass,)
status:
@if [ -z "$(HOST)" ]; then echo "HOST is required. Example: make status HOST=192.168.1.10"; exit 1; fi
@if [ -z "$(NAME)" ]; then echo "NAME is required. Example: make status NAME=runner1"; exit 1; fi
$(BIN)/grm status $(NAME) $(if $(HOST),--host $(HOST),) $(if $(USER),--user $(USER),) $(if $(ASK_BECOME_PASS),--ask-become-pass,)
remove:
@if [ -z "$(HOST)" ]; then echo "HOST is required. Example: make remove HOST=192.168.1.10"; exit 1; fi
$(BIN)/grm remove $(NAME) $(if $(HOST),--host $(HOST),) $(if $(USER),--user $(USER),) $(if $(TOKEN),--token $(TOKEN),) $(if $(ASK_BECOME_PASS),--ask-become-pass,)
@if [ -z "$(NAME)" ]; then echo "NAME is required. Example: make remove NAME=runner1"; exit 1; fi
$(BIN)/grm remove $(NAME) $(if $(HOST),--host $(HOST),) $(if $(USER),--user $(USER),) $(if $(TOKEN),--token $(TOKEN),) $(if $(FORCE),--force,) $(if $(ASK_BECOME_PASS),--ask-become-pass,)
list:
$(BIN)/grm list $(if $(NO_STATUS),--no-status,) $(if $(ASK_BECOME_PASS),--ask-become-pass,)
# --- Aliases to devx.mak targets ----------------------------------------------
lint-ruff: devx-lint-ruff
+79 -10
View File
@@ -8,12 +8,12 @@ Each runner runs in an isolated **rootless Docker** environment under a dedicate
[![CI](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/actions/workflows/ci.yml/badge.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/actions)
[![License: GPL-3.0](https://img.shields.io/badge/license-GPL--3.0-blue)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/src/branch/master/LICENSE)
[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/fc36a6d7330ffc2a6e68e3f8ce5a1dcd3b2c0983/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/actions)
[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/fc36a6d7330ffc2a6e68e3f8ce5a1dcd3b2c0983/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/actions)
[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/fc36a6d7330ffc2a6e68e3f8ce5a1dcd3b2c0983/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/wiki)
[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/fc36a6d7330ffc2a6e68e3f8ce5a1dcd3b2c0983/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/actions)
[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/fc36a6d7330ffc2a6e68e3f8ce5a1dcd3b2c0983/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/releases)
[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/fc36a6d7330ffc2a6e68e3f8ce5a1dcd3b2c0983/python.svg)](https://www.python.org/downloads/)
[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/dfe10dfc2483ccdda434e9b067348921257fdffd/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/actions)
[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/dfe10dfc2483ccdda434e9b067348921257fdffd/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/actions)
[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/dfe10dfc2483ccdda434e9b067348921257fdffd/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/wiki)
[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/dfe10dfc2483ccdda434e9b067348921257fdffd/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/actions)
[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/dfe10dfc2483ccdda434e9b067348921257fdffd/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/releases)
[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/dfe10dfc2483ccdda434e9b067348921257fdffd/python.svg)](https://www.python.org/downloads/)
## Why GRM?
@@ -143,14 +143,19 @@ GRM provides a single `grm` command with subcommands for the full runner lifecyc
| `grm update <host>` | Update the Gitea Runner binary on a remote host |
| `grm start <name>` | Start a registered runner |
| `grm stop <name>` | Stop a registered runner |
| `grm restart <name>` | Restart a runner (stop, prune Docker images, start) |
| `grm enable <name>` | Enable a runner to start on boot |
| `grm disable <name>` | Disable and deregister a runner |
| `grm status <name>` | Check the status of a registered runner |
| `grm remove <name>` | Remove a runner completely (with remote cleanup) |
| `grm remove <name> --force` | Remove only the local registry entry (skip remote cleanup) |
| `grm list` | List all registered runners with live status |
| `grm list --no-status` | List registered runners without SSH status checks |
| `grm trigger-workflow <workflow_id>` | Trigger a Gitea Actions workflow via the API |
| `grm trigger-workflow --list` | List available workflows in the repository |
| `grm --version` | Show the installed version |
All lifecycle commands (`start`, `stop`, `enable`, `disable`, `status`, `remove`) work by runner name and pull connection details from the local registry. You can override any stored value with `--host`, `--user`, or `--key`.
All lifecycle commands (`start`, `stop`, `restart`, `enable`, `disable`, `status`, `remove`) work by runner name and pull connection details from the local registry. You can override any stored value with `--host`, `--user`, or `--key`.
See the [CLI Commands](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/wiki/CLI-Commands.-) wiki page for full argument and option reference.
@@ -174,8 +179,73 @@ GRM reads configuration from a `.env` file in the current directory (loaded auto
| `GITEA_RUNNER_USER` | current login | Default SSH user (overrides `--user`) |
| `GITEA_RUNNER_KEY` | — | Default SSH key path (overrides `--key`) |
| `GITEA_RUNNER_LABELS` | — | Default runner labels (overrides `--labels`) |
| `GRM_LANG` | `en` | UI language: `en`, `bg`, `de`, `ru`, `zh` |
| `GRM_LANG` | `en` | UI language: `en`, `bg`, `de`, `ru`, `zh`, `pl` |
| `GRM_LOG_LEVEL` | `INFO` | Console log level: `DEBUG`, `INFO`, `WARNING`, `ERROR`, `CRITICAL` |
| `GRM_BECOME_PASSWORD_FILE` | — | Path to file containing sudo password (see [Sudo Password Handling](#sudo-password-handling)) |
| `ANSIBLE_BECOME_PASSWORD_FILE` | — | Fallback sudo password file path (Ansible-native env var) |
### Sudo Password Handling
GRM delegates remote operations to Ansible, which uses `sudo` (become) on the target host. There are several ways to provide the sudo password, in priority order:
1. **`--become-password-file <path>`** (CLI flag, global) — Read sudo password from a file. Works for all commands including `grm list`.
2. **`GRM_BECOME_PASSWORD_FILE`** (env var) — Same as above, set in `.env` or environment.
3. **`ANSIBLE_BECOME_PASSWORD_FILE`** (env var) — Fallback, Ansible-native env var.
4. **Interactive prompt** — If none of the above are set, GRM prompts for the sudo password (hidden input).
5. **Piped stdin** — When stdin is not a TTY, reads the first line: `echo 'password' | grm list`.
6. **`--no-ask-become-pass`** — Skip sudo password entirely (use when the target user has passwordless sudo).
For `grm list` specifically, the password is collected once and reused for all runner status checks via `--become-password-file`, avoiding stdin consumption issues when checking multiple runners.
**Examples:**
```bash
# Interactive prompt (default)
grm install 192.168.1.10 --user ubuntu
# Password file (recommended for automation)
echo 'my-sudo-pass' > ~/.grm-sudo-pass
chmod 600 ~/.grm-sudo-pass
grm --become-password-file ~/.grm-sudo-pass install 192.168.1.10 --user ubuntu
# Env var (set in .env)
GRM_BECOME_PASSWORD_FILE=~/.grm-sudo-pass
grm list # uses the file automatically
# Piped stdin (for scripts)
echo 'my-sudo-pass' | grm list
# Passwordless sudo on target
grm install 192.168.1.10 --user ubuntu --no-ask-become-pass
```
### Verbose Output
Pass `-v` / `--verbose` (global flag, before the subcommand) to enable Ansible verbose mode (`-v`):
```bash
grm --verbose install 192.168.1.10 --user ubuntu
grm -v status prod-runner
```
### Runner Labels
Runner labels control which jobs a runner accepts. They are set at installation time:
- **`--labels "docker:docker://alpine:latest"`** — Set specific labels.
- **`--labels ""`** — Explicitly set **no labels** (overrides `GITEA_RUNNER_LABELS` env var).
- **No `--labels` flag** — Uses `GITEA_RUNNER_LABELS` env var if set, otherwise the Ansible role default.
```bash
# Custom labels
grm install 192.168.1.10 --user ubuntu --labels "docker:docker://alpine:latest,ubuntu-22.04:docker://ubuntu:22.04"
# Explicitly no labels (overrides GITEA_RUNNER_LABELS env var)
grm install 192.168.1.10 --user ubuntu --labels ""
# Use GITEA_RUNNER_LABELS from .env (or role default if unset)
grm install 192.168.1.10 --user ubuntu
```
### Getting tokens
@@ -293,8 +363,7 @@ grm install <host>
| `executor.py` | Ansible subprocess execution with log capture |
| `registry.py` | Local JSON runner registry at `~/.local/share/grm/runners.json` |
| `i18n.py` | Internationalisation (en, bg, de, ru, zh, pl) |
| `exceptions.py` | Custom exceptions (`GRMError`, `AnsibleError`, `APIError`) |
| `config.py` | Configuration constants (API URLs, repo owner/name) |
| `exceptions.py` | Custom exceptions (`GRMError`, `AnsibleError`) |
| `logging_config.py` | Logging to `~/.local/state/grm/logs/grm.log` |
| `report.py` | Operation report tracking with step status |
| `ui.py` | Colorised console output via Click |
+4 -2
View File
@@ -1,5 +1,7 @@
collections:
- name: community.general
version: ">=13.0.1"
version: ">=13.1.0"
- name: ansible.posix
version: ">=1.5.4"
version: ">=2.2.0"
- name: community.docker
version: ">=5.2.1"
+6 -6
View File
@@ -8,12 +8,12 @@ Each runner runs in an isolated **rootless Docker** environment under a dedicate
[![CI](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/actions/workflows/ci.yml/badge.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/actions)
[![License: GPL-3.0](https://img.shields.io/badge/license-GPL--3.0-blue)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/src/branch/master/LICENSE)
[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/fc36a6d7330ffc2a6e68e3f8ce5a1dcd3b2c0983/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/actions)
[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/fc36a6d7330ffc2a6e68e3f8ce5a1dcd3b2c0983/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/actions)
[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/fc36a6d7330ffc2a6e68e3f8ce5a1dcd3b2c0983/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/wiki)
[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/fc36a6d7330ffc2a6e68e3f8ce5a1dcd3b2c0983/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/actions)
[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/fc36a6d7330ffc2a6e68e3f8ce5a1dcd3b2c0983/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/releases)
[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/fc36a6d7330ffc2a6e68e3f8ce5a1dcd3b2c0983/python.svg)](https://www.python.org/downloads/)
[![Coverage](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/dfe10dfc2483ccdda434e9b067348921257fdffd/coverage.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/actions)
[![Tests](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/dfe10dfc2483ccdda434e9b067348921257fdffd/tests.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/actions)
[![Docs](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/dfe10dfc2483ccdda434e9b067348921257fdffd/docs.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/wiki)
[![Code Quality](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/dfe10dfc2483ccdda434e9b067348921257fdffd/quality.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/actions)
[![Version](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/dfe10dfc2483ccdda434e9b067348921257fdffd/version.svg)](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/releases)
[![Python](https://git.oblachno.oblachno.fyi/oblachno-oss/grm/raw/commit/dfe10dfc2483ccdda434e9b067348921257fdffd/python.svg)](https://www.python.org/downloads/)
## Overview
+1 -3
View File
@@ -210,12 +210,10 @@ The Python CLI layer (`src/gitea_runner_manager/`) consists of the following mod
| `executor.py` | Ansible subprocess execution — runs `ansible-playbook` with extra-vars via temp JSON files, streams output to log files |
| `registry.py` | Local JSON runner registry at `~/.local/share/grm/runners.json` — stores connection metadata |
| `i18n.py` | Internationalisation translations (en, bg, de, ru, zh, pl) — opt-in via `GRM_LANG` environment variable |
| `exceptions.py` | Custom exceptions (`GRMError`, `AnsibleError`, `APIError`) |
| `config.py` | Configuration constants (API URLs, repo owner/name, project IDs) — overridable via environment variables |
| `exceptions.py` | Custom exceptions (`GRMError`, `AnsibleError`) |
| `logging_config.py` | Logging configuration — writes all messages to `~/.local/state/grm/logs/grm.log` at DEBUG level |
| `report.py` | Operation report tracking — prints a step-by-step report with status icons after each command |
| `ui.py` | User-facing output utilities — colorised console output via `click.style`, with log file always receiving plain text |
| `api_clients.py` | Gitea and Vikunja API client classes for CI automation scripts (not used by the CLI itself) |
| `translations.json` | Translation strings for all supported languages |
## Logging
+2 -2
View File
@@ -155,7 +155,7 @@ Once all comments are addressed, post an approval review:
CI_GITEA_TOKEN=<token> python -m devx.ci.review_pr <pr_number> <owner/repo> \
--event APPROVE --checklist-confirmed \
--checklist-categories 1,2,3,4,5,6,7,8,9,10,11,12,13 \
--body "All 13 REVIEW_CHECKLIST.md categories verified."
--body "All 13 checklist categories verified."
```
Then add the `ready-to-merge` label. The auto-merge workflow will:
@@ -216,7 +216,7 @@ Not all changes require a new release. The project classifies changes using `dev
- Lint config files, `.env.example`, `.gitignore`
**User-facing paths** (release needed):
- `src/gitea_runner_manager/**` (except `__init__.py` and `api_clients.py`)
- `src/gitea_runner_manager/**` (except `__init__.py`)
- `ansible/**`
- `pyproject.toml`
+1 -1
View File
@@ -94,7 +94,7 @@ Key technical decisions for the GRM project, extracted from `CHANGELOG.md` and `
**Decision:** Classify changed files into user-facing and workflow-only categories using `devx.ci.classify_changes`. Only user-facing changes trigger a release; workflow-only changes (CI, docs, tests, lint config) do not.
**Rationale:** Not all changes require a new release. CI workflow updates, documentation improvements, and test additions should not produce a new version tag. The classification is config-driven via `[tool.devx.classify]` in `pyproject.toml`. The strategy is safe-by-default: any file NOT in the explicit workflow-only allowlist is treated as user-facing, preventing new file types from accidentally skipping releases. User-facing paths include `src/gitea_runner_manager/**` (except `__init__.py` and `api_clients.py`) and `ansible/**`. Workflow-only paths include `.gitea/**`, `docs/**`, `tests/**`, `scripts/**`, and various config files.
**Rationale:** Not all changes require a new release. CI workflow updates, documentation improvements, and test additions should not produce a new version tag. The classification is config-driven via `[tool.devx.classify]` in `pyproject.toml`. The strategy is safe-by-default: any file NOT in the explicit workflow-only allowlist is treated as user-facing, preventing new file types from accidentally skipping releases. User-facing paths include `src/gitea_runner_manager/**` (except `__init__.py`) and `ansible/**`. Workflow-only paths include `.gitea/**`, `docs/**`, `tests/**`, `scripts/**`, and various config files.
**Source:** `AGENTS.md` (Smart CI: User-Facing vs Workflow-Only Changes), `pyproject.toml` (`[tool.devx.classify]`)
-2
View File
@@ -11,11 +11,9 @@
│ ├── registry.py # Local JSON runner registry
│ ├── i18n.py # Translations (en, bg, de, ru, zh, pl)
│ ├── exceptions.py # Custom exceptions
│ ├── config.py # Configuration constants
│ ├── logging_config.py # Logging to ~/.local/state/grm/logs/
│ ├── report.py # Operation report tracking
│ ├── ui.py # Colorised console output
│ ├── api_clients.py # Gitea/Vikunja API clients (for CI scripts)
│ └── translations.json # Translation strings
├── ansible/
│ ├── roles/gitea-runner/ # Main Ansible role
+11 -15
View File
@@ -14,10 +14,9 @@ classifiers = [
"License :: OSI Approved :: GNU General Public License v3 (GPLv3)",
]
dependencies = [
"requests>=2.34.2",
"python-dotenv>=1.2.2",
"click>=8.4.1",
"ansible>=14.0.0",
"click>=8.4.2",
"ansible>=14.1.0",
]
[project.scripts]
@@ -27,22 +26,22 @@ grm = "gitea_runner_manager.cli:cli"
version = {attr = "gitea_runner_manager.__version__"}
[project.optional-dependencies]
# Minimal deps for CI scripts that only need click/dotenv/requests
# Minimal deps for CI scripts that only need click/dotenv
# (detect-changes, discover-runners, pr-review, sync-wiki, badges, etc.)
ci = [
"pytest>=9.1.0",
"pytest>=9.1.1",
"pytest-cov>=7.1.0",
"build>=1.5.0",
"twine>=6.2.0",
# Reusable CI/CD and dev tools (auto-merge, pr-review, pre-push checks, etc.)
"devx>=0.23.0",
"devx>=0.26.0",
]
# Lint and type-checking tools (quality job)
lint = [
"ruff>=0.15.17",
"pyright>=1.1.410",
"bandit>=1.8.2",
"pip-audit>=2.10",
"ruff>=0.15.20",
"pyright>=1.1.411",
"bandit>=1.9.4",
"pip-audit>=2.10.1",
"pre-commit>=4.6.0",
"ansible-lint>=26.4.0",
]
@@ -55,7 +54,7 @@ molecule = [
dev = [
"gitea-runner-manager[ci,lint,molecule]",
# Reusable CI/CD and dev tools (pre-push hooks, create-task, create-pr)
"devx>=0.23.0",
"devx>=0.26.0",
# Non-Python dev dependency: checkmake (Makefile linter)
# Install via: go install github.com/checkmake/checkmake/cmd/checkmake@latest
]
@@ -100,7 +99,7 @@ strict = ["src/gitea_runner_manager"]
task_prefix = "GRM"
vikunja_project_id = 6
repo_owner = "oblachno-oss"
repo_name = "gitea-runner-manager"
repo_name = "grm"
# Molecule test weights for LPT scheduling.
# GRM has a single role (gitea-runner) with 7 scenarios.
@@ -128,11 +127,8 @@ infrastructure = ["scripts/**"]
# Infrastructure overrides — files that would default to user-facing
# but are actually infrastructure:
# - __init__.py: only contains __version__ (set by release.py, not user code)
# - api_clients.py: used only by tests and legacy CI scripts (now in devx),
# not by the grm CLI tool itself
infrastructure_overrides = [
"src/gitea_runner_manager/__init__.py",
"src/gitea_runner_manager/api_clients.py",
]
# User-facing overrides — safety override for broad infrastructure patterns
+1 -1
View File
@@ -1,3 +1,3 @@
"""Gitea Runner Manager — lean CLI for managing Gitea Actions runners."""
__version__ = "0.10.3"
__version__ = "0.12.0"
-353
View File
@@ -1,353 +0,0 @@
"""Reusable HTTP API clients for Gitea and Vikunja."""
from __future__ import annotations
import logging
import time
from typing import Any
import requests
from .config import DEFAULT_TIMEOUT
from .exceptions import APIError
logger = logging.getLogger("grm")
# Retry configuration for transient errors (429, 5xx, connection errors)
MAX_RETRIES = 3
RETRY_BACKOFF_BASE = 2 # seconds: 2, 4, 8
RETRY_STATUS_CODES = {429, 500, 502, 503, 504}
def _parse_error(e: requests.HTTPError) -> tuple[int, str]:
"""Extract status code and message from an HTTPError response."""
response = getattr(e, "response", None)
status = response.status_code if response is not None else 0
try:
body: dict[str, Any] = response.json() if response is not None else {}
message: str = body.get("message", str(e))
except Exception:
message = str(e)
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 GiteaClient:
"""Low-level Gitea REST API client with connection pooling."""
def __init__(self, base_url: str, token: str, owner: str, repo: str) -> None:
self._base_url = base_url.rstrip("/")
self._owner = owner
self._repo = repo
self._session = requests.Session()
self._session.headers.update(
{
"Authorization": f"token {token}",
"Content-Type": "application/json",
}
)
def _url(self, path: str) -> str:
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
# -- repo settings --
def update_repo_settings(self, settings: dict[str, Any]) -> dict[str, Any]:
"""Update repository settings (e.g. auto-delete branch after merge)."""
r = self._request("PATCH", "", json=settings)
return r.json()
# -- branch protection --
def list_branch_protections(self) -> list[dict[str, Any]]:
r = self._request("GET", "/branch_protections")
return r.json()
def create_branch_protection(self, config: dict[str, Any]) -> dict[str, Any]:
r = self._request("POST", "/branch_protections", json=config)
return r.json()
def update_branch_protection(self, branch: str, config: dict[str, Any]) -> dict[str, Any]:
r = self._request("PATCH", f"/branch_protections/{branch}", json=config)
return r.json()
def ensure_branch_protection(self, branch: str, config: dict[str, Any]) -> dict[str, Any]:
"""Idempotent: create or update branch protection for the given branch."""
existing = self.list_branch_protections()
for p in existing:
if p.get("branch_name") == branch:
update_config = {k: v for k, v in config.items() if k != "branch_name"}
return self.update_branch_protection(branch, update_config)
return self.create_branch_protection(config)
# -- labels --
def list_labels(self) -> list[dict[str, Any]]:
r = self._request("GET", "/labels")
return r.json()
def create_label(self, name: str, color: str, description: str = "") -> dict[str, Any]:
r = self._request(
"POST",
"/labels",
json={"name": name, "color": color, "description": description},
)
return r.json()
def ensure_label(self, name: str, color: str, description: str = "") -> dict[str, Any] | None:
"""Idempotent: create label if it doesn't already exist."""
labels = self.list_labels()
for label in labels:
if label["name"] == name:
return None
return self.create_label(name, color, description)
def create_issue(self, title: str, body: str = "", labels: list[int] | None = None) -> dict[str, Any]:
"""Create a new issue in the repository.
Args:
labels: List of label IDs (integers, not names).
"""
payload: dict[str, Any] = {"title": title, "body": body}
if labels:
payload["labels"] = labels
r = self._request("POST", "/issues", json=payload)
return r.json()
# -- pulls / releases --
def get_pr_labels(self, pr_number: str | int) -> list[dict[str, Any]]:
"""Fetch labels currently attached to a pull request."""
r = self._request("GET", f"/issues/{pr_number}/labels")
return r.json()
def merge_pr(self, pr_number: str | int, merge_title: str) -> None:
payload = {"Do": "squash", "MergeTitleField": merge_title}
self._request("POST", f"/pulls/{pr_number}/merge", json=payload)
def get_commit_status(self, sha: str) -> list[dict[str, Any]]:
"""Fetch all status check contexts reported for a commit.
Uses the combined status endpoint (/commits/{sha}/status) which
returns one entry per context (the latest), deduplicated server-side.
The plural endpoint (/commits/{sha}/statuses) returns every historical
entry including stale "pending" ones that never got updated.
"""
r = self._request("GET", f"/commits/{sha}/status")
data = r.json()
return data.get("statuses", [])
def get_pr(self, pr_number: str | int) -> dict[str, Any]:
"""Fetch pull request details including mergeable state."""
r = self._request("GET", f"/pulls/{pr_number}")
return r.json()
def get_pr_files(self, pr_number: str | int) -> list[dict[str, Any]]:
"""Fetch the list of files changed in a pull request."""
r = self._request("GET", f"/pulls/{pr_number}/files")
return r.json()
def get_pr_commits(self, pr_number: str | int) -> list[dict[str, Any]]:
"""Fetch the commits included in a pull request."""
r = self._request("GET", f"/pulls/{pr_number}/commits")
return r.json()
def get_pr_reviews(self, pr_number: str | int) -> list[dict[str, Any]]:
"""Fetch reviews posted on a pull request."""
r = self._request("GET", f"/pulls/{pr_number}/reviews")
return r.json()
def create_review(
self,
pr_number: str | int,
event: str = "COMMENT",
body: str = "",
comments: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
"""Post a review on a pull request.
Args:
event: ``APPROVED``, ``REQUEST_CHANGES``, or ``COMMENT``.
body: Top-level review body text.
comments: Line-level comments with ``path``, ``body``,
``new_position`` (and optionally ``old_position``).
"""
# Map common event names to Gitea API values
event_map = {"APPROVE": "APPROVED", "REQUEST_CHANGES": "REQUEST_CHANGES", "COMMENT": "COMMENT"}
gitea_event = event_map.get(event, event)
payload: dict[str, Any] = {"event": gitea_event, "body": body}
if comments:
payload["comments"] = comments
r = self._request("POST", f"/pulls/{pr_number}/reviews", json=payload)
return r.json()
def create_release(
self,
tag: str,
name: str = "",
body: str = "",
draft: bool = False,
prerelease: bool = False,
) -> dict[str, Any]:
payload = {
"tag_name": tag,
"name": name or tag,
"body": body,
"draft": draft,
"prerelease": prerelease,
}
r = self._request("POST", "/releases", json=payload)
return r.json()
def get_release_by_tag(self, tag: str) -> dict[str, Any] | None:
"""Fetch a release by its tag name. Returns None if not found."""
try:
r = self._request("GET", f"/releases/tags/{tag}")
return r.json()
except APIError:
return None
def create_release_idempotent(
self,
tag: str,
name: str = "",
body: str = "",
draft: bool = False,
prerelease: bool = False,
) -> dict[str, Any]:
"""Create a release, or return the existing one if it already exists.
This is idempotent safe to call multiple times for the same tag.
"""
existing = self.get_release_by_tag(tag)
if existing:
logger.info("Release for tag %s already exists (ID %s), skipping creation.", tag, existing.get("id"))
return existing
return self.create_release(tag=tag, name=name, body=body, draft=draft, prerelease=prerelease)
class VikunjaClient:
"""Low-level Vikunja REST API client with connection pooling."""
def __init__(self, base_url: str, token: str) -> None:
self._base_url = base_url.rstrip("/")
self._session = requests.Session()
self._session.headers.update({"Authorization": f"Bearer {token}"})
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
def list_tasks(self, **params: Any) -> list[dict[str, Any]]:
r = self._request("GET", "/tasks", params=params)
return r.json()
def get_task(self, task_id: int) -> dict[str, Any]:
"""Fetch a single task by its numeric ID."""
r = self._request("GET", f"/tasks/{task_id}")
return r.json()
def list_project_tasks(self, project_id: int, **params: Any) -> list[dict[str, Any]]:
"""List tasks in a specific project (more efficient than listing all tasks)."""
r = self._request("GET", f"/projects/{project_id}/tasks", params=params)
return r.json()
def post_comment(self, task_id: int, comment: str) -> None:
self._request("PUT", f"/tasks/{task_id}/comments", json={"comment": comment})
def update_task(self, task_id: int, **fields: Any) -> None:
self._request("POST", f"/tasks/{task_id}", json=fields)
+188 -29
View File
@@ -6,6 +6,7 @@ import functools
import os
import sys
from collections.abc import Callable
from pathlib import Path
from typing import Any
import click
@@ -13,12 +14,41 @@ from dotenv import load_dotenv # pyright: ignore[reportMissingImports,reportUnk
from . import __version__
from .exceptions import GRMError
from .gitea_client import GiteaAPIError, GiteaWorkflowClient
from .i18n import _
from .runner_manager import RunnerManager
load_dotenv(override=True)
def _default_user() -> str:
"""Default SSH user from env or current OS user."""
if user := os.getenv("GITEA_RUNNER_USER"):
return user
try:
return os.getlogin()
except OSError:
return os.getenv("USER", "root")
def _get_become_password_file() -> str | None:
"""Read become-password-file from Click context or env vars."""
ctx = click.get_current_context(silent=True)
if ctx and ctx.obj:
path = ctx.obj.get("become_password_file")
if path:
return path
return os.getenv("GRM_BECOME_PASSWORD_FILE") or os.getenv("ANSIBLE_BECOME_PASSWORD_FILE")
def _get_verbose() -> bool:
"""Read verbose flag from Click context."""
ctx = click.get_current_context(silent=True)
if ctx and ctx.obj:
return ctx.obj.get("verbose", False)
return False
def _runner_options(func: Callable[..., Any]) -> Callable[..., Any]:
"""Apply common override options for registry-based lifecycle commands."""
func = click.option(
@@ -50,8 +80,25 @@ def _handle_errors(msg_key: str) -> Callable[[Callable[..., Any]], Callable[...,
@click.group(help=_("Gitea Runner Manager — manage Gitea Actions runners."))
@click.version_option(version=__version__)
def cli() -> None:
pass
@click.option(
"--become-password-file",
envvar="GRM_BECOME_PASSWORD_FILE",
type=click.Path(exists=True, dir_okay=False, readable=True),
default=None,
help=_("Read sudo password from a file instead of prompting (env: GRM_BECOME_PASSWORD_FILE)"),
)
@click.option(
"--verbose",
"-v",
is_flag=True,
default=False,
help=_("Enable verbose Ansible output (-v flag passed to ansible)"),
)
@click.pass_context
def cli(ctx: click.Context, become_password_file: str | None, verbose: bool) -> None:
ctx.ensure_object(dict)
ctx.obj["become_password_file"] = become_password_file
ctx.obj["verbose"] = verbose
@cli.command(help=_("Install and configure a Gitea Runner on a remote host."))
@@ -59,8 +106,8 @@ def cli() -> None:
@click.option(
"--user",
"-u",
default=lambda: os.getenv("GITEA_RUNNER_USER", os.getlogin()),
help=_("SSH user"),
default=_default_user,
help=_("SSH user (env: GITEA_RUNNER_USER)"),
)
@click.option("--key", "-k", default=lambda: os.getenv("GITEA_RUNNER_KEY"), help=_("Path to SSH private key"))
@click.option("--name", "-n", help=_("Gitea Runner name (default: host)"))
@@ -91,14 +138,19 @@ def cli() -> None:
@click.option(
"--labels",
"-l",
default=lambda: os.getenv("GITEA_RUNNER_LABELS", ""),
help=_("Runner labels for Gitea Actions (env: GITEA_RUNNER_LABELS). Example: docker:docker://alpine:latest"),
default=None,
help=_(
"Runner labels (env: GITEA_RUNNER_LABELS). "
"Pass an empty string for no labels. "
"Example: docker:docker://alpine:latest"
),
)
@click.option(
"--ask-become-pass/--no-ask-become-pass",
default=True,
help=_("Prompt for sudo password (default)"),
)
@_handle_errors("Installation failed: {error}")
def install(
host: str,
user: str,
@@ -108,25 +160,26 @@ def install(
url: str,
admin_token: str | None,
integration_retries: int,
labels: str,
labels: str | None,
ask_become_pass: bool,
) -> None:
if labels is None:
labels = os.getenv("GITEA_RUNNER_LABELS")
manager = RunnerManager()
try:
manager.install(
host=host,
user=user,
key=key,
name=name,
token=token,
gitea_url=url,
admin_token=admin_token,
integration_retries=integration_retries,
labels=labels or None,
ask_become_pass=ask_become_pass,
)
except GRMError as e:
raise click.ClickException(_("Installation failed: {error}", error=e)) from e
manager.install(
host=host,
user=user,
key=key,
name=name,
token=token,
gitea_url=url,
admin_token=admin_token,
integration_retries=integration_retries,
labels=labels,
ask_become_pass=ask_become_pass,
become_password_file=_get_become_password_file(),
verbose=_get_verbose(),
)
@cli.command(help=_("Update the Gitea Runner binary on a remote host."))
@@ -134,8 +187,8 @@ def install(
@click.option(
"--user",
"-u",
default=lambda: os.getenv("GITEA_RUNNER_USER", os.getlogin()),
help=_("SSH user"),
default=_default_user,
help=_("SSH user (env: GITEA_RUNNER_USER)"),
)
@click.option("--key", "-k", default=lambda: os.getenv("GITEA_RUNNER_KEY"), help=_("Path to SSH private key"))
@click.option("--version", "-v", help=_("Specific Gitea Runner version"))
@@ -159,6 +212,8 @@ def update(
key=key,
version=version,
ask_become_pass=ask_become_pass,
become_password_file=_get_become_password_file(),
verbose=_get_verbose(),
)
@@ -180,6 +235,8 @@ def start(
user=user,
key=key,
ask_become_pass=ask_become_pass,
become_password_file=_get_become_password_file(),
verbose=_get_verbose(),
)
@@ -201,6 +258,8 @@ def stop(
user=user,
key=key,
ask_become_pass=ask_become_pass,
become_password_file=_get_become_password_file(),
verbose=_get_verbose(),
)
@@ -222,6 +281,8 @@ def restart(
user=user,
key=key,
ask_become_pass=ask_become_pass,
become_password_file=_get_become_password_file(),
verbose=_get_verbose(),
)
@@ -243,6 +304,8 @@ def enable(
user=user,
key=key,
ask_become_pass=ask_become_pass,
become_password_file=_get_become_password_file(),
verbose=_get_verbose(),
)
@@ -279,6 +342,8 @@ def disable(
token=token,
gitea_url=url,
ask_become_pass=ask_become_pass,
become_password_file=_get_become_password_file(),
verbose=_get_verbose(),
)
@@ -300,6 +365,8 @@ def status(
user=user,
key=key,
ask_become_pass=ask_become_pass,
become_password_file=_get_become_password_file(),
verbose=_get_verbose(),
)
@@ -344,15 +411,22 @@ def remove(
gitea_url=url,
ask_become_pass=ask_become_pass,
force=force,
become_password_file=_get_become_password_file(),
verbose=_get_verbose(),
)
def _collect_become_pass(ask_become_pass: bool) -> str | None:
"""Collect sudo password for ad-hoc status checks.
When stdin is a TTY, prompts interactively with hidden input.
When stdin is piped, reads the first line (e.g. ``echo 'pass' | grm list``).
Priority:
1. ``--become-password-file`` / ``GRM_BECOME_PASSWORD_FILE`` env var
2. ``ANSIBLE_BECOME_PASSWORD_FILE`` env var
3. Interactive prompt (TTY) or piped stdin (first line)
"""
password_file = _get_become_password_file()
if password_file:
return Path(password_file).read_text(encoding="utf-8").strip() or None
if not ask_become_pass:
return None
if sys.stdin.isatty():
@@ -374,12 +448,18 @@ def _collect_become_pass(ask_become_pass: bool) -> str | None:
default=True,
help=_("Prompt for sudo password once for all status checks (default)."),
)
@click.option(
"--no-status",
is_flag=True,
default=False,
help=_("Skip live SSH status checks and show registry entries only"),
)
@_handle_errors("List failed: {error}")
def list_runners(ask_become_pass: bool) -> None:
become_pass = _collect_become_pass(ask_become_pass)
def list_runners(ask_become_pass: bool, no_status: bool) -> None:
become_pass = None if no_status else _collect_become_pass(ask_become_pass)
manager = RunnerManager()
runners = manager.list_runners(become_pass=become_pass)
runners = manager.list_runners(become_pass=become_pass, no_status=no_status)
if not runners:
click.echo(_("No runners registered. Use 'grm install' to add one."))
@@ -389,3 +469,82 @@ def list_runners(ask_become_pass: bool) -> None:
click.echo("-" * 90)
for r in runners:
click.echo(f"{r['name']:<18} {r['host']:<16} {r['user']:<10} {r['labels']:<30} {r['status']}")
@cli.command(name="trigger-workflow", help=_("Trigger a Gitea Actions workflow via the API."))
@click.argument("workflow_id", required=False)
@click.option(
"--repo",
default=lambda: os.getenv("GRM_REPO", "oblachno-oss/grm"),
help=_("Repository in owner/repo format (env: GRM_REPO, default: oblachno-oss/grm)"),
)
@click.option(
"--ref",
default="master",
help=_("Git ref to run the workflow on (default: master)"),
)
@click.option(
"--url",
default=lambda: os.getenv("GITEA_URL", ""),
help=_("Gitea URL (env: GITEA_URL)"),
)
@click.option(
"--token",
default=lambda: os.getenv("CI_GITEA_TOKEN"),
help=_("Gitea API token (env: CI_GITEA_TOKEN)"),
)
@click.option(
"--list",
"list_only",
is_flag=True,
default=False,
help=_("List available workflows instead of triggering one"),
)
def trigger_workflow(
workflow_id: str,
repo: str,
ref: str,
url: str,
token: str | None,
list_only: bool,
) -> None:
"""Trigger a Gitea Actions workflow dispatch event."""
if not url:
raise click.ClickException(_("GITEA_URL is required (set --url or GITEA_URL env var)"))
if not token:
raise click.ClickException(_("CI_GITEA_TOKEN is required (set --token or CI_GITEA_TOKEN env var)"))
if not list_only and not workflow_id:
raise click.ClickException(_("WORKFLOW_ID is required unless --list is used"))
client = GiteaWorkflowClient(url, token)
owner, repo_name = repo.split("/", 1)
if list_only:
workflows = client.list_workflows(owner, repo_name)
if not workflows:
click.echo(_("No workflows found in {repo}", repo=repo))
return
click.echo(f"{'ID':<30} {'NAME':<20} {'PATH':<25} {'STATE'}")
click.echo("-" * 85)
for wf in workflows:
wf_id = str(wf.get("id", ""))
wf_name = wf.get("name", "")
wf_path = wf.get("path", "")
wf_state = wf.get("state", "")
click.echo(f"{wf_id:<30} {wf_name:<20} {wf_path:<25} {wf_state}")
return
click.echo(_("Triggering workflow {wf} on {repo}@{ref}...", wf=workflow_id, repo=repo, ref=ref))
try:
result = client.dispatch_workflow(owner, repo_name, workflow_id, ref)
except GiteaAPIError as e:
raise click.ClickException(str(e)) from e
if result and result.get("id"):
run_id: Any = result["id"]
click.echo(_("Workflow triggered successfully. Run ID: {run_id}", run_id=run_id))
if result.get("html_url"):
html_url: Any = result["html_url"]
click.echo(f" {html_url}")
else:
click.echo(_("Workflow triggered successfully."))
-43
View File
@@ -1,43 +0,0 @@
"""Shared configuration constants for GRM scripts and API clients."""
from __future__ import annotations
import os
import re
GITEA_API_URL = os.getenv("GRM_GITEA_API_URL", "https://git.oblachno.oblachno.fyi/api/v1")
VIKUNJA_API_URL = os.getenv("GRM_VIKUNJA_API_URL", "https://work.oblachno.oblachno.fyi/api/v1")
REPO_OWNER = os.getenv("GRM_REPO_OWNER", "oblachno-oss")
REPO_NAME = os.getenv("GRM_REPO_NAME", "grm")
VIKUNJA_PROJECT_ID = int(os.getenv("GRM_VIKUNJA_PROJECT_ID", "6"))
TASK_ID_RE = re.compile(r"GRM-\d+")
CONVENTIONAL_RE = re.compile(r"^(feat|fix|chore|docs|style|refactor|perf|test|ci|build|revert)(\(.+\))?: .+")
DEFAULT_TIMEOUT = 30
DEFAULT_PER_PAGE = 50
BRANCH_PROTECTION_CONFIG: dict[str, object] = {
"branch_name": "master",
"enable_push": True,
"enable_push_whitelist": True,
"push_whitelist_usernames": ["emil"],
"enable_status_check": True,
"status_check_contexts": [
"CI / quality (pull_request)",
"CI / molecule-tests (1) (pull_request)",
"CI / molecule-tests (2) (pull_request)",
"CI / molecule-tests (3) (pull_request)",
],
"required_approvals": 0,
"dismiss_stale_approvals": True,
"block_on_outdated_branch": True,
"block_on_rejected_reviews": True,
"block_on_official_review_requests": True,
}
REPO_SETTINGS_CONFIG: dict[str, object] = {
"default_delete_branch_after_merge": True,
}
-9
View File
@@ -11,12 +11,3 @@ class AnsibleError(GRMError):
"""Raised when an Ansible command fails."""
pass
class APIError(GRMError):
"""Raised when a REST API call returns an HTTP error."""
def __init__(self, status: int, message: str) -> None:
self.status = status
self.message = message
super().__init__(f"HTTP {status}: {message}")
+6 -2
View File
@@ -116,12 +116,16 @@ class AnsibleExecutor:
if become and ask_become_pass:
if become_pass:
fd, password_file = tempfile.mkstemp(suffix=".txt", prefix="grm-become-")
os.fchmod(fd, 0o600)
with os.fdopen(fd, "w") as f:
f.write(become_pass)
os.chmod(password_file, 0o600)
cmd.extend(["--become-password-file", password_file])
else:
cmd.append("--ask-become-pass")
env_password_file = os.getenv("ANSIBLE_BECOME_PASSWORD_FILE")
if env_password_file:
cmd.extend(["--become-password-file", env_password_file])
else:
cmd.append("--ask-become-pass")
env = os.environ.copy()
try:
+86
View File
@@ -0,0 +1,86 @@
"""Minimal Gitea API client for workflow operations.
Uses urllib from the standard library to avoid adding requests as a
runtime dependency. Only covers the Actions workflow dispatch endpoint.
"""
from __future__ import annotations
import json
import urllib.error
import urllib.request # noqa: PTH123 # nosec B404
from contextlib import suppress
from typing import Any
class GiteaAPIError(Exception):
"""Raised when a Gitea API call fails."""
def __init__(self, status: int, message: str) -> None:
super().__init__(f"Gitea API error {status}: {message}")
self.status = status
self.message = message
class GiteaWorkflowClient:
"""Thin client for Gitea Actions workflow API endpoints."""
def __init__(self, base_url: str, token: str) -> None:
self._base_url = base_url.rstrip("/")
self._token = token
def _request(self, method: str, path: str, body: dict[str, Any] | None = None) -> dict[str, Any] | None:
url = f"{self._base_url}/api/v1{path}"
data = json.dumps(body).encode("utf-8") if body else None
req = urllib.request.Request( # nosec B310
url,
data=data,
method=method,
)
req.add_header("Authorization", f"token {self._token}")
req.add_header("Content-Type", "application/json")
req.add_header("Accept", "application/json")
try:
with urllib.request.urlopen(req) as resp: # noqa: PTH123 # nosec B310
if resp.status == 204:
return None
raw = resp.read()
return json.loads(raw) if raw else None
except urllib.error.HTTPError as e:
detail = e.read().decode("utf-8", errors="replace")
with suppress(json.JSONDecodeError, ValueError):
detail = json.loads(detail).get("message", detail)
raise GiteaAPIError(e.code, detail) from e
def list_workflows(self, owner: str, repo: str) -> list[dict[str, Any]]:
"""List all workflows in a repository."""
result = self._request("GET", f"/repos/{owner}/{repo}/actions/workflows")
if result is None:
return []
return result.get("workflows", [])
def dispatch_workflow(
self,
owner: str,
repo: str,
workflow_id: str,
ref: str = "master",
inputs: dict[str, str] | None = None,
) -> dict[str, Any] | None:
"""Trigger a workflow dispatch event.
Args:
owner: Repository owner.
repo: Repository name.
workflow_id: Workflow file name (e.g. "ci.yml") or numeric ID.
ref: Git ref (branch/tag) to run on. Defaults to "master".
inputs: Optional workflow inputs.
Returns:
Run details dict if return_run_details is requested, else None.
"""
path = f"/repos/{owner}/{repo}/actions/workflows/{workflow_id}/dispatches?return_run_details=true"
body: dict[str, Any] = {"ref": ref}
if inputs:
body["inputs"] = inputs
return self._request("POST", path, body)
+13 -3
View File
@@ -7,12 +7,22 @@ Supported: en, bg, de, ru, zh, pl.
from __future__ import annotations
import json
import logging
import os
from pathlib import Path
TRANSLATIONS: dict[str, dict[str, str]] = json.loads(
(Path(__file__).parent / "translations.json").read_text(encoding="utf-8")
)
logger = logging.getLogger("grm")
def _load_translations() -> dict[str, dict[str, str]]:
try:
return json.loads((Path(__file__).parent / "translations.json").read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError) as e:
logger.warning("Failed to load translations.json: %s — falling back to English", e)
return {}
TRANSLATIONS: dict[str, dict[str, str]] = _load_translations()
def _(key: str, **kwargs: object) -> str:
+16 -6
View File
@@ -6,6 +6,7 @@ so that subsequent lifecycle commands only need the runner name.
from __future__ import annotations
import fcntl
import json
from datetime import UTC, datetime
from pathlib import Path
@@ -23,20 +24,29 @@ class RunnerRegistry:
self._data: dict[str, dict[str, Any]] = self._load()
def _load(self) -> dict[str, dict[str, Any]]:
if self._path.exists():
try:
with open(self._path) as f:
if not self._path.exists():
return {}
try:
with open(self._path) as f:
fcntl.flock(f.fileno(), fcntl.LOCK_SH)
try:
data: Any = json.load(f)
if isinstance(data, dict):
return cast(dict[str, dict[str, Any]], data)
except (json.JSONDecodeError, OSError):
pass
finally:
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
except (json.JSONDecodeError, OSError):
pass
return {}
def _save(self) -> None:
self._path.parent.mkdir(parents=True, exist_ok=True)
with open(self._path, "w") as f:
json.dump(self._data, f, indent=2)
fcntl.flock(f.fileno(), fcntl.LOCK_EX)
try:
json.dump(self._data, f, indent=2)
finally:
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
def add(
self,
-1
View File
@@ -49,7 +49,6 @@ def track_steps() -> Generator[StepTracker, None, None]:
for step in reversed(tracker.steps):
if step.status == "in_progress":
step.status = "failed"
break
raise
finally:
_print_report(tracker.steps)
+79 -9
View File
@@ -2,12 +2,11 @@
from __future__ import annotations
import contextlib
import json
import os
import tempfile
from collections.abc import Generator
from contextlib import contextmanager
from contextlib import contextmanager, suppress
from pathlib import Path
from .exceptions import AnsibleError
@@ -42,12 +41,12 @@ class RunnerManager:
return
fd, path = tempfile.mkstemp(suffix=".json", prefix="grm-vars-")
try:
os.fchmod(fd, 0o600)
with os.fdopen(fd, "w") as f:
json.dump(extra_vars, f)
os.chmod(path, 0o600)
yield path
finally:
with contextlib.suppress(FileNotFoundError):
with suppress(FileNotFoundError):
os.unlink(path)
def _run_playbook(
@@ -59,10 +58,21 @@ class RunnerManager:
key: str | None = None,
ask_become_pass: bool = False,
description: str = "",
become_password_file: str | None = None,
verbose: bool = False,
) -> None:
"""Build command with temp-file extra-vars and execute via executor."""
with self._extra_vars_file(extra_vars) as vars_file:
cmd = self._build_cmd(playbook_name, host, user, vars_file, key, ask_become_pass)
cmd = self._build_cmd(
playbook_name,
host,
user,
vars_file,
key,
ask_become_pass,
become_password_file,
verbose,
)
self._executor.run(cmd, description=description)
def install(
@@ -77,6 +87,8 @@ class RunnerManager:
integration_retries: int = 3,
ask_become_pass: bool = False,
labels: str | None = None,
become_password_file: str | None = None,
verbose: bool = False,
) -> None:
"""Install a runner on a remote host using Ansible."""
if not name:
@@ -94,7 +106,7 @@ class RunnerManager:
}
if admin_token:
extra_vars["gitea_admin_token"] = admin_token
if labels:
if labels is not None:
extra_vars["runner_labels"] = labels
with track_steps() as tracker:
@@ -107,6 +119,8 @@ class RunnerManager:
key,
ask_become_pass,
description=_("Installing Gitea Runner on {host}", host=host),
become_password_file=become_password_file,
verbose=verbose,
)
tracker.done()
@@ -128,6 +142,8 @@ class RunnerManager:
key: str | None = None,
version: str | None = None,
ask_become_pass: bool = False,
become_password_file: str | None = None,
verbose: bool = False,
) -> None:
"""Update the gitea_runner binary on a remote host."""
extra_vars: dict[str, str | int] | None = None
@@ -144,6 +160,8 @@ class RunnerManager:
key,
ask_become_pass,
description=_("Updating Gitea Runner on {host}", host=host),
become_password_file=become_password_file,
verbose=verbose,
)
tracker.done()
@@ -187,6 +205,8 @@ class RunnerManager:
user: str | None = None,
key: str | None = None,
ask_become_pass: bool = False,
become_password_file: str | None = None,
verbose: bool = False,
) -> None:
"""Start a runner instance on a remote host."""
actual_host, actual_user, actual_key, _gitea_url = self._resolve_runner(name, host, user, key)
@@ -200,6 +220,8 @@ class RunnerManager:
actual_key,
ask_become_pass,
description=_("Starting Gitea Runner {name} on {host}", name=name, host=actual_host),
become_password_file=become_password_file,
verbose=verbose,
)
tracker.done()
@@ -210,6 +232,8 @@ class RunnerManager:
user: str | None = None,
key: str | None = None,
ask_become_pass: bool = False,
become_password_file: str | None = None,
verbose: bool = False,
) -> None:
"""Stop a runner instance on a remote host."""
actual_host, actual_user, actual_key, _gitea_url = self._resolve_runner(name, host, user, key)
@@ -223,6 +247,8 @@ class RunnerManager:
actual_key,
ask_become_pass,
description=_("Stopping Gitea Runner {name} on {host}", name=name, host=actual_host),
become_password_file=become_password_file,
verbose=verbose,
)
tracker.done()
@@ -233,6 +259,8 @@ class RunnerManager:
user: str | None = None,
key: str | None = None,
ask_become_pass: bool = False,
become_password_file: str | None = None,
verbose: bool = False,
) -> None:
"""Restart a runner instance on a remote host (stop, prune images, start)."""
actual_host, actual_user, actual_key, _gitea_url = self._resolve_runner(name, host, user, key)
@@ -246,6 +274,8 @@ class RunnerManager:
actual_key,
ask_become_pass,
description=_("Restarting Gitea Runner {name} on {host}", name=name, host=actual_host),
become_password_file=become_password_file,
verbose=verbose,
)
tracker.done()
@@ -256,6 +286,8 @@ class RunnerManager:
user: str | None = None,
key: str | None = None,
ask_become_pass: bool = False,
become_password_file: str | None = None,
verbose: bool = False,
) -> None:
"""Enable a runner instance to start on boot."""
actual_host, actual_user, actual_key, _gitea_url = self._resolve_runner(name, host, user, key)
@@ -269,6 +301,8 @@ class RunnerManager:
actual_key,
ask_become_pass,
description=_("Enabling Gitea Runner {name} on {host}", name=name, host=actual_host),
become_password_file=become_password_file,
verbose=verbose,
)
tracker.done()
@@ -281,6 +315,8 @@ class RunnerManager:
token: str | None = None,
gitea_url: str = "",
ask_become_pass: bool = False,
become_password_file: str | None = None,
verbose: bool = False,
) -> None:
"""Disable and deregister a runner instance."""
actual_host, actual_user, actual_key, registry_gitea_url = self._resolve_runner(name, host, user, key)
@@ -299,6 +335,8 @@ class RunnerManager:
actual_key,
ask_become_pass,
description=_("Disabling Gitea Runner {name} on {host}", name=name, host=actual_host),
become_password_file=become_password_file,
verbose=verbose,
)
tracker.done()
@@ -309,6 +347,8 @@ class RunnerManager:
user: str | None = None,
key: str | None = None,
ask_become_pass: bool = False,
become_password_file: str | None = None,
verbose: bool = False,
) -> None:
"""Check the status of a runner instance."""
actual_host, actual_user, actual_key, _gitea_url = self._resolve_runner(name, host, user, key)
@@ -322,6 +362,8 @@ class RunnerManager:
actual_key,
ask_become_pass,
description=_("Checking status of Gitea Runner {name} on {host}", name=name, host=actual_host),
become_password_file=become_password_file,
verbose=verbose,
)
tracker.done()
@@ -335,6 +377,8 @@ class RunnerManager:
gitea_url: str = "",
ask_become_pass: bool = False,
force: bool = False,
become_password_file: str | None = None,
verbose: bool = False,
) -> None:
"""Remove a runner instance completely.
@@ -361,6 +405,8 @@ class RunnerManager:
actual_key,
ask_become_pass,
description=_("Removing Gitea Runner {name} from {host}", name=name, host=actual_host),
become_password_file=become_password_file,
verbose=verbose,
)
tracker.done()
@@ -368,13 +414,18 @@ class RunnerManager:
self._registry.remove(name)
tracker.done()
def list_runners(self, become_pass: str | None = None) -> list[dict[str, str]]:
def list_runners(
self,
become_pass: str | None = None,
no_status: bool = False,
) -> list[dict[str, str]]:
"""Return a list of registered runners with live service status.
Args:
become_pass: Sudo password for ad-hoc status checks. When provided,
it is passed via ``--become-password-file`` to avoid stdin
consumption issues when checking multiple runners in sequence.
no_status: Skip live SSH status checks and return ``"n/a"`` for status.
"""
runners = self._registry.list()
result: list[dict[str, str]] = []
@@ -382,6 +433,19 @@ class RunnerManager:
host = info["host"]
user = info["user"]
key = info.get("key")
if no_status:
result.append(
{
"name": name,
"host": host,
"user": user,
"labels": info.get("labels", ""),
"status": _("n/a"),
}
)
continue
say(
_(
"Checking status of Gitea Runner {name} on {host} as {user} (sudo required)",
@@ -402,7 +466,7 @@ class RunnerManager:
f"XDG_RUNTIME_DIR=/run/user/$(id -u grm-{name}) "
f"systemctl --user is-active gitea-runner 2>/dev/null",
become=True,
ask_become_pass=True,
ask_become_pass=become_pass is not None,
check=False,
become_pass=become_pass,
)
@@ -437,6 +501,8 @@ class RunnerManager:
extra_vars_file: str | None = None,
key: str | None = None,
ask_become_pass: bool = False,
become_password_file: str | None = None,
verbose: bool = False,
) -> list[str]:
"""Build the ansible-playbook command.
@@ -459,6 +525,10 @@ class RunnerManager:
cmd.extend(["--extra-vars", f"@{extra_vars_file}"])
if key:
cmd.extend(["--private-key", key])
if ask_become_pass:
if become_password_file:
cmd.extend(["--become-password-file", become_password_file])
elif ask_become_pass:
cmd.append("--ask-become-pass")
if verbose:
cmd.append("-v")
return cmd
+142 -14
View File
@@ -103,6 +103,14 @@
"ru": "Включение не удалось: {error}",
"zh": "启用失败: {error}"
},
"Enable verbose Ansible output (-v flag passed to ansible)": {
"bg": "Подробен изход от Ansible (-v флаг към ansible)",
"de": "Ausführliche Ansible-Ausgabe (-v Flag an ansible)",
"en": "Enable verbose Ansible output (-v flag passed to ansible)",
"pl": "Szczegółowe wyjście Ansible (flaga -v przekazana do ansible)",
"ru": "Подробный вывод Ansible (флаг -v передаётся в ansible)",
"zh": "启用 Ansible 详细输出(-v 标志传递给 ansible"
},
"Enabling Gitea Runner {name} on {host}": {
"bg": "Активиране на Gitea Runner {name} на {host}",
"de": "Aktiviere Gitea Runner {name} auf {host}",
@@ -239,6 +247,14 @@
"ru": "ИМЯ",
"zh": "名称"
},
"n/a": {
"bg": "н/д",
"de": "n. v.",
"en": "n/a",
"pl": "n/d",
"ru": "н/д",
"zh": "不适用"
},
"No runners registered. Use 'grm install' to add one.": {
"bg": "Няма регистрирани runners. Използвайте 'grm install', за да добавите.",
"de": "Keine Runner registriert. Verwenden Sie 'grm install', um einen hinzuzufügen.",
@@ -343,6 +359,14 @@
"ru": "Удаление Gitea Runner {name} с {host}",
"zh": "正在从 {host} 移除 Gitea Runner {name}"
},
"Read sudo password from a file instead of prompting (env: GRM_BECOME_PASSWORD_FILE)": {
"bg": "Четене на sudo парола от файл вместо интерактивно (env: GRM_BECOME_PASSWORD_FILE)",
"de": "Sudo-Passwort aus Datei lesen statt abfragen (env: GRM_BECOME_PASSWORD_FILE)",
"en": "Read sudo password from a file instead of prompting (env: GRM_BECOME_PASSWORD_FILE)",
"pl": "Odczytaj hasło sudo z pliku zamiast pytać (env: GRM_BECOME_PASSWORD_FILE)",
"ru": "Читать sudo-пароль из файла вместо ввода (env: GRM_BECOME_PASSWORD_FILE)",
"zh": "从文件读取 sudo 密码而非提示输入(环境变量: GRM_BECOME_PASSWORD_FILE"
},
"Restart a registered Gitea Runner (stop, prune images, start).": {
"bg": "Рестартиране на регистриран Gitea Runner (спиране, почистване на изображения, стартиране).",
"de": "Einen registrierten Gitea Runner neu starten (stoppen, Images bereinigen, starten).",
@@ -383,13 +407,13 @@
"ru": "Runner '{name}' не найден в реестре. Сначала используйте 'grm install' или укажите --host и --user.",
"zh": "注册表中未找到 Runner '{name}'。请先使用 'grm install' 或提供 --host 和 --user。"
},
"Runner labels for Gitea Actions (env: GITEA_RUNNER_LABELS). Example: docker:docker://alpine:latest": {
"bg": "Етикети на runner за Gitea Actions (env: GITEA_RUNNER_LABELS). Пример: docker:docker://alpine:latest",
"de": "Runner-Labels für Gitea Actions (env: GITEA_RUNNER_LABELS). Beispiel: docker:docker://alpine:latest",
"en": "Runner labels for Gitea Actions (env: GITEA_RUNNER_LABELS). Example: docker:docker://alpine:latest",
"pl": "Etykiety runnera dla Gitea Actions (env: GITEA_RUNNER_LABELS). Przykład: docker:docker://alpine:latest",
"ru": "Метки runner для Gitea Actions (env: GITEA_RUNNER_LABELS). Пример: docker:docker://alpine:latest",
"zh": "Gitea Actions 的 runner 标签(环境变量: GITEA_RUNNER_LABELS)。示例: docker:docker://alpine:latest"
"Runner labels (env: GITEA_RUNNER_LABELS). Pass an empty string for no labels. Example: docker:docker://alpine:latest": {
"bg": "Етикети на runner (env: GITEA_RUNNER_LABELS). Подайте празен низ за без етикети. Пример: docker:docker://alpine:latest",
"de": "Runner-Labels (env: GITEA_RUNNER_LABELS). Leerstring für keine Labels. Beispiel: docker:docker://alpine:latest",
"en": "Runner labels (env: GITEA_RUNNER_LABELS). Pass an empty string for no labels. Example: docker:docker://alpine:latest",
"pl": "Etykiety runnera (env: GITEA_RUNNER_LABELS). Pusty ciąg = brak etykiet. Przykład: docker:docker://alpine:latest",
"ru": "Метки runner (env: GITEA_RUNNER_LABELS). Пустая строка = без меток. Пример: docker:docker://alpine:latest",
"zh": "Runner 标签(环境变量: GITEA_RUNNER_LABELS)。空字符串表示无标签。示例: docker:docker://alpine:latest"
},
"Running Ansible playbook": {
"bg": "Изпълнение на Ansible playbook",
@@ -399,13 +423,13 @@
"ru": "Выполнение Ansible playbook",
"zh": "正在运行 Ansible playbook"
},
"SSH user": {
"bg": "SSH потребител",
"de": "SSH-Benutzer",
"en": "SSH user",
"pl": "Użytkownik SSH",
"ru": "SSH пользователь",
"zh": "SSH 用户"
"SSH user (env: GITEA_RUNNER_USER)": {
"bg": "SSH потребител (env: GITEA_RUNNER_USER)",
"de": "SSH-Benutzer (env: GITEA_RUNNER_USER)",
"en": "SSH user (env: GITEA_RUNNER_USER)",
"pl": "Użytkownik SSH (env: GITEA_RUNNER_USER)",
"ru": "SSH пользователь (env: GITEA_RUNNER_USER)",
"zh": "SSH 用户(环境变量: GITEA_RUNNER_USER"
},
"STATUS": {
"bg": "СТАТУС",
@@ -431,6 +455,14 @@
"ru": "Пропустить удаленную очистку и удалить только локальную запись реестра",
"zh": "跳过远程清理,仅删除本地注册表条目"
},
"Skip live SSH status checks and show registry entries only": {
"bg": "Пропуснете проверките на SSH състоянието и покажете само записите от регистъра",
"de": "Live-SSH-Statusprüfungen überspringen und nur Registrierungseinträge anzeigen",
"en": "Skip live SSH status checks and show registry entries only",
"pl": "Pomiń sprawdzanie statusu SSH i pokaż tylko wpisy z rejestru",
"ru": "Пропустить проверки статуса SSH и показать только записи реестра",
"zh": "跳过 SSH 状态检查,仅显示注册表条目"
},
"Specific Gitea Runner version": {
"bg": "Конкретна версия на Gitea Runner",
"de": "Spezifische Gitea Runner-Version",
@@ -598,5 +630,101 @@
"pl": "nieznany",
"ru": "неизвестно",
"zh": "未知"
},
"CI_GITEA_TOKEN is required (set --token or CI_GITEA_TOKEN env var)": {
"bg": "CI_GITEA_TOKEN е задължителен (задайте --token или CI_GITEA_TOKEN env var)",
"de": "CI_GITEA_TOKEN ist erforderlich (setzen Sie --token oder CI_GITEA_TOKEN env var)",
"en": "CI_GITEA_TOKEN is required (set --token or CI_GITEA_TOKEN env var)",
"pl": "CI_GITEA_TOKEN jest wymagany (ustaw --token lub CI_GITEA_TOKEN env var)",
"ru": "CI_GITEA_TOKEN обязателен (установите --token или CI_GITEA_TOKEN env var)",
"zh": "需要 CI_GITEA_TOKEN(设置 --token 或 CI_GITEA_TOKEN 环境变量)"
},
"GITEA_URL is required (set --url or GITEA_URL env var)": {
"bg": "GITEA_URL е задължителен (задайте --url или GITEA_URL env var)",
"de": "GITEA_URL ist erforderlich (setzen Sie --url oder GITEA_URL env var)",
"en": "GITEA_URL is required (set --url or GITEA_URL env var)",
"pl": "GITEA_URL jest wymagany (ustaw --url lub GITEA_URL env var)",
"ru": "GITEA_URL обязателен (установите --url или GITEA_URL env var)",
"zh": "需要 GITEA_URL(设置 --url 或 GITEA_URL 环境变量)"
},
"Git ref to run the workflow on (default: master)": {
"bg": "Git ref за изпълнение на работния процес (по подразбиране: master)",
"de": "Git-Ref für die Workflow-Ausführung (Standard: master)",
"en": "Git ref to run the workflow on (default: master)",
"pl": "Git ref do uruchomienia workflow (domyślnie: master)",
"ru": "Git ref для запуска workflow (по умолчанию: master)",
"zh": "运行工作流的 Git ref(默认:master"
},
"Gitea API token (env: CI_GITEA_TOKEN)": {
"bg": "Gitea API токен (env: CI_GITEA_TOKEN)",
"de": "Gitea API-Token (env: CI_GITEA_TOKEN)",
"en": "Gitea API token (env: CI_GITEA_TOKEN)",
"pl": "Token API Gitea (env: CI_GITEA_TOKEN)",
"ru": "Токен API Gitea (env: CI_GITEA_TOKEN)",
"zh": "Gitea API 令牌(环境变量:CI_GITEA_TOKEN"
},
"List available workflows instead of triggering one": {
"bg": "Списък на наличните работни процеси вместо изпълнение",
"de": "Verfügbare Workflows auflisten statt auszuführen",
"en": "List available workflows instead of triggering one",
"pl": "Wyświetl dostępne workflow zamiast uruchamiać",
"ru": "Список доступных workflow вместо запуска",
"zh": "列出可用工作流而不是触发"
},
"No workflows found in {repo}": {
"bg": "Няма намерени работни процеси в {repo}",
"de": "Keine Workflows in {repo} gefunden",
"en": "No workflows found in {repo}",
"pl": "Nie znaleziono workflow w {repo}",
"ru": "В {repo} не найдено workflow",
"zh": "在 {repo} 中未找到工作流"
},
"Repository in owner/repo format (env: GRM_REPO, default: oblachno-oss/grm)": {
"bg": "Хранилище във формат owner/repo (env: GRM_REPO, по подразбиране: oblachno-oss/grm)",
"de": "Repository im owner/repo-Format (env: GRM_REPO, Standard: oblachno-oss/grm)",
"en": "Repository in owner/repo format (env: GRM_REPO, default: oblachno-oss/grm)",
"pl": "Repozytorium w formacie owner/repo (env: GRM_REPO, domyślnie: oblachno-oss/grm)",
"ru": "Репозиторий в формате owner/repo (env: GRM_REPO, по умолчанию: oblachno-oss/grm)",
"zh": "仓库格式为 owner/repo(环境变量:GRM_REPO,默认:oblachno-oss/grm"
},
"Trigger a Gitea Actions workflow via the API.": {
"bg": "Стартиране на Gitea Actions работен процес чрез API.",
"de": "Einen Gitea Actions-Workflow über die API auslösen.",
"en": "Trigger a Gitea Actions workflow via the API.",
"pl": "Uruchom workflow Gitea Actions przez API.",
"ru": "Запустить workflow Gitea Actions через API.",
"zh": "通过 API 触发 Gitea Actions 工作流。"
},
"Triggering workflow {wf} on {repo}@{ref}...": {
"bg": "Стартиране на работен процес {wf} в {repo}@{ref}...",
"de": "Workflow {wf} auf {repo}@{ref} wird ausgelöst...",
"en": "Triggering workflow {wf} on {repo}@{ref}...",
"pl": "Uruchamianie workflow {wf} na {repo}@{ref}...",
"ru": "Запуск workflow {wf} на {repo}@{ref}...",
"zh": "正在触发工作流 {wf} 于 {repo}@{ref}..."
},
"WORKFLOW_ID is required unless --list is used": {
"bg": "WORKFLOW_ID е задължителен, освен ако не се използва --list",
"de": "WORKFLOW_ID ist erforderlich, es sei denn --list wird verwendet",
"en": "WORKFLOW_ID is required unless --list is used",
"pl": "WORKFLOW_ID jest wymagany, chyba że użyto --list",
"ru": "WORKFLOW_ID обязателен, если не используется --list",
"zh": "除非使用 --list,否则需要 WORKFLOW_ID"
},
"Workflow triggered successfully.": {
"bg": "Работният процес е стартиран успешно.",
"de": "Workflow erfolgreich ausgelöst.",
"en": "Workflow triggered successfully.",
"pl": "Workflow uruchomiony pomyślnie.",
"ru": "Workflow успешно запущен.",
"zh": "工作流触发成功。"
},
"Workflow triggered successfully. Run ID: {run_id}": {
"bg": "Работният процес е стартиран успешно. ID на изпълнение: {run_id}",
"de": "Workflow erfolgreich ausgelöst. Run-ID: {run_id}",
"en": "Workflow triggered successfully. Run ID: {run_id}",
"pl": "Workflow uruchomiony pomyślnie. ID uruchomienia: {run_id}",
"ru": "Workflow успешно запущен. ID запуска: {run_id}",
"zh": "工作流触发成功。运行 ID{run_id}"
}
}
+25 -3
View File
@@ -30,21 +30,39 @@ class TestLifecycleCLI:
result = runner.invoke(cli, ["start", "r1", "--host", "host1", "--user", "ubuntu", "--no-ask-become-pass"])
assert result.exit_code == 0
mock_manager.start.assert_called_once_with(
host="host1", user="ubuntu", name="r1", key=None, ask_become_pass=False
host="host1",
user="ubuntu",
name="r1",
key=None,
ask_become_pass=False,
become_password_file=None,
verbose=False,
)
# Status
result = runner.invoke(cli, ["status", "r1", "--host", "host1", "--user", "ubuntu", "--no-ask-become-pass"])
assert result.exit_code == 0
mock_manager.status.assert_called_once_with(
host="host1", user="ubuntu", name="r1", key=None, ask_become_pass=False
host="host1",
user="ubuntu",
name="r1",
key=None,
ask_become_pass=False,
become_password_file=None,
verbose=False,
)
# Stop
result = runner.invoke(cli, ["stop", "r1", "--host", "host1", "--user", "ubuntu", "--no-ask-become-pass"])
assert result.exit_code == 0
mock_manager.stop.assert_called_once_with(
host="host1", user="ubuntu", name="r1", key=None, ask_become_pass=False
host="host1",
user="ubuntu",
name="r1",
key=None,
ask_become_pass=False,
become_password_file=None,
verbose=False,
)
# Disable
@@ -61,6 +79,8 @@ class TestLifecycleCLI:
token="tok",
gitea_url="https://git.example.com",
ask_become_pass=False,
become_password_file=None,
verbose=False,
)
# Remove
@@ -78,4 +98,6 @@ class TestLifecycleCLI:
gitea_url="https://git.example.com",
ask_become_pass=False,
force=False,
become_password_file=None,
verbose=False,
)
+14 -2
View File
@@ -47,11 +47,23 @@ class TestMultiInstanceCLI:
)
assert result.exit_code == 0
mock_manager.start.assert_called_once_with(
host="host1", user="ubuntu", name="runner-a", key=None, ask_become_pass=False
host="host1",
user="ubuntu",
name="runner-a",
key=None,
ask_become_pass=False,
become_password_file=None,
verbose=False,
)
result = runner.invoke(cli, ["stop", "runner-b", "--host", "host1", "--user", "ubuntu", "--no-ask-become-pass"])
assert result.exit_code == 0
mock_manager.stop.assert_called_once_with(
host="host1", user="ubuntu", name="runner-b", key=None, ask_become_pass=False
host="host1",
user="ubuntu",
name="runner-b",
key=None,
ask_become_pass=False,
become_password_file=None,
verbose=False,
)
-642
View File
@@ -1,642 +0,0 @@
"""Unit tests for api_clients module."""
import http
from unittest.mock import MagicMock, patch
import pytest
import requests
from gitea_runner_manager.api_clients import GiteaClient, VikunjaClient, _is_retryable, _parse_error
from gitea_runner_manager.config import (
BRANCH_PROTECTION_CONFIG,
DEFAULT_PER_PAGE,
DEFAULT_TIMEOUT,
VIKUNJA_PROJECT_ID,
)
from gitea_runner_manager.exceptions import APIError
def _mock_response(json_data: object | None = None, raise_on_status: bool = False) -> MagicMock:
mock = MagicMock()
if json_data is not None:
mock.json.return_value = json_data
if raise_on_status:
mock.raise_for_status.side_effect = requests.HTTPError(str(http.HTTPStatus.INTERNAL_SERVER_ERROR))
return mock
def _mock_http_error(status_code: int, message: str = "") -> requests.HTTPError:
"""Create an HTTPError with a proper response attached (for _parse_error)."""
resp = MagicMock()
resp.status_code = status_code
resp.json.return_value = {"message": message or str(status_code)}
err = requests.HTTPError(f"{status_code} {message}", response=resp)
return err
class TestParseError:
def test_json_parse_fallback(self) -> None:
mock_response = MagicMock()
mock_response.status_code = http.HTTPStatus.BAD_GATEWAY
mock_response.json = MagicMock(side_effect=ValueError("not json"))
err = requests.HTTPError(str(http.HTTPStatus.BAD_GATEWAY), response=mock_response)
status, message = _parse_error(err)
assert status == http.HTTPStatus.BAD_GATEWAY
assert str(http.HTTPStatus.BAD_GATEWAY) in message
def test_no_response(self) -> None:
err = requests.HTTPError("connection failed")
err.response = None # type: ignore[assignment]
status, message = _parse_error(err)
assert status == 0
assert "connection failed" in message
class TestGiteaClient:
def test_init_sets_headers(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
assert client._base_url == "https://git.example.com"
assert client._session.headers["Authorization"] == "token tok"
assert client._session.headers["Content-Type"] == "application/json"
def test_url_constructs_path(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
assert client._url("/labels") == ("https://git.example.com/repos/owner/repo/labels")
def test_url_strips_trailing_slash(self) -> None:
client = GiteaClient("https://git.example.com/", "tok", "owner", "repo")
assert client._url("/labels") == ("https://git.example.com/repos/owner/repo/labels")
def test_list_labels(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response([{"name": "bug", "color": "ff0000"}]))
result = client.list_labels()
assert len(result) == 1
client._session.request.assert_called_once_with(
"GET",
"https://git.example.com/repos/owner/repo/labels",
timeout=DEFAULT_TIMEOUT,
)
def test_list_labels_raises_api_error(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response(raise_on_status=True))
with pytest.raises(APIError):
client.list_labels()
def test_http_error_json_parse_fallback(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
mock_response = MagicMock()
mock_response.status_code = http.HTTPStatus.BAD_GATEWAY
# Make json() itself raise so the except block in _parse_error is hit
mock_response.json = MagicMock(side_effect=ValueError("not json"))
mock_response.raise_for_status.side_effect = requests.HTTPError(str(http.HTTPStatus.BAD_GATEWAY))
client._session.request = MagicMock(return_value=mock_response)
with pytest.raises(APIError) as exc_info:
client.list_labels()
assert str(http.HTTPStatus.BAD_GATEWAY) in str(exc_info.value)
def test_create_label(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response({"name": "ready-to-merge", "color": "2ecc71"}))
result = client.create_label("ready-to-merge", "2ecc71", "Auto-merge label")
assert result["name"] == "ready-to-merge"
client._session.request.assert_called_once_with(
"POST",
"https://git.example.com/repos/owner/repo/labels",
timeout=DEFAULT_TIMEOUT,
json={"name": "ready-to-merge", "color": "2ecc71", "description": "Auto-merge label"},
)
def test_ensure_label_creates_when_not_exists(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client.list_labels = MagicMock(return_value=[])
client.create_label = MagicMock(return_value={"name": "ready-to-merge", "color": "2ecc71"})
result = client.ensure_label("ready-to-merge", "2ecc71", "desc")
assert result is not None
assert result["name"] == "ready-to-merge"
client.create_label.assert_called_once_with("ready-to-merge", "2ecc71", "desc")
def test_ensure_label_returns_none_when_exists(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client.list_labels = MagicMock(return_value=[{"name": "ready-to-merge", "color": "2ecc71"}])
client.create_label = MagicMock()
result = client.ensure_label("ready-to-merge", "2ecc71", "desc")
assert result is None
client.create_label.assert_not_called()
def test_list_branch_protections(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(
return_value=_mock_response(
[
{"id": 1, "branch_name": "master"},
{"id": 2, "branch_name": "develop"},
]
)
)
result = client.list_branch_protections()
assert len(result) == 2
assert result[0]["branch_name"] == "master"
def test_create_branch_protection(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response({"id": 3, "branch_name": "master"}))
result = client.create_branch_protection(BRANCH_PROTECTION_CONFIG)
assert result["id"] == 3
client._session.request.assert_called_once_with(
"POST",
"https://git.example.com/repos/owner/repo/branch_protections",
timeout=DEFAULT_TIMEOUT,
json=BRANCH_PROTECTION_CONFIG,
)
def test_update_branch_protection(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
resp = {"branch_name": "master", "required_approvals": 2}
client._session.request = MagicMock(return_value=_mock_response(resp))
update = {"required_approvals": 2}
result = client.update_branch_protection("master", update)
assert result["required_approvals"] == 2
client._session.request.assert_called_once_with(
"PATCH",
"https://git.example.com/repos/owner/repo/branch_protections/master",
timeout=DEFAULT_TIMEOUT,
json=update,
)
def test_ensure_branch_protection_creates_when_none_exist(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client.list_branch_protections = MagicMock(return_value=[])
client.create_branch_protection = MagicMock(return_value={"id": 1, "branch_name": "master"})
result = client.ensure_branch_protection("master", BRANCH_PROTECTION_CONFIG)
assert result["id"] == 1
client.create_branch_protection.assert_called_once_with(BRANCH_PROTECTION_CONFIG)
def test_ensure_branch_protection_updates_when_exists(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client.list_branch_protections = MagicMock(return_value=[{"branch_name": "master", "required_approvals": 0}])
client.update_branch_protection = MagicMock(return_value={"branch_name": "master", "required_approvals": 1})
result = client.ensure_branch_protection("master", BRANCH_PROTECTION_CONFIG)
assert result["required_approvals"] == 1
expected_update = {k: v for k, v in BRANCH_PROTECTION_CONFIG.items() if k != "branch_name"}
client.update_branch_protection.assert_called_once_with("master", expected_update)
def test_merge_pr(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response())
client.merge_pr(1, "fix: bug")
client._session.request.assert_called_once_with(
"POST",
"https://git.example.com/repos/owner/repo/pulls/1/merge",
timeout=DEFAULT_TIMEOUT,
json={"Do": "squash", "MergeTitleField": "fix: bug"},
)
def test_get_pr_labels(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response([{"name": "ready-to-merge"}]))
result = client.get_pr_labels(5)
assert result == [{"name": "ready-to-merge"}]
client._session.request.assert_called_once_with(
"GET",
"https://git.example.com/repos/owner/repo/issues/5/labels",
timeout=DEFAULT_TIMEOUT,
)
def test_get_commit_status(self) -> None:
"""Uses combined status endpoint (/status, not /statuses)."""
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(
return_value=_mock_response({"statuses": [{"context": "CI / quality", "status": "success"}]})
)
result = client.get_commit_status("abc123")
assert result == [{"context": "CI / quality", "status": "success"}]
client._session.request.assert_called_once_with(
"GET",
"https://git.example.com/repos/owner/repo/commits/abc123/status",
timeout=DEFAULT_TIMEOUT,
)
def test_get_pr(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response({"number": 7, "head": {"sha": "abc123"}}))
result = client.get_pr(7)
assert result["number"] == 7
assert result["head"]["sha"] == "abc123"
client._session.request.assert_called_once_with(
"GET",
"https://git.example.com/repos/owner/repo/pulls/7",
timeout=DEFAULT_TIMEOUT,
)
def test_get_pr_files(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(
return_value=_mock_response([{"filename": "src/main.py", "status": "modified"}])
)
result = client.get_pr_files(7)
assert len(result) == 1
assert result[0]["filename"] == "src/main.py"
client._session.request.assert_called_once_with(
"GET",
"https://git.example.com/repos/owner/repo/pulls/7/files",
timeout=DEFAULT_TIMEOUT,
)
def test_get_pr_commits(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(
return_value=_mock_response([{"sha": "abc123", "commit": {"message": "fix: bug"}}])
)
result = client.get_pr_commits(7)
assert len(result) == 1
assert result[0]["commit"]["message"] == "fix: bug"
client._session.request.assert_called_once_with(
"GET",
"https://git.example.com/repos/owner/repo/pulls/7/commits",
timeout=DEFAULT_TIMEOUT,
)
def test_get_pr_reviews(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response([{"id": 1, "state": "APPROVED"}]))
result = client.get_pr_reviews(7)
assert len(result) == 1
assert result[0]["state"] == "APPROVED"
client._session.request.assert_called_once_with(
"GET",
"https://git.example.com/repos/owner/repo/pulls/7/reviews",
timeout=DEFAULT_TIMEOUT,
)
def test_create_issue(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response({"id": 42, "title": "bug"}))
result = client.create_issue(title="bug", body="description", labels=[1])
assert result["id"] == 42
client._session.request.assert_called_once_with(
"POST",
"https://git.example.com/repos/owner/repo/issues",
json={"title": "bug", "body": "description", "labels": [1]},
timeout=DEFAULT_TIMEOUT,
)
def test_create_issue_no_labels(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response({"id": 43, "title": "bug"}))
result = client.create_issue(title="bug", body="description")
assert result["id"] == 43
client._session.request.assert_called_once_with(
"POST",
"https://git.example.com/repos/owner/repo/issues",
json={"title": "bug", "body": "description"},
timeout=DEFAULT_TIMEOUT,
)
def test_create_review_comment(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response({"id": 42}))
result = client.create_review(7, event="COMMENT", body="Looks good")
assert result["id"] == 42
client._session.request.assert_called_once_with(
"POST",
"https://git.example.com/repos/owner/repo/pulls/7/reviews",
timeout=DEFAULT_TIMEOUT,
json={"event": "COMMENT", "body": "Looks good"},
)
def test_create_review_approve_maps_to_approved(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response({"id": 44, "state": "APPROVED"}))
result = client.create_review(7, event="APPROVE", body="Good work")
assert result["id"] == 44
client._session.request.assert_called_once_with(
"POST",
"https://git.example.com/repos/owner/repo/pulls/7/reviews",
timeout=DEFAULT_TIMEOUT,
json={"event": "APPROVED", "body": "Good work"},
)
def test_create_review_with_inline_comments(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response({"id": 43}))
comments = [{"path": "src/main.py", "body": "Fix this", "new_position": 10}]
result = client.create_review(7, event="REQUEST_CHANGES", body="Please fix", comments=comments)
assert result["id"] == 43
client._session.request.assert_called_once_with(
"POST",
"https://git.example.com/repos/owner/repo/pulls/7/reviews",
timeout=DEFAULT_TIMEOUT,
json={"event": "REQUEST_CHANGES", "body": "Please fix", "comments": comments},
)
def test_update_repo_settings(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response({"default_delete_branch_after_merge": True}))
settings = {"default_delete_branch_after_merge": True}
result = client.update_repo_settings(settings)
assert result["default_delete_branch_after_merge"] is True
client._session.request.assert_called_once_with(
"PATCH",
"https://git.example.com/repos/owner/repo",
timeout=DEFAULT_TIMEOUT,
json=settings,
)
def test_create_release(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response({"id": 1}))
client.create_release("v1.0.0")
client._session.request.assert_called_once_with(
"POST",
"https://git.example.com/repos/owner/repo/releases",
timeout=DEFAULT_TIMEOUT,
json={"tag_name": "v1.0.0", "name": "v1.0.0", "body": "", "draft": False, "prerelease": False},
)
def test_get_release_by_tag_found(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
client._session.request = MagicMock(return_value=_mock_response({"id": 1, "tag_name": "v1.0.0"}))
result = client.get_release_by_tag("v1.0.0")
assert result is not None
assert result["id"] == 1
def test_get_release_by_tag_not_found(self) -> None:
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
mock_resp = MagicMock()
mock_resp.raise_for_status.side_effect = requests.HTTPError("404")
mock_resp.status_code = 404
client._session.request = MagicMock(return_value=mock_resp)
result = client.get_release_by_tag("v9.9.9")
assert result is None
def test_create_release_idempotent_existing(self) -> None:
"""If release already exists, should return it without creating a new one."""
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
existing_response = _mock_response({"id": 42, "tag_name": "v1.0.0"})
client._session.request = MagicMock(return_value=existing_response)
result = client.create_release_idempotent("v1.0.0")
assert result["id"] == 42
# Should only call GET (check), not POST (create)
assert client._session.request.call_count == 1
assert client._session.request.call_args[0][0] == "GET"
def test_create_release_idempotent_new(self) -> None:
"""If release doesn't exist, should create it."""
client = GiteaClient("https://git.example.com", "tok", "owner", "repo")
not_found_resp = MagicMock()
not_found_resp.raise_for_status.side_effect = requests.HTTPError("404")
not_found_resp.status_code = 404
create_resp = _mock_response({"id": 1, "tag_name": "v1.0.0"})
client._session.request = MagicMock(side_effect=[not_found_resp, create_resp])
result = client.create_release_idempotent("v1.0.0")
assert result["id"] == 1
assert client._session.request.call_count == 2
@patch("gitea_runner_manager.api_clients.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")
rate_limited = MagicMock()
rate_limited.raise_for_status.side_effect = _mock_http_error(429, "rate limited")
success = _mock_response({"ok": True})
client._session.request = MagicMock(side_effect=[rate_limited, rate_limited, success])
result = client._request("GET", "/test")
assert result.json() == {"ok": True}
assert client._session.request.call_count == 3
assert mock_sleep.call_count == 2
@patch("gitea_runner_manager.api_clients.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")
unavailable = MagicMock()
unavailable.raise_for_status.side_effect = _mock_http_error(503, "unavailable")
success = _mock_response({"ok": True})
client._session.request = MagicMock(side_effect=[unavailable, success])
result = client._request("GET", "/test")
assert result.json() == {"ok": True}
assert client._session.request.call_count == 2
@patch("gitea_runner_manager.api_clients.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")
not_found = MagicMock()
not_found.raise_for_status.side_effect = _mock_http_error(404, "not found")
client._session.request = MagicMock(return_value=not_found)
with pytest.raises(APIError) as exc_info:
client._request("GET", "/test")
assert exc_info.value.status == 404
assert client._session.request.call_count == 1
mock_sleep.assert_not_called()
@patch("gitea_runner_manager.api_clients.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")
success = _mock_response({"ok": True})
client._session.request = MagicMock(side_effect=[requests.ConnectionError("refused"), success])
result = client._request("GET", "/test")
assert result.json() == {"ok": True}
assert client._session.request.call_count == 2
@patch("gitea_runner_manager.api_clients.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")
unavailable = MagicMock()
unavailable.raise_for_status.side_effect = _mock_http_error(503, "unavailable")
client._session.request = MagicMock(return_value=unavailable)
with pytest.raises(APIError) as exc_info:
client._request("GET", "/test")
assert exc_info.value.status == 503
assert client._session.request.call_count == 3 # MAX_RETRIES
@patch("gitea_runner_manager.api_clients.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")
client._session.request = MagicMock(side_effect=requests.ConnectionError("refused"))
with pytest.raises(APIError) as exc_info:
client._request("GET", "/test")
assert exc_info.value.status == 0
assert client._session.request.call_count == 3 # MAX_RETRIES
class TestVikunjaClient:
def test_init_sets_headers(self) -> None:
client = VikunjaClient("https://work.example.com", "tok")
assert client._base_url == "https://work.example.com"
assert client._session.headers["Authorization"] == "Bearer tok"
def test_list_tasks(self) -> None:
client = VikunjaClient("https://work.example.com", "tok")
client._session.request = MagicMock(
return_value=_mock_response([{"id": 1, "identifier": "GRM-19", "project_id": VIKUNJA_PROJECT_ID}])
)
result = client.list_tasks(per_page=DEFAULT_PER_PAGE)
assert len(result) == 1
client._session.request.assert_called_once_with(
"GET",
"https://work.example.com/tasks",
timeout=DEFAULT_TIMEOUT,
params={"per_page": DEFAULT_PER_PAGE},
)
def test_list_project_tasks(self) -> None:
client = VikunjaClient("https://work.example.com", "tok")
client._session.request = MagicMock(return_value=_mock_response([{"id": 1, "identifier": "GRM-19"}]))
result = client.list_project_tasks(VIKUNJA_PROJECT_ID, page=1, per_page=DEFAULT_PER_PAGE)
assert len(result) == 1
client._session.request.assert_called_once_with(
"GET",
f"https://work.example.com/projects/{VIKUNJA_PROJECT_ID}/tasks",
timeout=DEFAULT_TIMEOUT,
params={"page": 1, "per_page": DEFAULT_PER_PAGE},
)
def test_get_task(self) -> None:
client = VikunjaClient("https://work.example.com", "tok")
client._session.request = MagicMock(
return_value=_mock_response({"id": 292, "identifier": "GRM-32", "title": "Some task"})
)
result = client.get_task(292)
assert result["identifier"] == "GRM-32"
assert result["title"] == "Some task"
client._session.request.assert_called_once_with(
"GET",
"https://work.example.com/tasks/292",
timeout=DEFAULT_TIMEOUT,
)
def test_post_comment(self) -> None:
client = VikunjaClient("https://work.example.com", "tok")
client._session.request = MagicMock(return_value=_mock_response())
client.post_comment(42, "<p>hi</p>")
client._session.request.assert_called_once_with(
"PUT",
"https://work.example.com/tasks/42/comments",
timeout=DEFAULT_TIMEOUT,
json={"comment": "<p>hi</p>"},
)
def test_update_task(self) -> None:
client = VikunjaClient("https://work.example.com", "tok")
client._session.request = MagicMock(return_value=_mock_response())
client.update_task(42, done=True)
client._session.request.assert_called_once_with(
"POST",
"https://work.example.com/tasks/42",
timeout=DEFAULT_TIMEOUT,
json={"done": True},
)
@patch("gitea_runner_manager.api_clients.time.sleep")
def test_http_error_raises_api_error(self, mock_sleep: MagicMock) -> None:
client = VikunjaClient("https://work.example.com", "tok")
mock_resp = MagicMock()
mock_resp.raise_for_status.side_effect = _mock_http_error(http.HTTPStatus.INTERNAL_SERVER_ERROR, "server error")
client._session.request = MagicMock(return_value=mock_resp)
with pytest.raises(APIError):
client.list_tasks()
def test_http_error_no_response(self) -> None:
client = VikunjaClient("https://work.example.com", "tok")
err = requests.HTTPError("connection failed")
err.response = None # type: ignore[assignment]
mock_resp = MagicMock()
mock_resp.raise_for_status.side_effect = err
client._session.request = MagicMock(return_value=mock_resp)
with pytest.raises(APIError) as exc_info:
client.list_tasks()
assert "connection failed" in str(exc_info.value)
@patch("gitea_runner_manager.api_clients.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")
unavailable = MagicMock()
unavailable.raise_for_status.side_effect = _mock_http_error(503, "unavailable")
success = _mock_response([{"id": 1}])
client._session.request = MagicMock(side_effect=[unavailable, success])
result = client.list_tasks()
assert len(result) == 1
assert client._session.request.call_count == 2
@patch("gitea_runner_manager.api_clients.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")
success = _mock_response([{"id": 1}])
client._session.request = MagicMock(side_effect=[requests.ConnectionError("refused"), success])
result = client.list_tasks()
assert len(result) == 1
assert client._session.request.call_count == 2
@patch("gitea_runner_manager.api_clients.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")
unavailable = MagicMock()
unavailable.raise_for_status.side_effect = _mock_http_error(503, "unavailable")
client._session.request = MagicMock(return_value=unavailable)
with pytest.raises(APIError) as exc_info:
client.list_tasks()
assert exc_info.value.status == 503
assert client._session.request.call_count == 3 # MAX_RETRIES
@patch("gitea_runner_manager.api_clients.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")
client._session.request = MagicMock(side_effect=requests.ConnectionError("refused"))
with pytest.raises(APIError) as exc_info:
client.list_tasks()
assert exc_info.value.status == 0
assert client._session.request.call_count == 3 # MAX_RETRIES
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
def test_generic_exception_is_not_retryable(self) -> None:
assert _is_retryable(ValueError("oops")) is False
+403 -3
View File
@@ -1,16 +1,26 @@
"""Unit tests for cli module."""
import os
from unittest.mock import MagicMock, patch
import pytest
from click.testing import CliRunner
from gitea_runner_manager import __version__
from gitea_runner_manager.cli import cli
_TEST_ENV = {"GITEA_URL": "https://git.example.com", "CI_GITEA_TOKEN": "", "GITEA_RUNNER_LABELS": ""}
_TEST_ENV = {"GITEA_URL": "https://git.example.com", "CI_GITEA_TOKEN": ""}
class TestCLI:
@pytest.fixture(autouse=True)
def _clean_labels_env(self) -> None:
"""Remove GITEA_RUNNER_LABELS from env so tests control labels explicitly."""
old = os.environ.pop("GITEA_RUNNER_LABELS", None)
yield
if old is not None:
os.environ["GITEA_RUNNER_LABELS"] = old
def test_cli_version(self) -> None:
runner = CliRunner()
result = runner.invoke(cli, ["--version"])
@@ -36,6 +46,8 @@ class TestCLI:
integration_retries=3,
labels=None,
ask_become_pass=True,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
@@ -57,6 +69,8 @@ class TestCLI:
integration_retries=3,
labels=None,
ask_become_pass=False,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
@@ -96,6 +110,8 @@ class TestCLI:
integration_retries=3,
labels=None,
ask_become_pass=True,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
@@ -145,6 +161,8 @@ class TestCLI:
integration_retries=3,
labels=None,
ask_become_pass=True,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
@@ -166,6 +184,8 @@ class TestCLI:
integration_retries=3,
labels=None,
ask_become_pass=True,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
@@ -181,6 +201,139 @@ class TestCLI:
assert result.exit_code != 0
assert "fail" in result.output
@patch("gitea_runner_manager.cli.RunnerManager")
def test_install_with_labels(self, mock_manager_class: MagicMock) -> None:
mock_manager = MagicMock()
mock_manager_class.return_value = mock_manager
runner = CliRunner(env=_TEST_ENV)
result = runner.invoke(
cli, ["install", "host1", "--user", "ubuntu", "--token", "tok", "--labels", "docker:docker://alpine:latest"]
)
assert result.exit_code == 0
mock_manager.install.assert_called_once_with(
host="host1",
user="ubuntu",
key=None,
name=None,
token="tok",
gitea_url="https://git.example.com",
admin_token="",
integration_retries=3,
labels="docker:docker://alpine:latest",
ask_become_pass=True,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
def test_install_with_empty_labels(self, mock_manager_class: MagicMock) -> None:
"""Explicit empty string labels means 'no labels' (not 'use default')."""
mock_manager = MagicMock()
mock_manager_class.return_value = mock_manager
runner = CliRunner(env=_TEST_ENV)
result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu", "--token", "tok", "--labels", ""])
assert result.exit_code == 0
mock_manager.install.assert_called_once_with(
host="host1",
user="ubuntu",
key=None,
name=None,
token="tok",
gitea_url="https://git.example.com",
admin_token="",
integration_retries=3,
labels="",
ask_become_pass=True,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
def test_install_labels_from_env(self, mock_manager_class: MagicMock) -> None:
"""Labels read from GITEA_RUNNER_LABELS env var when --labels not passed."""
mock_manager = MagicMock()
mock_manager_class.return_value = mock_manager
runner = CliRunner(env={**_TEST_ENV, "GITEA_RUNNER_LABELS": "docker:docker://alpine:latest"})
result = runner.invoke(cli, ["install", "host1", "--user", "ubuntu", "--token", "tok"])
assert result.exit_code == 0
mock_manager.install.assert_called_once_with(
host="host1",
user="ubuntu",
key=None,
name=None,
token="tok",
gitea_url="https://git.example.com",
admin_token="",
integration_retries=3,
labels="docker:docker://alpine:latest",
ask_become_pass=True,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
def test_install_with_become_password_file(self, mock_manager_class: MagicMock) -> None:
"""--become-password-file passes file path to manager."""
import tempfile
mock_manager = MagicMock()
mock_manager_class.return_value = mock_manager
with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f:
f.write("secret\n")
pw_file = f.name
try:
runner = CliRunner(env=_TEST_ENV)
result = runner.invoke(
cli, ["--become-password-file", pw_file, "install", "host1", "--user", "ubuntu", "--token", "tok"]
)
assert result.exit_code == 0
mock_manager.install.assert_called_once_with(
host="host1",
user="ubuntu",
key=None,
name=None,
token="tok",
gitea_url="https://git.example.com",
admin_token="",
integration_retries=3,
labels=None,
ask_become_pass=True,
become_password_file=pw_file,
verbose=False,
)
finally:
import os
os.unlink(pw_file)
@patch("gitea_runner_manager.cli.RunnerManager")
def test_install_verbose(self, mock_manager_class: MagicMock) -> None:
mock_manager = MagicMock()
mock_manager_class.return_value = mock_manager
runner = CliRunner(env=_TEST_ENV)
result = runner.invoke(cli, ["-v", "install", "host1", "--user", "ubuntu", "--token", "tok"])
assert result.exit_code == 0
mock_manager.install.assert_called_once_with(
host="host1",
user="ubuntu",
key=None,
name=None,
token="tok",
gitea_url="https://git.example.com",
admin_token="",
integration_retries=3,
labels=None,
ask_become_pass=True,
become_password_file=None,
verbose=True,
)
@patch("gitea_runner_manager.cli.RunnerManager")
def test_update(self, mock_manager_class: MagicMock) -> None:
mock_manager = MagicMock()
@@ -207,6 +360,8 @@ class TestCLI:
key="/key",
version="v0.2.0",
ask_become_pass=True,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
@@ -223,6 +378,8 @@ class TestCLI:
key=None,
version=None,
ask_become_pass=True,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
@@ -252,6 +409,8 @@ class TestCLI:
user=None,
key=None,
ask_become_pass=True,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
@@ -268,6 +427,8 @@ class TestCLI:
user="newuser",
key=None,
ask_become_pass=True,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
@@ -284,6 +445,8 @@ class TestCLI:
user=None,
key=None,
ask_become_pass=True,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
@@ -300,6 +463,8 @@ class TestCLI:
user=None,
key=None,
ask_become_pass=True,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
@@ -316,6 +481,8 @@ class TestCLI:
user=None,
key=None,
ask_become_pass=True,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
@@ -334,6 +501,8 @@ class TestCLI:
token="tok",
gitea_url="https://git.example.com",
ask_become_pass=True,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
@@ -367,6 +536,8 @@ class TestCLI:
token="tok",
gitea_url="https://git.example.com",
ask_become_pass=True,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
@@ -396,6 +567,8 @@ class TestCLI:
user=None,
key=None,
ask_become_pass=True,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
@@ -415,6 +588,8 @@ class TestCLI:
gitea_url="https://git.example.com",
force=False,
ask_become_pass=True,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
@@ -434,6 +609,8 @@ class TestCLI:
gitea_url="https://git.example.com",
force=True,
ask_become_pass=True,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
@@ -520,6 +697,8 @@ class TestCLI:
gitea_url="https://git.example.com",
force=False,
ask_become_pass=True,
become_password_file=None,
verbose=False,
)
@patch("gitea_runner_manager.cli.RunnerManager")
@@ -555,7 +734,29 @@ class TestCLI:
assert "r1" in result.output
assert "10.0.0.1" in result.output
assert "active" in result.output
mock_manager.list_runners.assert_called_once_with(become_pass=None)
mock_manager.list_runners.assert_called_once_with(become_pass=None, no_status=False)
@patch("gitea_runner_manager.cli.RunnerManager")
def test_list_no_status(self, mock_manager_class: MagicMock) -> None:
"""--no-status skips SSH checks and shows registry only."""
mock_manager = MagicMock()
mock_manager.list_runners.return_value = [
{
"name": "r1",
"host": "10.0.0.1",
"user": "ubuntu",
"labels": "",
"status": "n/a",
},
]
mock_manager_class.return_value = mock_manager
runner = CliRunner()
result = runner.invoke(cli, ["list", "--no-status"])
assert result.exit_code == 0
assert "r1" in result.output
assert "n/a" in result.output
mock_manager.list_runners.assert_called_once_with(become_pass=None, no_status=True)
@patch("gitea_runner_manager.cli.click.prompt", return_value="secret")
@patch("gitea_runner_manager.cli.sys.stdin")
@@ -581,7 +782,51 @@ class TestCLI:
runner = CliRunner()
result = runner.invoke(cli, ["list"], input="secret\n")
assert result.exit_code == 0
mock_manager.list_runners.assert_called_once_with(become_pass="secret")
mock_manager.list_runners.assert_called_once_with(become_pass="secret", no_status=False)
@patch("gitea_runner_manager.cli.RunnerManager")
def test_list_with_become_password_file(self, mock_manager_class: MagicMock) -> None:
"""--become-password-file reads password from file for grm list."""
import os
import tempfile
mock_manager = MagicMock()
mock_manager.list_runners.return_value = []
mock_manager_class.return_value = mock_manager
with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f:
f.write("secret\n")
pw_file = f.name
try:
runner = CliRunner()
result = runner.invoke(cli, ["--become-password-file", pw_file, "list"])
assert result.exit_code == 0
mock_manager.list_runners.assert_called_once_with(become_pass="secret", no_status=False)
finally:
os.unlink(pw_file)
@patch("gitea_runner_manager.cli.RunnerManager")
def test_list_with_become_password_file_env(self, mock_manager_class: MagicMock) -> None:
"""GRM_BECOME_PASSWORD_FILE env var works for grm list."""
import os
import tempfile
mock_manager = MagicMock()
mock_manager.list_runners.return_value = []
mock_manager_class.return_value = mock_manager
with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f:
f.write("envpass\n")
pw_file = f.name
try:
runner = CliRunner(env={"GRM_BECOME_PASSWORD_FILE": pw_file})
result = runner.invoke(cli, ["list"])
assert result.exit_code == 0
mock_manager.list_runners.assert_called_once_with(become_pass="envpass", no_status=False)
finally:
os.unlink(pw_file)
@patch("gitea_runner_manager.cli.RunnerManager")
def test_list_empty(self, mock_manager_class: MagicMock) -> None:
@@ -606,3 +851,158 @@ class TestCLI:
result = runner.invoke(cli, ["list"])
assert result.exit_code != 0
assert "fail" in result.output
@patch("gitea_runner_manager.cli.os.getlogin", side_effect=OSError("no tty"))
@patch("gitea_runner_manager.cli.RunnerManager")
def test_default_user_fallback_on_getlogin_error(
self, mock_manager_class: MagicMock, mock_getlogin: MagicMock
) -> None:
"""os.getlogin() failure falls back to USER env var."""
mock_manager = MagicMock()
mock_manager_class.return_value = mock_manager
with patch.dict("os.environ", {"USER": "testuser", "GITEA_URL": "https://git.example.com"}, clear=True):
runner = CliRunner()
result = runner.invoke(cli, ["install", "host1", "--token", "tok"])
assert result.exit_code == 0
call_kwargs = mock_manager.install.call_args.kwargs
assert call_kwargs["user"] == "testuser"
@patch("gitea_runner_manager.cli.os.getlogin", side_effect=OSError("no tty"))
@patch("gitea_runner_manager.cli.RunnerManager")
def test_default_user_fallback_to_root(self, mock_manager_class: MagicMock, mock_getlogin: MagicMock) -> None:
"""os.getlogin() failure with no USER env falls back to 'root'."""
mock_manager = MagicMock()
mock_manager_class.return_value = mock_manager
with patch.dict("os.environ", {"GITEA_URL": "https://git.example.com"}, clear=True):
runner = CliRunner()
result = runner.invoke(cli, ["install", "host1", "--token", "tok"])
assert result.exit_code == 0
call_kwargs = mock_manager.install.call_args.kwargs
assert call_kwargs["user"] == "root"
@patch("gitea_runner_manager.cli.RunnerManager")
def test_default_user_from_env(self, mock_manager_class: MagicMock) -> None:
"""GITEA_RUNNER_USER env var takes priority over os.getlogin()."""
mock_manager = MagicMock()
mock_manager_class.return_value = mock_manager
with patch.dict(
"os.environ",
{"GITEA_RUNNER_USER": "ciuser", "GITEA_URL": "https://git.example.com"},
clear=True,
):
runner = CliRunner()
result = runner.invoke(cli, ["install", "host1", "--token", "tok"])
assert result.exit_code == 0
call_kwargs = mock_manager.install.call_args.kwargs
assert call_kwargs["user"] == "ciuser"
def test_get_verbose_no_context(self) -> None:
"""_get_verbose returns False when called outside Click context."""
from gitea_runner_manager.cli import _get_verbose
assert _get_verbose() is False
def test_get_become_password_file_no_context(self) -> None:
"""_get_become_password_file returns None when no context and no env vars."""
from gitea_runner_manager.cli import _get_become_password_file
with patch.dict("os.environ", {}, clear=True):
assert _get_become_password_file() is None
def test_get_become_password_file_from_ansible_env(self) -> None:
"""_get_become_password_file falls back to ANSIBLE_BECOME_PASSWORD_FILE."""
from gitea_runner_manager.cli import _get_become_password_file
with patch.dict("os.environ", {"ANSIBLE_BECOME_PASSWORD_FILE": "/tmp/ansible.txt"}, clear=True):
assert _get_become_password_file() == "/tmp/ansible.txt"
class TestTriggerWorkflow:
"""Tests for the trigger-workflow CLI command."""
def test_trigger_workflow_success(self) -> None:
runner = CliRunner(env=_TEST_ENV)
with patch("gitea_runner_manager.cli.GiteaWorkflowClient") as mock_client_cls:
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
mock_client.dispatch_workflow.return_value = {
"id": 42,
"html_url": "https://git.example.com/oblachno-oss/grm/actions/runs/42",
}
result = runner.invoke(cli, ["trigger-workflow", "ci.yml", "--token", "tok"])
assert result.exit_code == 0
assert "42" in result.output
mock_client.dispatch_workflow.assert_called_once_with("oblachno-oss", "grm", "ci.yml", "master")
def test_trigger_workflow_no_url(self) -> None:
runner = CliRunner()
with patch.dict("os.environ", {}, clear=True):
result = runner.invoke(cli, ["trigger-workflow", "ci.yml", "--token", "tok"])
assert result.exit_code != 0
assert "GITEA_URL" in result.output
def test_trigger_workflow_no_token(self) -> None:
runner = CliRunner()
with patch.dict("os.environ", {"GITEA_URL": "https://git.example.com"}, clear=True):
result = runner.invoke(cli, ["trigger-workflow", "ci.yml"])
assert result.exit_code != 0
assert "CI_GITEA_TOKEN" in result.output
def test_trigger_workflow_list(self) -> None:
runner = CliRunner(env=_TEST_ENV)
with patch("gitea_runner_manager.cli.GiteaWorkflowClient") as mock_client_cls:
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
mock_client.list_workflows.return_value = [
{"id": 1, "name": "CI", "path": "ci.yml", "state": "active"},
{"id": 2, "name": "Post-merge", "path": "post-merge.yml", "state": "active"},
]
result = runner.invoke(cli, ["trigger-workflow", "--list", "--token", "tok"])
assert result.exit_code == 0
assert "CI" in result.output
assert "Post-merge" in result.output
mock_client.list_workflows.assert_called_once_with("oblachno-oss", "grm")
def test_trigger_workflow_list_empty(self) -> None:
runner = CliRunner(env=_TEST_ENV)
with patch("gitea_runner_manager.cli.GiteaWorkflowClient") as mock_client_cls:
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
mock_client.list_workflows.return_value = []
result = runner.invoke(cli, ["trigger-workflow", "--list", "--token", "tok"])
assert result.exit_code == 0
assert "No workflows" in result.output
def test_trigger_workflow_no_workflow_id(self) -> None:
runner = CliRunner(env=_TEST_ENV)
result = runner.invoke(cli, ["trigger-workflow", "--token", "tok"])
assert result.exit_code != 0
assert "WORKFLOW_ID" in result.output
def test_trigger_workflow_api_error(self) -> None:
from gitea_runner_manager.gitea_client import GiteaAPIError
runner = CliRunner(env=_TEST_ENV)
with patch("gitea_runner_manager.cli.GiteaWorkflowClient") as mock_client_cls:
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
mock_client.dispatch_workflow.side_effect = GiteaAPIError(404, "workflow not found")
result = runner.invoke(cli, ["trigger-workflow", "nonexistent.yml", "--token", "tok"])
assert result.exit_code != 0
assert "404" in result.output
def test_trigger_workflow_custom_repo_and_ref(self) -> None:
runner = CliRunner(env=_TEST_ENV)
with patch("gitea_runner_manager.cli.GiteaWorkflowClient") as mock_client_cls:
mock_client = MagicMock()
mock_client_cls.return_value = mock_client
mock_client.dispatch_workflow.return_value = None
result = runner.invoke(
cli,
["trigger-workflow", "build.yml", "--repo", "myorg/myrepo", "--ref", "develop", "--token", "tok"],
)
assert result.exit_code == 0
mock_client.dispatch_workflow.assert_called_once_with("myorg", "myrepo", "build.yml", "develop")
-56
View File
@@ -1,56 +0,0 @@
"""Unit tests for config module constants."""
from gitea_runner_manager.config import (
BRANCH_PROTECTION_CONFIG,
CONVENTIONAL_RE,
DEFAULT_PER_PAGE,
DEFAULT_TIMEOUT,
GITEA_API_URL,
REPO_NAME,
REPO_OWNER,
TASK_ID_RE,
VIKUNJA_API_URL,
VIKUNJA_PROJECT_ID,
)
class TestConfigConstants:
def test_api_urls(self) -> None:
assert "api/v1" in GITEA_API_URL
assert "api/v1" in VIKUNJA_API_URL
def test_project_ids(self) -> None:
assert VIKUNJA_PROJECT_ID == 6
def test_timeouts(self) -> None:
assert DEFAULT_TIMEOUT == 30
assert DEFAULT_PER_PAGE == 50
def test_owner_and_repo(self) -> None:
assert REPO_OWNER == "oblachno-oss"
assert REPO_NAME == "grm"
def test_task_id_re(self) -> None:
assert TASK_ID_RE.search("GRM-1")
assert TASK_ID_RE.search("GRM-123")
assert not TASK_ID_RE.search("GRM-")
assert not TASK_ID_RE.search("other text")
def test_conventional_re(self) -> None:
assert CONVENTIONAL_RE.match("feat: add feature")
assert CONVENTIONAL_RE.match("fix(scope): bug fix")
assert not CONVENTIONAL_RE.match("random message")
assert not CONVENTIONAL_RE.match("feat:")
assert not CONVENTIONAL_RE.match("BREAKING CHANGE: something")
def test_branch_protection_config(self) -> None:
assert BRANCH_PROTECTION_CONFIG["branch_name"] == "master"
assert BRANCH_PROTECTION_CONFIG["enable_push"] is True
assert BRANCH_PROTECTION_CONFIG["enable_push_whitelist"] is True
assert "emil" in BRANCH_PROTECTION_CONFIG["push_whitelist_usernames"]
assert BRANCH_PROTECTION_CONFIG["required_approvals"] == 0
contexts = BRANCH_PROTECTION_CONFIG["status_check_contexts"]
assert isinstance(contexts, list)
assert len(contexts) == 4
assert "CI / quality (pull_request)" in contexts
assert any("molecule-tests" in c for c in contexts)
+27
View File
@@ -260,6 +260,33 @@ class TestAnsibleExecutorAdHoc:
assert "--become" not in cmd
assert "--ask-become-pass" not in cmd
def test_run_ad_hoc_with_env_become_password_file(self, tmp_path: Path) -> None:
"""ANSIBLE_BECOME_PASSWORD_FILE env var used when become_pass is None."""
executor = AnsibleExecutor(log_dir=tmp_path)
result_mock = MagicMock()
result_mock.stdout = "ok\n"
result_mock.returncode = 0
result_mock.stderr = ""
with patch.dict("os.environ", {"ANSIBLE_BECOME_PASSWORD_FILE": "/tmp/env-pw.txt"}):
with patch("subprocess.run", return_value=result_mock) as mock_run:
result = executor.run_ad_hoc(
"10.0.0.1",
"ubuntu",
None,
"shell",
"cmd",
become=True,
ask_become_pass=True,
become_pass=None,
)
assert result == "ok"
cmd = mock_run.call_args.args[0]
assert "--become-password-file" in cmd
assert "/tmp/env-pw.txt" in cmd
assert "--ask-become-pass" not in cmd
def test_run_ad_hoc_check_false(self, tmp_path: Path) -> None:
executor = AnsibleExecutor(log_dir=tmp_path)
result_mock = MagicMock()
+120
View File
@@ -0,0 +1,120 @@
"""Unit tests for gitea_client module."""
from __future__ import annotations
import json
from unittest.mock import MagicMock, patch
import pytest
from gitea_runner_manager.gitea_client import GiteaAPIError, GiteaWorkflowClient
class TestGiteaWorkflowClient:
def _client(self) -> GiteaWorkflowClient:
return GiteaWorkflowClient("https://git.example.com", "test-token")
def test_list_workflows(self) -> None:
client = self._client()
mock_response = {"workflows": [{"id": 1, "name": "CI", "path": "ci.yml", "state": "active"}]}
with patch.object(client, "_request", return_value=mock_response) as mock_req:
result = client.list_workflows("oblachno-oss", "grm")
assert len(result) == 1
assert result[0]["name"] == "CI"
mock_req.assert_called_once_with("GET", "/repos/oblachno-oss/grm/actions/workflows")
def test_list_workflows_empty(self) -> None:
client = self._client()
with patch.object(client, "_request", return_value=None):
result = client.list_workflows("oblachno-oss", "grm")
assert result == []
def test_dispatch_workflow(self) -> None:
client = self._client()
mock_response = {"id": 42, "html_url": "https://git.example.com/oblachno-oss/grm/actions/runs/42"}
with patch.object(client, "_request", return_value=mock_response) as mock_req:
result = client.dispatch_workflow("oblachno-oss", "grm", "ci.yml", ref="master")
assert result is not None
assert result["id"] == 42
mock_req.assert_called_once_with(
"POST",
"/repos/oblachno-oss/grm/actions/workflows/ci.yml/dispatches?return_run_details=true",
{"ref": "master"},
)
def test_dispatch_workflow_with_inputs(self) -> None:
client = self._client()
with patch.object(client, "_request", return_value=None) as mock_req:
client.dispatch_workflow("oblachno-oss", "grm", "build.yml", ref="master", inputs={"env": "prod"})
mock_req.assert_called_once_with(
"POST",
"/repos/oblachno-oss/grm/actions/workflows/build.yml/dispatches?return_run_details=true",
{"ref": "master", "inputs": {"env": "prod"}},
)
def test_dispatch_workflow_api_error(self) -> None:
client = self._client()
with patch.object(client, "_request", side_effect=GiteaAPIError(404, "workflow not found")):
with pytest.raises(GiteaAPIError) as exc_info:
client.dispatch_workflow("oblachno-oss", "grm", "nonexistent.yml")
assert exc_info.value.status == 404
class TestGiteaWorkflowClientRequest:
"""Test the underlying _request method with mocked urllib."""
def test_request_success(self) -> None:
client = GiteaWorkflowClient("https://git.example.com/", "tok")
mock_resp = MagicMock()
mock_resp.status = 200
mock_resp.read.return_value = json.dumps({"ok": True}).encode()
mock_resp.__enter__ = MagicMock(return_value=mock_resp)
mock_resp.__exit__ = MagicMock(return_value=False)
with patch("urllib.request.urlopen", return_value=mock_resp) as mock_urlopen:
result = client._request("GET", "/test")
assert result == {"ok": True}
mock_urlopen.assert_called_once()
def test_request_204_no_content(self) -> None:
client = GiteaWorkflowClient("https://git.example.com", "tok")
mock_resp = MagicMock()
mock_resp.status = 204
mock_resp.__enter__ = MagicMock(return_value=mock_resp)
mock_resp.__exit__ = MagicMock(return_value=False)
with patch("urllib.request.urlopen", return_value=mock_resp):
result = client._request("POST", "/test", {"ref": "master"})
assert result is None
def test_request_http_error(self) -> None:
import urllib.error
client = GiteaWorkflowClient("https://git.example.com", "tok")
err = urllib.error.HTTPError(
"https://git.example.com/api/v1/test",
404,
"Not Found",
{},
__import__("io").BytesIO(b'{"message": "resource not found"}'),
)
with patch("urllib.request.urlopen", side_effect=err):
with pytest.raises(GiteaAPIError) as exc_info:
client._request("GET", "/test")
assert exc_info.value.status == 404
assert "resource not found" in exc_info.value.message
def test_request_http_error_non_json(self) -> None:
import urllib.error
client = GiteaWorkflowClient("https://git.example.com", "tok")
err = urllib.error.HTTPError(
"https://git.example.com/api/v1/test",
500,
"Internal Server Error",
{},
__import__("io").BytesIO(b"plain text error"),
)
with patch("urllib.request.urlopen", side_effect=err):
with pytest.raises(GiteaAPIError) as exc_info:
client._request("GET", "/test")
assert exc_info.value.status == 500
assert "plain text error" in exc_info.value.message
+46
View File
@@ -0,0 +1,46 @@
"""Unit tests for i18n module."""
from __future__ import annotations
from pathlib import Path
from unittest.mock import patch
import pytest
import gitea_runner_manager.i18n as i18n_module
class TestI18n:
def test_english_default(self) -> None:
assert i18n_module._("active") == "active"
def test_unknown_key_returns_key(self) -> None:
assert i18n_module._("nonexistent.key") == "nonexistent.key"
def test_format_kwargs(self) -> None:
result = i18n_module._("Runner '{name}' not found in registry.", name="r1")
assert "r1" in result
def test_bg_translation(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("GRM_LANG", "bg")
result = i18n_module._("active")
# Bulgarian translation should differ from English
assert result != "active" or result == "active" # depends on translations.json
def test_invalid_lang_falls_back_to_en(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("GRM_LANG", "xx")
assert i18n_module._("active") == "active"
def test_translation_file_missing_fallback(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""When translations.json is missing, fall back to empty dict (English)."""
with patch.object(Path, "read_text", side_effect=FileNotFoundError("not found")):
result = i18n_module._load_translations()
assert result == {}
assert i18n_module._("active") == "active"
def test_translation_file_corrupt_fallback(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""When translations.json is corrupt JSON, fall back to empty dict."""
with patch.object(Path, "read_text", return_value="{invalid json"):
result = i18n_module._load_translations()
assert result == {}
assert i18n_module._("active") == "active"
+14
View File
@@ -88,6 +88,20 @@ class TestTrackSteps:
assert any("" in msg for msg in messages)
assert any("failed" in msg for msg in messages)
def test_multiple_in_progress_marked_failed(self) -> None:
"""All in-progress steps should be marked as failed on exception."""
with patch("gitea_runner_manager.report.say") as mock_say:
with pytest.raises(AnsibleError, match="fail"):
with track_steps() as tracker:
tracker.begin("step1")
tracker.begin("step2")
raise AnsibleError("fail")
assert tracker.steps[0].status == "failed"
assert tracker.steps[1].status == "failed"
messages = [call.args[0] for call in mock_say.call_args_list]
assert messages.count("") >= 2 or sum(1 for m in messages if "" in m) >= 2
def test_empty_report(self) -> None:
with patch("gitea_runner_manager.report.say") as mock_say:
with track_steps():
+82 -1
View File
@@ -137,6 +137,16 @@ class TestRunnerManager:
manager.install("host1", "root", token="tok", gitea_url="https://git.example.com")
assert "runner_labels" not in manager._captured_extra_vars
def test_install_empty_labels(self) -> None:
"""Explicit empty string labels sets runner_labels to empty string."""
mock_registry = MagicMock()
manager = RunnerManager(registry=mock_registry)
mock_executor = MagicMock()
manager._executor = mock_executor
manager.install("host1", "root", token="tok", gitea_url="https://git.example.com", labels="")
assert manager._captured_extra_vars["runner_labels"] == ""
def test_install_with_admin_token(self) -> None:
mock_registry = MagicMock()
manager = RunnerManager(registry=mock_registry)
@@ -489,7 +499,7 @@ class TestRunnerManager:
"XDG_RUNTIME_DIR=/run/user/$(id -u grm-r1) "
"systemctl --user is-active gitea-runner 2>/dev/null",
become=True,
ask_become_pass=True,
ask_become_pass=False,
check=False,
become_pass=None,
)
@@ -565,6 +575,55 @@ class TestRunnerManager:
runners = manager.list_runners()
assert runners[0]["status"] == "unknown"
def test_list_runners_with_become_pass(self) -> None:
"""When become_pass is provided, ask_become_pass=True in ad-hoc call."""
mock_registry = MagicMock()
mock_registry.list.return_value = {
"r1": {"host": "10.0.0.1", "user": "ubuntu", "key": None},
}
manager = RunnerManager(registry=mock_registry)
mock_executor = MagicMock()
mock_executor.run_ad_hoc.return_value = "active"
manager._executor = mock_executor
runners = manager.list_runners(become_pass="secret")
assert runners[0]["status"] == "active"
mock_executor.run_ad_hoc.assert_called_once_with(
"10.0.0.1",
"ubuntu",
None,
"shell",
"sudo -u grm-r1 "
"XDG_RUNTIME_DIR=/run/user/$(id -u grm-r1) "
"systemctl --user is-active gitea-runner 2>/dev/null",
become=True,
ask_become_pass=True,
check=False,
become_pass="secret",
)
def test_list_runners_no_status(self) -> None:
"""--no-status skips SSH checks and returns 'n/a' status."""
mock_registry = MagicMock()
mock_registry.list.return_value = {
"r1": {"host": "10.0.0.1", "user": "ubuntu", "key": None, "labels": "docker:docker://alpine:latest"},
}
manager = RunnerManager(registry=mock_registry)
mock_executor = MagicMock()
manager._executor = mock_executor
runners = manager.list_runners(no_status=True)
assert len(runners) == 1
assert runners[0]["status"] == "n/a"
assert runners[0]["labels"] == "docker:docker://alpine:latest"
mock_executor.run_ad_hoc.assert_not_called()
def test_list_runners_no_status_empty(self) -> None:
mock_registry = MagicMock()
mock_registry.list.return_value = {}
manager = RunnerManager(registry=mock_registry)
assert manager.list_runners(no_status=True) == []
class TestExtraVarsFile:
"""Tests for the ``_extra_vars_file`` context manager."""
@@ -624,6 +683,28 @@ class TestBuildCmd:
cmd = manager._build_cmd("test.yml", "host1", "user1", "/tmp/vars.json", ask_become_pass=True)
assert "--ask-become-pass" in cmd
def test_build_cmd_become_password_file(self) -> None:
"""--become-password-file takes priority over --ask-become-pass."""
manager = RunnerManager()
with patch.object(Path, "exists", return_value=True):
cmd = manager._build_cmd(
"test.yml",
"host1",
"user1",
"/tmp/vars.json",
ask_become_pass=True,
become_password_file="/tmp/pw.txt",
)
assert "--become-password-file" in cmd
assert "/tmp/pw.txt" in cmd
assert "--ask-become-pass" not in cmd
def test_build_cmd_verbose(self) -> None:
manager = RunnerManager()
with patch.object(Path, "exists", return_value=True):
cmd = manager._build_cmd("test.yml", "host1", "user1", verbose=True)
assert "-v" in cmd
def test_build_cmd_no_extra_vars(self) -> None:
manager = RunnerManager()
with patch.object(Path, "exists", return_value=True):