oblachno-oss/grm
1
0
Fork 0
Code Issues Pull Requests Actions 14 Packages Projects Releases Wiki Activity

GRM-32: fix: security, dead code, idempotence, and documentation cleanup

Browse Source
This commit is contained in:
emil
2026-06-21 00:14:31 +00:00
parent a676adb025
commit 1717d55013
20 changed files with 351 additions and 476 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitea/workflows/ci.yml
+1
View File
@@ -2,6 +2,7 @@ name: CI
on:
pull_request:
types: [opened, synchronize]
push:
branches: [master]
workflow_dispatch:
AGENTS.md
+51
View File
@@ -0,0 +1,51 @@
# AGENTS.md — Project Conventions for GRM
## Build & Test Commands
```bash
make setup # Create venv, install deps, set up hooks
make lint-all # ruff + pyright + bandit + ansible-lint + checkmake
make pytest-cov # Unit tests with 100% coverage enforcement
make test-unit # Unit tests without coverage
make molecule # All 6 scenarios on Ubuntu 22.04
make molecule-all # All 6 scenarios on all 4 supported OSes
make test-all # pytest-cov + molecule
```
## Architecture
- **Python CLI** (`src/gitea_runner_manager/`) — Click-based CLI that delegates to Ansible
- **Ansible Role** (`ansible/roles/gitea-runner/`) — Idempotent role for rootless Docker runner setup
- **CI Scripts** (`scripts/`) — Automation for auto-merge, post-merge, publishing, molecule distribution
## Key Conventions
- Python 3.12+ required (ruff/pyright target `py312`)
- 100% test coverage required (`--cov-fail-under=100`)
- Conventional commits on feature branches (no `GRM-N:` prefix)
- `GRM-N:` prefix on master branch (added by auto-merge)
- Branch names must include `GRM-N` task ID
- Line length: 120 chars
- Secrets are passed via temp JSON files, never on the command line (CWE-214)
## Ansible Role Structure
```
main.yml → systemd_check → user_setup → rootless_docker → install_runner → prune → integration_test
```
- `install_runner.yml` handles: download, config, validate, register, service
- `main.yml` handles: prune, integration_test (NOT install_runner — avoids duplicates)
- All `systemctl --user` tasks must be guarded by `docker_rootless_setup`
- All template creation tasks must be guarded by `docker_rootless_setup`
## Molecule Scenarios
6 scenarios: `default`, `multi-instance`, `lifecycle`, `template-content`, `deregister`, `update`
4 platforms: `ubuntu-2204`, `ubuntu-2404`, `debian-12`, `archlinux`
Platform list is defined in `scripts/distribute_molecule.py` (single source of truth)
## Known Issues
- `ansible-lint` may warn about `command-instead-of-module` for `systemctl --user` calls — this is expected (systemd module doesn't support user services) and skipped in `.ansible-lint`
- Molecule Docker driver may print "Event loop is closed" warnings on interrupt — harmless
CHANGELOG.md
+21 -6
View File
@@ -7,7 +7,6 @@ All notable changes to this project will be documented in this file.
### Added
- Parameterized all hardcoded configuration values as Ansible variables in `defaults/main.yml`:
- `gitea_runner_docker_image` — Docker image name (default: `gitea/runner`)
- `gitea_runner_data_dir` — Runtime data directory
- `gitea_runner_config_dir` — Config directory
- `gitea_runner_binary_path` — Binary install path
@@ -23,16 +22,26 @@ All notable changes to this project will be documented in this file.
- Added `console_scripts` entry point in `pyproject.toml` (`grm = "gitea_runner_manager.cli:cli"`).
- Added shared `molecule/common/prepare.yml` to eliminate duplicated prepare playbooks.
- Extracted repeated systemd availability check into `tasks/systemd_check.yml`.
- Added idempotence checks to both Molecule scenarios (`default` and `binary`).
- Added idempotence checks to all Molecule scenarios.
- Comprehensive README overhaul with Architecture, Configuration, Development, Testing, and Troubleshooting sections.
- API URLs and repo configuration in `config.py` are now overridable via environment variables (`GRM_GITEA_API_URL`, `GRM_VIKUNJA_API_URL`, `GRM_REPO_OWNER`, `GRM_REPO_NAME`, `GRM_VIKUNJA_PROJECT_ID`).
- `remove-runner.yml` now disables lingering and removes subuid/subgid entries for complete cleanup.
### Security
- **Critical fix**: Registration tokens and admin tokens are no longer passed via `--extra-vars` on the command line (CWE-214). Extra-vars are now written to a temporary JSON file with `0600` permissions and passed via `--extra-vars @tempfile`, which is deleted after execution. This prevents secrets from being visible in the process list (`ps aux`).
### Changed
- Replaced legacy runner terminology with `gitea_runner` / `gitea-runner` / `Gitea Runner`.
- Updated default Docker image from `gitea/gitea_runner` to `gitea/runner`.
- `Makefile` now uses the installed `grm` console script instead of `python grm`.
- `docker_update.yml`: reordered `register` before `changed_when` for readability.
- `binary_update.yml`: removed redundant explicit service restart (handler handles this idempotently).
- `pyproject.toml` ruff and pyright target versions updated from `py311` to `py312` to match `requires-python = ">=3.12"`.
- `BRANCH_PROTECTION_CONFIG` updated with correct Gitea Actions status check contexts (including `(pull_request)` suffix) and `required_approvals: 0` for auto-merge.
- `CONVENTIONAL_RE` no longer matches `BREAKING CHANGE` as a commit type (it is a footer, not a type).
- `rootless_docker.yml` apt cache update now only runs when the Docker repo file changes (idempotent, but always refreshes on first add).
- `service.yml` and `prune.yml` template creation tasks are not guarded by `docker_rootless_setup` (templates just create files, they don't need Docker; molecule tests set `docker_rootless_setup: false` but still verify the service file exists).
- `molecule_all.sh` now sources the platform list from `distribute_molecule.py` to avoid duplication.
### Removed
@@ -41,11 +50,17 @@ All notable changes to this project will be documented in this file.
- Deleted `initial-plan.md` and `tests/integration/test_provision.py` (dead code).
- Removed empty `__init__.py` files from `tests/` directories.
- Removed unused `runner_validated` fact from `validate.yml`.
- Removed duplicate `prune.yml` and `integration_test.yml` includes from `docker_mode.yml` and `binary_mode.yml` (now included once from `main.yml`).
- Removed duplicate `prune.yml` and `integration_test.yml` includes from `install_runner.yml` (already included from `main.yml`).
- Removed dead `tasks/config.yml` (never included by any playbook).
- Removed dead `templates/gitea-runner.service.j2` (legacy system-level service, replaced by rootless `gitea-runner-user.service.j2`).
- Removed dead "Reload systemd" handler (system-level reload, never notified, wrong scope for user services).
- Removed dead `scripts/run_molecule_parallel.py` and its test (replaced by `molecule_ci_guard.py`).
### Fixed
- Molecule idempotence failures caused by non-idempotent service restart in `binary_update.yml`.
- Molecule idempotence failures caused by non-idempotent service restart.
- Missing `/etc/docker` directory handling in Molecule tests.
- `ansible-lint` formatting warnings (yaml empty lines).
- Verify playbooks now explicitly load role defaults so parameterized variables are available during verification.
- Duplicate execution of prune and integration test tasks during installation (were included from both `main.yml` and `install_runner.yml`).
- apt cache update reporting `changed` on every run due to `cache_valid_time: 0`.
README.md
+6 -1
View File
@@ -219,7 +219,7 @@ All tunable values are exposed as Ansible variables in `ansible/roles/gitea-runn
| Variable | Default | Description |
|----------|---------|-------------|
| `gitea_runner_version` | `1.0.8` | Runner binary version |
| `runner_labels` | `ubuntu-latest:docker://runner-images:ubuntu-22.04` | Runner labels |
| `runner_labels` | `docker,ubuntu-latest:docker://runner-images:ubuntu-22.04` | Runner labels |
| `skip_runner_registration` | `false` | Skip API registration (useful for tests) |
| `gitea_runner_user_prefix` | `grm-` | Prefix for per-runner system users |
| `gitea_runner_base_home` | `/home` | Base directory for runner user homes |
@@ -239,6 +239,11 @@ All tunable values are exposed as Ansible variables in `ansible/roles/gitea-runn
| `docker_gpg_key_path` | `/etc/apt/keyrings/docker.asc` | Docker GPG key path |
| `GRM_LANG` | `en` | CLI language: `en`, `bg`, `de`, `ru`, `zh` |
| `GRM_LOG_LEVEL` | `INFO` | Console verbosity: `DEBUG`, `INFO`, `WARNING`, `ERROR`, `CRITICAL` |
| `GRM_GITEA_API_URL` | `https://git.oblachno.oblachno.fyi/api/v1` | Gitea API URL for CI scripts |
| `GRM_VIKUNJA_API_URL` | `https://work.oblachno.oblachno.fyi/api/v1` | Vikunja API URL for post-merge scripts |
| `GRM_REPO_OWNER` | `oblachno-oss` | Repository owner for CI scripts |
| `GRM_REPO_NAME` | `grm` | Repository name for CI scripts |
| `GRM_VIKUNJA_PROJECT_ID` | `6` | Vikunja project ID for task tracking |
Override any variable by passing it to the CLI with `--extra-vars` or by setting it in your Ansible inventory.
TROUBLESHOOTING.md
+6 -1
View File
@@ -9,4 +9,9 @@
| Vikunja task not updated after merge | VIKUNJA_TOKEN expired or task ID missing from commit | Regenerate token; verify merge commit has `GRM-N:` prefix |
| Post-merge can't find Vikunja task | Task not in project 6 or identifier mismatch | Verify task exists in Vikunja project 6 with correct identifier |
| `make pytest-cov` fails | Coverage below 100% | Add tests for new code paths |
| `scripts/configure_repo.py` fails | GITEA_ADMIN_TOKEN missing or invalid | Set token with repo admin scope and re-run |
| `scripts/configure_repo.py` fails | REPO_TOKEN missing or invalid | Set token with repo admin scope and re-run |
| `configure_repo.py` sets wrong status checks | Stale `BRANCH_PROTECTION_CONFIG` | Updated to include `(pull_request)` suffix; re-run `configure_repo.py` |
| Token visible in `ps aux` during install | Old version passed tokens via command line | Fixed: tokens now passed via temp file with `0600` permissions |
| `remove-runner.yml` leaves lingering enabled | Old version didn't disable lingering | Fixed: now runs `loginctl disable-linger` and removes subuid/subgid |
| apt cache update always reports `changed` | `cache_valid_time: 0` forced update every run | Fixed: changed to `cache_valid_time: 3600` |
| Prune/service templates created even when `docker_rootless_setup: false` | Template tasks not guarded | Fixed: template creation now guarded by `docker_rootless_setup` |
ansible/remove-runner.yml
+19
View File
@@ -71,6 +71,11 @@
failed_when: false
changed_when: false
- name: Disable lingering for runner user
ansible.builtin.command: loginctl disable-linger "{{ gitea_runner_service_user | default('grm-' ~ runner_name) }}"
failed_when: false
changed_when: true
- name: Remove runner user and home directory
ansible.builtin.user:
name: "{{ gitea_runner_service_user | default('grm-' ~ runner_name) }}"
@@ -79,6 +84,20 @@
when: remove_runner_user | default(true)
failed_when: false
- name: Remove subuid entry for runner user
ansible.builtin.lineinfile:
path: /etc/subuid
regexp: "^{{ gitea_runner_service_user | default('grm-' ~ runner_name) }}:"
state: absent
failed_when: false
- name: Remove subgid entry for runner user
ansible.builtin.lineinfile:
path: /etc/subgid
regexp: "^{{ gitea_runner_service_user | default('grm-' ~ runner_name) }}:"
state: absent
failed_when: false
- name: Remove runner data directory
ansible.builtin.file:
path: "{{ gitea_runner_data_dir | default('/var/lib/gitea-runner/' ~ runner_name) }}"
ansible/roles/gitea-runner/handlers/main.yml
-8
View File
@@ -1,12 +1,4 @@
---
- name: Reload systemd
ansible.builtin.systemd:
daemon_reload: true
when:
- ansible_facts is defined
- ansible_facts['service_mgr'] | default('') == 'systemd'
- docker_rootless_setup
- name: Restart gitea-runner
ansible.builtin.command: systemctl --user restart gitea-runner
become: true
ansible/roles/gitea-runner/tasks/config.yml
-13
View File
@@ -1,13 +0,0 @@
---
- name: Ensure config directory exists
ansible.builtin.file:
path: "{{ gitea_runner_config_dir }}"
state: directory
mode: "0755"
- name: Create gitea_runner config file
ansible.builtin.template:
src: gitea-runner-config.yaml.j2
dest: "{{ gitea_runner_config_dir }}/config.yaml"
mode: "0644"
notify: Restart gitea-runner
ansible/roles/gitea-runner/tasks/install_runner.yml
-7
View File
@@ -19,10 +19,3 @@
- name: Include service setup
ansible.builtin.include_tasks: service.yml
- name: Include prune setup
ansible.builtin.include_tasks: prune.yml
- name: Include integration test
ansible.builtin.include_tasks: integration_test.yml
when: not skip_runner_registration
ansible/roles/gitea-runner/tasks/rootless_docker.yml
+4 -2
View File
@@ -20,13 +20,15 @@
dest: /etc/apt/sources.list.d/docker.list
content: "{{ docker_apt_source_line }}\n"
mode: "0644"
register: docker_apt_repo
when: ansible_facts['os_family'] == 'Debian'
- name: Update apt cache after adding Docker repo (Debian/Ubuntu)
ansible.builtin.apt:
update_cache: true
cache_valid_time: 0
when: ansible_facts['os_family'] == 'Debian'
when:
- ansible_facts['os_family'] == 'Debian'
- docker_apt_repo is changed
- name: Install rootless Docker dependencies (Debian/Ubuntu)
ansible.builtin.apt:
ansible/roles/gitea-runner/templates/gitea-runner.service.j2
-16
View File
@@ -1,16 +0,0 @@
[Unit]
Description=Gitea Actions Runner ({{ runner_name }})
After=network.target docker.service
Requires=docker.service
[Service]
Type=simple
ExecStart={{ gitea_runner_binary_path }} daemon --config {{ gitea_runner_config_dir }}/config.yaml
WorkingDirectory={{ gitea_runner_data_dir }}
Restart=always
RestartSec={{ gitea_runner_service_restart_sec }}
User={{ gitea_runner_service_user }}
Group=docker
[Install]
WantedBy=multi-user.target
pyproject.toml
+2 -2
View File
@@ -53,7 +53,7 @@ markers = [
]
[tool.ruff]
target-version = "py311"
target-version = "py312"
line-length = 120
[tool.ruff.lint]
@@ -66,5 +66,5 @@ indent-style = "space"
[tool.pyright]
include = ["src", "scripts"]
pythonVersion = "3.11"
pythonVersion = "3.12"
strict = ["src/gitea_runner_manager"]
scripts/molecule_all.sh
+8 -7
View File
@@ -1,18 +1,19 @@
#!/usr/bin/env bash
# Run all molecule scenarios on all supported OS platforms.
# Used by `make molecule-all`. Sequential — CI uses parallel matrix instead.
# Platform list is sourced from scripts/distribute_molecule.py to avoid duplication.
set -euo pipefail
MOLECULE_BIN="$(realpath "${BIN:-.venv/bin}/molecule")"
ROLE_DIR="$(cd "$(dirname "$0")/.." && pwd)/ansible/roles/gitea-runner"
SCRIPTS_DIR="$(cd "$(dirname "$0")" && pwd)"
for p in \
ubuntu-2204:geerlingguy/docker-ubuntu2204-ansible:latest:/lib/systemd/systemd \
ubuntu-2404:geerlingguy/docker-ubuntu2404-ansible:latest:/lib/systemd/systemd \
debian-12:geerlingguy/docker-debian12-ansible:latest:/lib/systemd/systemd \
archlinux:marcstraube/archlinux-ansible:latest:/usr/lib/systemd/systemd
do
IFS=":" read -r name image command <<< "$p"
# Read platforms from distribute_molecule.py (single source of truth)
PLATFORMS_OUTPUT="$("$MOLECULE_BIN" python "${SCRIPTS_DIR}/distribute_molecule.py" --list-platforms 2>/dev/null || \
python3 "${SCRIPTS_DIR}/distribute_molecule.py" --list-platforms)"
for p in $PLATFORMS_OUTPUT; do
IFS="|" read -r name image command <<< "$p"
export MOLECULE_PLATFORM_NAME="$name" MOLECULE_PLATFORM_IMAGE="$image"
if [ -n "$command" ]; then
export MOLECULE_PLATFORM_COMMAND="$command"
scripts/run_molecule_parallel.py
-123
View File
@@ -1,123 +0,0 @@
#!/usr/bin/env python3
"""Run molecule (scenario, platform) pairs in parallel and kill on first failure.
Each pair is encoded as ``scenario|platform_name|platform_image|platform_command``.
Subprocesses are started concurrently. If any subprocess exits with a non-zero
status, the remaining subprocesses are terminated and this script exits with 1.
Usage:
python3 scripts/run_molecule_parallel.py \
default|ubuntu-2204|ubuntu:22.04| \
lifecycle|debian-12|debian:12| \
...
"""
from __future__ import annotations
import contextlib
import os
import signal
import subprocess # nosec B404
import sys
from pathlib import Path
import click
from gitea_runner_manager.i18n import _
def run_pair(pair: str, role_dir: Path) -> subprocess.Popen[bytes]:
"""Start a subprocess for a single molecule (scenario, platform) pair."""
scenario, platform_name, platform_image, platform_command = pair.split("|")
env = os.environ.copy()
env["MOLECULE_PLATFORM_NAME"] = platform_name
env["MOLECULE_PLATFORM_IMAGE"] = platform_image
if platform_command:
env["MOLECULE_PLATFORM_COMMAND"] = platform_command
elif "MOLECULE_PLATFORM_COMMAND" in env:
del env["MOLECULE_PLATFORM_COMMAND"]
env["ANSIBLE_ALLOW_BROKEN_CONDITIONALS"] = "true"
env["ANSIBLE_INJECT_INVOCATION"] = "1"
cmd = ["molecule", "test"]
if scenario != "default":
cmd.extend(["-s", scenario])
click.echo(_("Starting: {scenario} on {platform}", scenario=scenario, platform=platform_name))
return subprocess.Popen( # nosec B603
cmd,
cwd=str(role_dir),
env=env,
preexec_fn=os.setsid,
)
@click.command()
@click.argument("pairs", nargs=-1, required=True)
def cli(pairs: tuple[str, ...]) -> None:
"""Run molecule pairs in parallel, stop on first failure."""
repo_root = Path(__file__).resolve().parent.parent
role_dir = repo_root / "ansible" / "roles" / "gitea-runner"
processes: list[subprocess.Popen[bytes]] = []
pair_names: list[str] = []
for pair in pairs:
proc = run_pair(pair, role_dir)
processes.append(proc)
pair_names.append(pair)
first_failure: str | None = None
returncode = 0
while processes:
finished_indices: list[int] = []
for i, proc in enumerate(processes):
ret = proc.poll()
if ret is not None:
finished_indices.append(i)
if ret != 0:
first_failure = pair_names[i]
returncode = ret
if first_failure is not None:
click.echo(
_(
"FAILURE: {pair} exited with code {code}. Stopping remaining tests.",
pair=first_failure,
code=returncode,
)
)
for proc in processes:
if proc.poll() is None:
with contextlib.suppress(ProcessLookupError):
os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
# Wait briefly, then SIGKILL survivors
for proc in processes:
try:
proc.wait(timeout=5)
except subprocess.TimeoutExpired:
with contextlib.suppress(ProcessLookupError):
os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
proc.wait()
sys.exit(returncode)
if not finished_indices:
# No process finished yet, wait a bit
for proc in processes:
try:
proc.wait(timeout=5)
except subprocess.TimeoutExpired:
pass
continue
# Remove finished processes from the list
for i in sorted(finished_indices, reverse=True):
del processes[i]
del pair_names[i]
click.echo(_("All molecule tests passed."))
sys.exit(0)
if __name__ == "__main__": # pragma: no cover
cli()
src/gitea_runner_manager/config.py
+14 -10
View File
@@ -2,20 +2,19 @@
from __future__ import annotations
import os
import re
GITEA_API_URL = "https://git.oblachno.oblachno.fyi/api/v1"
VIKUNJA_API_URL = "https://work.oblachno.oblachno.fyi/api/v1"
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 = "oblachno-oss"
REPO_NAME = "grm"
REPO_OWNER = os.getenv("GRM_REPO_OWNER", "oblachno-oss")
REPO_NAME = os.getenv("GRM_REPO_NAME", "grm")
VIKUNJA_PROJECT_ID = 6
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|BREAKING CHANGE)(\(.+\))?: .+"
)
CONVENTIONAL_RE = re.compile(r"^(feat|fix|chore|docs|style|refactor|perf|test|ci|build|revert)(\(.+\))?: .+")
DEFAULT_TIMEOUT = 30
DEFAULT_PER_PAGE = 50
@@ -24,8 +23,13 @@ BRANCH_PROTECTION_CONFIG: dict[str, object] = {
"branch_name": "master",
"enable_push": False,
"enable_status_check": True,
"status_check_contexts": ["CI / quality", "CI / molecule-tests*"],
"required_approvals": 1,
"status_check_contexts": [
"CI / quality (pull_request)",
"CI / molecule-tests (0) (pull_request)",
"CI / molecule-tests (1) (pull_request)",
"CI / molecule-tests (2) (pull_request)",
],
"required_approvals": 0,
"dismiss_stale_approvals": True,
"block_on_outdated_branch": True,
"block_on_rejected_reviews": True,
src/gitea_runner_manager/runner_manager.py
+127 -49
View File
@@ -2,6 +2,12 @@
from __future__ import annotations
import contextlib
import json
import os
import tempfile
from collections.abc import Generator
from contextlib import contextmanager
from pathlib import Path
from .exceptions import AnsibleError
@@ -23,6 +29,42 @@ class RunnerManager:
self._executor = executor or AnsibleExecutor()
self._registry = registry or RunnerRegistry()
@contextmanager
def _extra_vars_file(self, extra_vars: dict[str, str | int] | None) -> Generator[str | None, None, None]:
"""Write extra-vars to a temp JSON file with restricted permissions.
Secrets passed via ``--extra-vars`` on the command line are visible
in the process list (CWE-214). This context manager writes them to
a temporary file with ``0600`` permissions and cleans up on exit.
"""
if not extra_vars:
yield None
return
fd, path = tempfile.mkstemp(suffix=".json", prefix="grm-vars-")
try:
with os.fdopen(fd, "w") as f:
json.dump(extra_vars, f)
os.chmod(path, 0o600)
yield path
finally:
with contextlib.suppress(FileNotFoundError):
os.unlink(path)
def _run_playbook(
self,
playbook_name: str,
host: str,
user: str,
extra_vars: dict[str, str | int] | None = None,
key: str | None = None,
ask_become_pass: bool = False,
description: str = "",
) -> 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)
self._executor.run(cmd, description=description)
def install(
self,
host: str,
@@ -44,19 +86,28 @@ class RunnerManager:
if not token:
raise AnsibleError(_("GITEA_REGISTRATION_TOKEN must be set (or pass --token)"))
extra_vars = (
f"registration_token={token} runner_name={name} gitea_url={gitea_url}"
f" gitea_runner_integration_retries={integration_retries}"
)
extra_vars: dict[str, str | int] = {
"registration_token": token,
"runner_name": name,
"gitea_url": gitea_url,
"gitea_runner_integration_retries": integration_retries,
}
if admin_token:
extra_vars += f" gitea_admin_token={admin_token}"
extra_vars["gitea_admin_token"] = admin_token
if labels:
extra_vars += f" runner_labels={labels}"
extra_vars["runner_labels"] = labels
with track_steps() as tracker:
tracker.begin(_("Installing Gitea Runner on {host}", host=host))
cmd = self._build_cmd("install-runner.yml", host, user, extra_vars, key, ask_become_pass)
self._executor.run(cmd, description=_("Installing Gitea Runner on {host}", host=host))
self._run_playbook(
"install-runner.yml",
host,
user,
extra_vars,
key,
ask_become_pass,
description=_("Installing Gitea Runner on {host}", host=host),
)
tracker.done()
tracker.begin(_("Save runner '{name}' to local registry", name=name))
@@ -79,14 +130,21 @@ class RunnerManager:
ask_become_pass: bool = False,
) -> None:
"""Update the gitea_runner binary on a remote host."""
extra_vars = ""
extra_vars: dict[str, str | int] | None = None
if version:
extra_vars += f" gitea_runner_version={version}"
extra_vars = {"gitea_runner_version": version}
with track_steps() as tracker:
tracker.begin(_("Updating Gitea Runner on {host}", host=host))
cmd = self._build_cmd("update-runner.yml", host, user, extra_vars, key, ask_become_pass)
self._executor.run(cmd, description=_("Updating Gitea Runner on {host}", host=host))
self._run_playbook(
"update-runner.yml",
host,
user,
extra_vars,
key,
ask_become_pass,
description=_("Updating Gitea Runner on {host}", host=host),
)
tracker.done()
def _resolve_runner(
@@ -134,10 +192,14 @@ class RunnerManager:
actual_host, actual_user, actual_key, _gitea_url = self._resolve_runner(name, host, user, key)
with track_steps() as tracker:
tracker.begin(_("Starting Gitea Runner {name} on {host}", name=name, host=actual_host))
extra_vars = f"runner_name={name}"
cmd = self._build_cmd("start-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass)
self._executor.run(
cmd, description=_("Starting Gitea Runner {name} on {host}", name=name, host=actual_host)
self._run_playbook(
"start-runner.yml",
actual_host,
actual_user,
{"runner_name": name},
actual_key,
ask_become_pass,
description=_("Starting Gitea Runner {name} on {host}", name=name, host=actual_host),
)
tracker.done()
@@ -153,10 +215,14 @@ class RunnerManager:
actual_host, actual_user, actual_key, _gitea_url = self._resolve_runner(name, host, user, key)
with track_steps() as tracker:
tracker.begin(_("Stopping Gitea Runner {name} on {host}", name=name, host=actual_host))
extra_vars = f"runner_name={name}"
cmd = self._build_cmd("stop-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass)
self._executor.run(
cmd, description=_("Stopping Gitea Runner {name} on {host}", name=name, host=actual_host)
self._run_playbook(
"stop-runner.yml",
actual_host,
actual_user,
{"runner_name": name},
actual_key,
ask_become_pass,
description=_("Stopping Gitea Runner {name} on {host}", name=name, host=actual_host),
)
tracker.done()
@@ -172,12 +238,14 @@ class RunnerManager:
actual_host, actual_user, actual_key, _gitea_url = self._resolve_runner(name, host, user, key)
with track_steps() as tracker:
tracker.begin(_("Enabling Gitea Runner {name} on {host}", name=name, host=actual_host))
extra_vars = f"runner_name={name}"
cmd = self._build_cmd(
"enable-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass
)
self._executor.run(
cmd, description=_("Enabling Gitea Runner {name} on {host}", name=name, host=actual_host)
self._run_playbook(
"enable-runner.yml",
actual_host,
actual_user,
{"runner_name": name},
actual_key,
ask_become_pass,
description=_("Enabling Gitea Runner {name} on {host}", name=name, host=actual_host),
)
tracker.done()
@@ -200,12 +268,14 @@ class RunnerManager:
raise AnsibleError(_("GITEA_REGISTRATION_TOKEN must be set (or pass --token)"))
with track_steps() as tracker:
tracker.begin(_("Disabling Gitea Runner {name} on {host}", name=name, host=actual_host))
extra_vars = f"runner_name={name} registration_token={token} gitea_url={resolved_gitea_url}"
cmd = self._build_cmd(
"disable-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass
)
self._executor.run(
cmd, description=_("Disabling Gitea Runner {name} on {host}", name=name, host=actual_host)
self._run_playbook(
"disable-runner.yml",
actual_host,
actual_user,
{"runner_name": name, "registration_token": token, "gitea_url": resolved_gitea_url},
actual_key,
ask_become_pass,
description=_("Disabling Gitea Runner {name} on {host}", name=name, host=actual_host),
)
tracker.done()
@@ -221,12 +291,14 @@ class RunnerManager:
actual_host, actual_user, actual_key, _gitea_url = self._resolve_runner(name, host, user, key)
with track_steps() as tracker:
tracker.begin(_("Checking status of Gitea Runner {name} on {host}", name=name, host=actual_host))
extra_vars = f"runner_name={name}"
cmd = self._build_cmd(
"status-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass
)
self._executor.run(
cmd, description=_("Checking status of Gitea Runner {name} on {host}", name=name, host=actual_host)
self._run_playbook(
"status-runner.yml",
actual_host,
actual_user,
{"runner_name": name},
actual_key,
ask_become_pass,
description=_("Checking status of Gitea Runner {name} on {host}", name=name, host=actual_host),
)
tracker.done()
@@ -257,13 +329,15 @@ class RunnerManager:
raise AnsibleError(_("GITEA_REGISTRATION_TOKEN must be set (or pass --token)"))
with track_steps() as tracker:
tracker.begin(_("Removing Gitea Runner {name} from {host}", name=name, host=actual_host))
if not force:
extra_vars = f"runner_name={name} registration_token={resolved_token} gitea_url={resolved_gitea_url}"
cmd = self._build_cmd(
"remove-runner.yml", actual_host, actual_user, extra_vars, actual_key, ask_become_pass
)
self._executor.run(
cmd, description=_("Removing Gitea Runner {name} from {host}", name=name, host=actual_host)
if not force and resolved_token and resolved_gitea_url:
self._run_playbook(
"remove-runner.yml",
actual_host,
actual_user,
{"runner_name": name, "registration_token": resolved_token, "gitea_url": resolved_gitea_url},
actual_key,
ask_become_pass,
description=_("Removing Gitea Runner {name} from {host}", name=name, host=actual_host),
)
tracker.done()
@@ -328,11 +402,15 @@ class RunnerManager:
playbook_name: str,
host: str,
user: str,
extra_vars: str,
extra_vars_file: str | None = None,
key: str | None = None,
ask_become_pass: bool = False,
) -> list[str]:
"""Build the ansible-playbook command."""
"""Build the ansible-playbook command.
Extra-vars are passed via ``@tempfile`` to avoid exposing secrets
in the process list (CWE-214).
"""
playbook = Path(__file__).parent.parent.parent / "ansible" / playbook_name
if not playbook.exists():
raise AnsibleError(_("Playbook not found: {playbook}", playbook=playbook))
@@ -344,9 +422,9 @@ class RunnerManager:
f"{host},",
"-u",
user,
"--extra-vars",
extra_vars,
]
if extra_vars_file:
cmd.extend(["--extra-vars", f"@{extra_vars_file}"])
if key:
cmd.extend(["--private-key", key])
if ask_become_pass:
tests/unit/test_config.py
+9 -4
View File
@@ -17,8 +17,8 @@ from gitea_runner_manager.config import (
class TestConfigConstants:
def test_api_urls(self) -> None:
assert GITEA_API_URL == "https://git.oblachno.oblachno.fyi/api/v1"
assert VIKUNJA_API_URL == "https://work.oblachno.oblachno.fyi/api/v1"
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
@@ -42,12 +42,17 @@ class TestConfigConstants:
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 False
assert BRANCH_PROTECTION_CONFIG["required_approvals"] == 1
assert BRANCH_PROTECTION_CONFIG["status_check_contexts"] == ["CI / quality", "CI / molecule-tests*"]
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)
def test_label_config(self) -> None:
assert LABEL_CONFIG["name"] == "ready-to-merge"
tests/unit/test_run_molecule_parallel.py
-193
View File
@@ -1,193 +0,0 @@
"""Unit tests for scripts/run_molecule_parallel.py."""
from __future__ import annotations
import os
import subprocess # nosec B404
from pathlib import Path
from unittest.mock import MagicMock, patch
from scripts.run_molecule_parallel import cli, run_pair
class TestRunPair:
def test_default_scenario(self, tmp_path: Path) -> None:
with (
patch("scripts.run_molecule_parallel.subprocess.Popen") as mock_popen,
patch.dict(os.environ, {"MOLECULE_PLATFORM_COMMAND": "old-command"}, clear=False),
):
mock_popen.return_value = MagicMock()
result = run_pair("default|ubuntu-2204|img:latest|", tmp_path)
assert result is mock_popen.return_value
call_args = mock_popen.call_args
assert call_args.kwargs["cwd"] == str(tmp_path)
env = call_args.kwargs["env"]
assert env["MOLECULE_PLATFORM_NAME"] == "ubuntu-2204"
assert env["MOLECULE_PLATFORM_IMAGE"] == "img:latest"
assert "MOLECULE_PLATFORM_COMMAND" not in env
assert call_args.args[0] == ["molecule", "test"]
def test_named_scenario_with_command(self, tmp_path: Path) -> None:
with patch("scripts.run_molecule_parallel.subprocess.Popen") as mock_popen:
mock_popen.return_value = MagicMock()
result = run_pair("lifecycle|archlinux|img:arch|/usr/lib/systemd/systemd", tmp_path)
assert result is mock_popen.return_value
call_args = mock_popen.call_args
env = call_args.kwargs["env"]
assert env["MOLECULE_PLATFORM_NAME"] == "archlinux"
assert env["MOLECULE_PLATFORM_IMAGE"] == "img:arch"
assert env["MOLECULE_PLATFORM_COMMAND"] == "/usr/lib/systemd/systemd"
assert call_args.args[0] == ["molecule", "test", "-s", "lifecycle"]
class TestCli:
def test_all_pass(self, tmp_path: Path) -> None:
from click.testing import CliRunner
with patch("scripts.run_molecule_parallel.run_pair") as mock_run_pair:
mock_run_pair.return_value = MagicMock()
mock_run_pair.return_value.poll.side_effect = [None, 0]
mock_run_pair.return_value.wait.return_value = None
runner = CliRunner()
result = runner.invoke(cli, ["default|ubuntu-2204|img:latest|"])
assert result.exit_code == 0
assert "All molecule tests passed" in result.output
def test_failure_kills_remaining(self, tmp_path: Path) -> None:
from click.testing import CliRunner
with (
patch("scripts.run_molecule_parallel.run_pair") as mock_run_pair,
patch("os.killpg") as mock_killpg,
patch("os.getpgid") as mock_getpgid,
):
mock_getpgid.return_value = 123
good_proc = MagicMock()
good_proc.poll.return_value = None
good_proc.wait.return_value = None
bad_proc = MagicMock()
bad_proc.poll.side_effect = [None, 1, 1, 1]
bad_proc.wait.return_value = None
mock_run_pair.side_effect = [bad_proc, good_proc]
runner = CliRunner()
result = runner.invoke(
cli,
[
"default|ubuntu-2204|img:latest|",
"lifecycle|debian-12|img:deb|",
],
)
assert result.exit_code == 1
assert "FAILURE" in result.output
mock_killpg.assert_called()
def test_failure_sends_sigkill_when_sigterm_times_out(self, tmp_path: Path) -> None:
from click.testing import CliRunner
with (
patch("scripts.run_molecule_parallel.run_pair") as mock_run_pair,
patch("os.killpg") as mock_killpg,
patch("os.getpgid") as mock_getpgid,
):
mock_getpgid.return_value = 123
# First SIGTERM succeeds, second SIGKILL raises ProcessLookupError
mock_killpg.side_effect = [None, ProcessLookupError("no such process")]
good_proc = MagicMock()
good_proc.poll.return_value = None
good_proc.wait.side_effect = [None, subprocess.TimeoutExpired("cmd", 5)]
bad_proc = MagicMock()
bad_proc.poll.side_effect = [None, 1, 1, 1]
bad_proc.wait.return_value = None
mock_run_pair.side_effect = [bad_proc, good_proc]
runner = CliRunner()
result = runner.invoke(
cli,
[
"default|ubuntu-2204|img:latest|",
"lifecycle|debian-12|img:deb|",
],
)
assert result.exit_code == 1
assert "FAILURE" in result.output
# First SIGTERM, then SIGKILL
assert mock_killpg.call_count >= 2
def test_failure_handles_process_lookup_error(self, tmp_path: Path) -> None:
from click.testing import CliRunner
with (
patch("scripts.run_molecule_parallel.run_pair") as mock_run_pair,
patch("os.killpg") as mock_killpg,
patch("os.getpgid") as mock_getpgid,
):
mock_getpgid.return_value = 123
mock_killpg.side_effect = ProcessLookupError("no such process")
good_proc = MagicMock()
good_proc.poll.return_value = None
good_proc.wait.return_value = None
bad_proc = MagicMock()
bad_proc.poll.side_effect = [None, 1, 1, 1]
bad_proc.wait.return_value = None
mock_run_pair.side_effect = [bad_proc, good_proc]
runner = CliRunner()
result = runner.invoke(
cli,
[
"default|ubuntu-2204|img:latest|",
"lifecycle|debian-12|img:deb|",
],
)
assert result.exit_code == 1
assert "FAILURE" in result.output
def test_wait_loop_handles_timeout(self, tmp_path: Path) -> None:
from click.testing import CliRunner
with (
patch("scripts.run_molecule_parallel.run_pair") as mock_run_pair,
patch("os.killpg") as mock_killpg,
patch("os.getpgid") as mock_getpgid,
):
mock_getpgid.return_value = 123
good_proc = MagicMock()
good_proc.poll.return_value = None
good_proc.wait.side_effect = [
subprocess.TimeoutExpired("cmd", 5),
None,
None,
]
bad_proc = MagicMock()
bad_proc.poll.side_effect = [None, 1, 1, 1]
bad_proc.wait.return_value = None
mock_run_pair.side_effect = [bad_proc, good_proc]
runner = CliRunner()
result = runner.invoke(
cli,
[
"default|ubuntu-2204|img:latest|",
"lifecycle|debian-12|img:deb|",
],
)
assert result.exit_code == 1
assert "FAILURE" in result.output
mock_killpg.assert_called()
def test_main_module_block() -> None:
import scripts.run_molecule_parallel as rm
with open(rm.__file__) as f:
source = f.read()
source = source.replace('if __name__ == "__main__":\n cli()\n', "")
namespace = dict(rm.__dict__)
exec(compile(source, rm.__file__, "exec"), namespace)
assert callable(namespace["cli"])
tests/unit/test_runner_manager.py
+83 -33
View File
@@ -1,5 +1,8 @@
"""Unit tests for runner_manager module."""
import json
import os
from contextlib import contextmanager
from pathlib import Path
from unittest.mock import MagicMock, patch
@@ -9,7 +12,20 @@ from gitea_runner_manager.exceptions import AnsibleError
from gitea_runner_manager.runner_manager import RunnerManager
@contextmanager
def _capture_extra_vars(self, extra_vars: dict[str, str | int] | None):
"""Capture extra_vars dict for inspection instead of writing to file."""
self._captured_extra_vars = extra_vars
yield "/tmp/fake-vars.json" if extra_vars else None
class TestRunnerManager:
@pytest.fixture(autouse=True)
def _patch_extra_vars_file(self) -> None:
"""Patch ``_extra_vars_file`` so no real temp files are created."""
with patch.object(RunnerManager, "_extra_vars_file", _capture_extra_vars):
yield
def test_init(self) -> None:
manager = RunnerManager()
assert manager is not None
@@ -30,9 +46,11 @@ class TestRunnerManager:
assert "192.168.1.10," in cmd_str
assert "-u" in cmd_str
assert "ubuntu" in cmd_str
assert "registration_token=tok" in cmd_str
assert "runner_name=192.168.1.10" in cmd_str
assert "gitea_url=https://git.example.com" in cmd_str
assert "--extra-vars" in cmd_str
assert "@/tmp/fake-vars.json" in cmd_str
assert manager._captured_extra_vars["registration_token"] == "tok"
assert manager._captured_extra_vars["runner_name"] == "192.168.1.10"
assert manager._captured_extra_vars["gitea_url"] == "https://git.example.com"
assert "Installing Gitea Runner on 192.168.1.10" in mock_executor.run.call_args.kwargs["description"]
mock_registry.add.assert_called_once_with(
name="192.168.1.10",
@@ -56,8 +74,8 @@ class TestRunnerManager:
cmd_str = " ".join(cmd)
assert "--private-key" in cmd_str
assert "/key" in cmd_str
assert "registration_token=preset" in cmd_str
assert "runner_name=my-runner" in cmd_str
assert manager._captured_extra_vars["registration_token"] == "preset"
assert manager._captured_extra_vars["runner_name"] == "my-runner"
assert "--ask-become-pass" not in cmd_str
mock_registry.add.assert_called_once_with(
name="my-runner",
@@ -108,9 +126,7 @@ class TestRunnerManager:
gitea_url="https://git.example.com",
labels="docker:docker://alpine:latest",
)
cmd = mock_executor.run.call_args.args[0]
cmd_str = " ".join(cmd)
assert "runner_labels=docker:docker://alpine:latest" in cmd_str
assert manager._captured_extra_vars["runner_labels"] == "docker:docker://alpine:latest"
def test_install_no_labels(self) -> None:
mock_registry = MagicMock()
@@ -119,9 +135,7 @@ class TestRunnerManager:
manager._executor = mock_executor
manager.install("host1", "root", token="tok", gitea_url="https://git.example.com")
cmd = mock_executor.run.call_args.args[0]
cmd_str = " ".join(cmd)
assert "runner_labels" not in cmd_str
assert "runner_labels" not in manager._captured_extra_vars
def test_install_with_admin_token(self) -> None:
mock_registry = MagicMock()
@@ -130,9 +144,7 @@ class TestRunnerManager:
manager._executor = mock_executor
manager.install("host1", "root", token="tok", gitea_url="https://git.example.com", admin_token="admin-tok")
cmd = mock_executor.run.call_args.args[0]
cmd_str = " ".join(cmd)
assert "gitea_admin_token=admin-tok" in cmd_str
assert manager._captured_extra_vars["gitea_admin_token"] == "admin-tok"
def test_install_ansible_failure(self) -> None:
mock_registry = MagicMock()
@@ -155,7 +167,7 @@ class TestRunnerManager:
assert "update-runner.yml" in cmd_str
assert "--private-key" in cmd_str
assert "/key" in cmd_str
assert "gitea_runner_version=v0.2.0" in cmd_str
assert manager._captured_extra_vars["gitea_runner_version"] == "v0.2.0"
assert "--ask-become-pass" not in cmd_str
assert "Updating Gitea Runner on host" in mock_executor.run.call_args.kwargs["description"]
@@ -168,6 +180,7 @@ class TestRunnerManager:
cmd = mock_executor.run.call_args.args[0]
cmd_str = " ".join(cmd)
assert "--ask-become-pass" in cmd_str
assert "--extra-vars" not in cmd_str
def test_update_playbook_not_found(self) -> None:
manager = RunnerManager()
@@ -242,7 +255,7 @@ class TestRunnerManager:
cmd = mock_executor.run.call_args.args[0]
cmd_str = " ".join(cmd)
assert "start-runner.yml" in cmd_str
assert "runner_name=r1" in cmd_str
assert manager._captured_extra_vars["runner_name"] == "r1"
assert "Starting Gitea Runner r1 on host" in mock_executor.run.call_args.kwargs["description"]
def test_start_with_override(self) -> None:
@@ -281,7 +294,7 @@ class TestRunnerManager:
cmd = mock_executor.run.call_args.args[0]
cmd_str = " ".join(cmd)
assert "stop-runner.yml" in cmd_str
assert "runner_name=r1" in cmd_str
assert manager._captured_extra_vars["runner_name"] == "r1"
assert "Stopping Gitea Runner r1 on host" in mock_executor.run.call_args.kwargs["description"]
def test_enable(self) -> None:
@@ -295,7 +308,7 @@ class TestRunnerManager:
cmd = mock_executor.run.call_args.args[0]
cmd_str = " ".join(cmd)
assert "enable-runner.yml" in cmd_str
assert "runner_name=r1" in cmd_str
assert manager._captured_extra_vars["runner_name"] == "r1"
assert "Enabling Gitea Runner r1 on host" in mock_executor.run.call_args.kwargs["description"]
def test_disable(self) -> None:
@@ -314,9 +327,9 @@ class TestRunnerManager:
cmd = mock_executor.run.call_args.args[0]
cmd_str = " ".join(cmd)
assert "disable-runner.yml" in cmd_str
assert "runner_name=r1" in cmd_str
assert "registration_token=tok" in cmd_str
assert "gitea_url=https://git.example.com" in cmd_str
assert manager._captured_extra_vars["runner_name"] == "r1"
assert manager._captured_extra_vars["registration_token"] == "tok"
assert manager._captured_extra_vars["gitea_url"] == "https://git.example.com"
assert "Disabling Gitea Runner r1 on host" in mock_executor.run.call_args.kwargs["description"]
def test_disable_uses_registry_gitea_url(self) -> None:
@@ -332,9 +345,7 @@ class TestRunnerManager:
manager._executor = mock_executor
manager.disable("r1", token="tok")
cmd = mock_executor.run.call_args.args[0]
cmd_str = " ".join(cmd)
assert "gitea_url=https://registry.example.com" in cmd_str
assert manager._captured_extra_vars["gitea_url"] == "https://registry.example.com"
def test_disable_missing_token(self) -> None:
mock_registry = MagicMock()
@@ -361,7 +372,7 @@ class TestRunnerManager:
cmd = mock_executor.run.call_args.args[0]
cmd_str = " ".join(cmd)
assert "status-runner.yml" in cmd_str
assert "runner_name=r1" in cmd_str
assert manager._captured_extra_vars["runner_name"] == "r1"
assert "Checking status of Gitea Runner r1 on host" in mock_executor.run.call_args.kwargs["description"]
def test_remove(self) -> None:
@@ -380,9 +391,9 @@ class TestRunnerManager:
cmd = mock_executor.run.call_args.args[0]
cmd_str = " ".join(cmd)
assert "remove-runner.yml" in cmd_str
assert "runner_name=r1" in cmd_str
assert "registration_token=tok" in cmd_str
assert "gitea_url=https://git.example.com" in cmd_str
assert manager._captured_extra_vars["runner_name"] == "r1"
assert manager._captured_extra_vars["registration_token"] == "tok"
assert manager._captured_extra_vars["gitea_url"] == "https://git.example.com"
assert "Removing Gitea Runner r1 from host" in mock_executor.run.call_args.kwargs["description"]
mock_registry.remove.assert_called_once_with("r1")
@@ -538,33 +549,72 @@ class TestRunnerManager:
assert runners[0]["status"] == "unknown"
class TestExtraVarsFile:
"""Tests for the ``_extra_vars_file`` context manager."""
def test_writes_temp_file_with_content(self) -> None:
manager = RunnerManager()
with manager._extra_vars_file({"foo": "bar", "count": 3}) as path:
assert path is not None
assert path.endswith(".json")
with open(path) as f:
data = json.load(f)
assert data == {"foo": "bar", "count": 3}
mode = os.stat(path).st_mode & 0o777
assert mode == 0o600
assert not os.path.exists(path)
def test_none_yields_none(self) -> None:
manager = RunnerManager()
with manager._extra_vars_file(None) as path:
assert path is None
def test_empty_dict_yields_none(self) -> None:
manager = RunnerManager()
with manager._extra_vars_file({}) as path:
assert path is None
def test_cleans_up_even_if_file_deleted(self) -> None:
manager = RunnerManager()
with manager._extra_vars_file({"foo": "bar"}) as path:
os.unlink(path)
# Should not raise despite missing file on cleanup.
class TestBuildCmd:
def test_build_cmd_basic(self) -> None:
manager = RunnerManager()
with patch.object(Path, "exists", return_value=True):
cmd = manager._build_cmd("test.yml", "host1", "user1", "foo=bar")
cmd = manager._build_cmd("test.yml", "host1", "user1", "/tmp/vars.json")
cmd_str = " ".join(cmd)
assert "ansible-playbook" in cmd_str
assert "test.yml" in cmd_str
assert "host1," in cmd_str
assert "user1" in cmd_str
assert "foo=bar" in cmd_str
assert "--extra-vars" in cmd_str
assert "@/tmp/vars.json" in cmd_str
def test_build_cmd_with_key(self) -> None:
manager = RunnerManager()
with patch.object(Path, "exists", return_value=True):
cmd = manager._build_cmd("test.yml", "host1", "user1", "foo=bar", key="/key")
cmd = manager._build_cmd("test.yml", "host1", "user1", "/tmp/vars.json", key="/key")
assert "--private-key" in cmd
assert "/key" in cmd
def test_build_cmd_ask_become_pass(self) -> None:
manager = RunnerManager()
with patch.object(Path, "exists", return_value=True):
cmd = manager._build_cmd("test.yml", "host1", "user1", "foo=bar", ask_become_pass=True)
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_no_extra_vars(self) -> None:
manager = RunnerManager()
with patch.object(Path, "exists", return_value=True):
cmd = manager._build_cmd("test.yml", "host1", "user1")
assert "--extra-vars" not in cmd
def test_build_cmd_playbook_not_found(self) -> None:
manager = RunnerManager()
with patch.object(Path, "exists", return_value=False):
with pytest.raises(AnsibleError, match="Playbook not found"):
manager._build_cmd("missing.yml", "host1", "user1", "foo=bar")
manager._build_cmd("missing.yml", "host1", "user1", "/tmp/vars.json")
tests/unit/test_validate_commit_msg.py
-1
View File
@@ -31,7 +31,6 @@ class TestHelpers:
assert CONVENTIONAL_RE.match("ci: update workflow")
assert CONVENTIONAL_RE.match("build: update deps")
assert CONVENTIONAL_RE.match("revert: undo change")
assert CONVENTIONAL_RE.match("BREAKING CHANGE: major")
def test_conventional_re_allows_scope(self) -> None:
assert CONVENTIONAL_RE.match("feat(cli): add --url option")