From ef3b882e5b818d19832d19b743f73d9808e9c352 Mon Sep 17 00:00:00 2001 From: emil Date: Thu, 9 Jul 2026 11:51:50 +0000 Subject: [PATCH] DEVX-124: feat: extract shared utilities from infra and grm into devx --- AGENTS.md | 14 ++- src/devx/ci/record_deployed_tag.py | 48 +++++++++ src/devx/ci/validate_deploy_ref.py | 85 ++++++++++++++++ src/devx/translations.json | 32 ++++++ src/devx/utils/confirm.py | 27 +++++ src/devx/utils/crypto.py | 73 +++++++++++++ src/devx/utils/json_registry.py | 128 +++++++++++++++++++++++ src/devx/utils/logging.py | 48 +++++++++ src/devx/utils/network.py | 102 +++++++++++++++++++ src/devx/utils/ssh.py | 132 ++++++++++++++++++++++++ src/devx/utils/step_tracker.py | 102 +++++++++++++++++++ src/devx/utils/vault.py | 135 +++++++++++++++++++++++++ tests/unit/test_record_deployed_tag.py | 62 ++++++++++++ tests/unit/test_utils_confirm.py | 28 +++++ tests/unit/test_utils_crypto.py | 69 +++++++++++++ tests/unit/test_utils_json_registry.py | 110 ++++++++++++++++++++ tests/unit/test_utils_logging.py | 53 ++++++++++ tests/unit/test_utils_network.py | 71 +++++++++++++ tests/unit/test_utils_ssh.py | 96 ++++++++++++++++++ tests/unit/test_utils_step_tracker.py | 124 +++++++++++++++++++++++ tests/unit/test_utils_vault.py | 134 ++++++++++++++++++++++++ tests/unit/test_validate_deploy_ref.py | 70 +++++++++++++ 22 files changed, 1742 insertions(+), 1 deletion(-) create mode 100644 src/devx/ci/record_deployed_tag.py create mode 100644 src/devx/ci/validate_deploy_ref.py create mode 100644 src/devx/utils/confirm.py create mode 100644 src/devx/utils/crypto.py create mode 100644 src/devx/utils/json_registry.py create mode 100644 src/devx/utils/logging.py create mode 100644 src/devx/utils/network.py create mode 100644 src/devx/utils/ssh.py create mode 100644 src/devx/utils/step_tracker.py create mode 100644 src/devx/utils/vault.py create mode 100644 tests/unit/test_record_deployed_tag.py create mode 100644 tests/unit/test_utils_confirm.py create mode 100644 tests/unit/test_utils_crypto.py create mode 100644 tests/unit/test_utils_json_registry.py create mode 100644 tests/unit/test_utils_logging.py create mode 100644 tests/unit/test_utils_network.py create mode 100644 tests/unit/test_utils_ssh.py create mode 100644 tests/unit/test_utils_step_tracker.py create mode 100644 tests/unit/test_utils_vault.py create mode 100644 tests/unit/test_validate_deploy_ref.py diff --git a/AGENTS.md b/AGENTS.md index 2435f6d..b5fee4f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -88,7 +88,9 @@ src/devx/ │ ├── integration_guard.py # Run pytest with cross-runner fail-fast │ ├── check_translations.py # Translation completeness check │ ├── doc_coverage.py # Documentation coverage check -│ └── lint_docs.py # Documentation linter (structure, links, headings, code blocks, orphans) +│ ├── lint_docs.py # Documentation linter (structure, links, headings, code blocks, orphans) +│ ├── validate_deploy_ref.py # Validate git tag for deployments (--github-output) +│ └── record_deployed_tag.py # Record deployed tag to Gitea repo variable ├── tools/ # Developer tooling modules (run locally or by CI) │ ├── setup.py # Environment setup (venv, deps, hooks) │ ├── install_tools.py # Install actionlint, git-cliff, act_runner, tea, hadolint, vale @@ -113,6 +115,16 @@ src/devx/ │ ├── pre_push_check.py # Validate Vikunja task existence before push │ └── _shared.py # Shared tool utilities ├── opentofu.py # OpenTofu output helpers (get_tofu_output, get_tofu_vm_ip, get_tofu_vm_field) +├── utils/ # Shared utilities (reusable across projects) +│ ├── api.py # API response helpers (is_truthy, is_falsy) +│ ├── ssh.py # SSH exec + wait_for_ssh (pure-Python socket check) +│ ├── crypto.py # Secret generation (shell-safe passwords) +│ ├── vault.py # Ansible vault encrypt/decrypt helpers +│ ├── network.py # HTTP connectivity check + wait_for_ssh +│ ├── confirm.py # Typed confirmation validation for destructive ops +│ ├── json_registry.py # File-locked JSON registry for local state +│ ├── step_tracker.py # Multi-step operation tracking with reports +│ └── logging.py # XDG-compliant logging configuration └── molecule/ # Optional molecule testing helpers (for Ansible projects) ├── discover_runners.py # Dynamic Gitea runner discovery ├── distribute_molecule.py # Distribute molecule scenarios across runners (LPT scheduling, --roles-root for multi-role) diff --git a/src/devx/ci/record_deployed_tag.py b/src/devx/ci/record_deployed_tag.py new file mode 100644 index 0000000..69b25b1 --- /dev/null +++ b/src/devx/ci/record_deployed_tag.py @@ -0,0 +1,48 @@ +#!/usr/bin/env python3 +"""Record the deployed git tag for a given environment. + +Writes the tag to a Gitea repository variable so it can be queried +later via the Gitea API or ``devx.ci.get_deployed_tag``. + +Usage:: + + python -m devx.ci.record_deployed_tag --env production --tag v0.28.1 + python -m devx.ci.record_deployed_tag --env staging --tag master-abc1234 +""" + +from __future__ import annotations + +import sys + +import click + +from devx.api_clients import GiteaClient +from devx.config import GITEA_API_URL, REPO_NAME, REPO_OWNER +from devx.i18n import _ +from devx.tokens import get_ci_token + + +@click.command() +@click.option( + "--env", + "env_name", + type=click.Choice(["staging", "production"]), + required=True, +) +@click.option("--tag", required=True, help=_("Git tag or ref that was deployed")) +def main(env_name: str, tag: str) -> None: + """Record the deployed tag for the given environment.""" + try: + token = get_ci_token() + except click.ClickException as exc: + click.echo(f"Error: {exc.message}", err=True) + sys.exit(1) + + var_name = f"{env_name.upper()}_DEPLOY_TAG" + client = GiteaClient(GITEA_API_URL, token, REPO_OWNER, REPO_NAME) + client.set_repo_variable(var_name, tag) + click.echo(f"Recorded {var_name} = {tag}") + + +if __name__ == "__main__": # pragma: no cover + main() diff --git a/src/devx/ci/validate_deploy_ref.py b/src/devx/ci/validate_deploy_ref.py new file mode 100644 index 0000000..fddb6d6 --- /dev/null +++ b/src/devx/ci/validate_deploy_ref.py @@ -0,0 +1,85 @@ +#!/usr/bin/env python3 +"""Resolve and validate the git tag to deploy. + +Shared between staging and production deployments. Ensures a concrete +git tag is used — never a moving branch ref — so deployments are +reproducible and rollback-friendly. + +Usage in workflows:: + + # Production (tag required) + python -m devx.ci.validate_deploy_ref --tag "$TAG" --github-output + + # Staging force-deploy (tag required) + python -m devx.ci.validate_deploy_ref --tag "$TAG" --github-output + + # Staging PR-triggered (PR SHA is already concrete, no tag needed) + python -m devx.ci.validate_deploy_ref --allow-empty --github-output + +Writes ``deploy-ref=`` to ``$GITHUB_OUTPUT`` when ``--github-output`` +is passed, otherwise prints the ref to stdout. +""" + +from __future__ import annotations + +import os +import subprocess # nosec B404 +import sys + +import click + +from devx.i18n import _ + + +@click.command() +@click.option("--tag", default="", help=_("Git tag to deploy (e.g. v0.28.1).")) +@click.option( + "--allow-empty", + is_flag=True, + help=_("Allow empty tag (PR mode where SHA is concrete)."), +) +@click.option( + "--github-output", + is_flag=True, + help=_("Write deploy-ref to $GITHUB_OUTPUT file."), +) +def main(tag: str, allow_empty: bool, github_output: bool) -> None: + """Resolve and validate the deploy ref, exiting non-zero on failure.""" + if not tag: + if not allow_empty: + click.echo( + "::error::No tag specified. Deployments require a concrete git tag " + "(e.g. v0.28.1). Use --allow-empty only for PR-triggered staging deploys " + "where the checkout SHA is already concrete.", + err=True, + ) + sys.exit(1) + ref = "" + click.echo("No tag specified — using checkout ref (PR mode).") + else: + result = subprocess.run( # nosec B603, B607 + ["git", "rev-parse", "-q", "--verify", f"refs/tags/{tag}"], + capture_output=True, + text=True, + check=False, + ) + if result.returncode != 0: + click.echo(f"::error::Tag '{tag}' does not exist in the repository.", err=True) + sys.exit(1) + ref = tag + commit = result.stdout.strip()[:8] + click.echo(f"Deploying tag: {tag} (commit {commit})") + + if github_output: + github_output_path = os.environ.get("GITHUB_OUTPUT") + if not github_output_path: + click.echo("::error::GITHUB_OUTPUT environment variable not set.", err=True) + sys.exit(1) + with open(github_output_path, "a") as f: + f.write(f"deploy-ref={ref}\n") + else: + click.echo(ref) + + +if __name__ == "__main__": # pragma: no cover + main() diff --git a/src/devx/translations.json b/src/devx/translations.json index 5b5f0fc..0ad8d27 100644 --- a/src/devx/translations.json +++ b/src/devx/translations.json @@ -3566,5 +3566,37 @@ "pl": "{separator}", "ru": "{separator}", "zh": "{separator}" + }, + "Allow empty tag (PR mode where SHA is concrete).": { + "bg": "Allow empty tag (PR mode where SHA is concrete).", + "de": "Allow empty tag (PR mode where SHA is concrete).", + "en": "Allow empty tag (PR mode where SHA is concrete).", + "pl": "Allow empty tag (PR mode where SHA is concrete).", + "ru": "Allow empty tag (PR mode where SHA is concrete).", + "zh": "Allow empty tag (PR mode where SHA is concrete)." + }, + "Git tag or ref that was deployed": { + "bg": "Git tag or ref that was deployed", + "de": "Git tag or ref that was deployed", + "en": "Git tag or ref that was deployed", + "pl": "Git tag or ref that was deployed", + "ru": "Git tag or ref that was deployed", + "zh": "Git tag or ref that was deployed" + }, + "Git tag to deploy (e.g. v0.28.1).": { + "bg": "Git tag to deploy (e.g. v0.28.1).", + "de": "Git tag to deploy (e.g. v0.28.1).", + "en": "Git tag to deploy (e.g. v0.28.1).", + "pl": "Git tag to deploy (e.g. v0.28.1).", + "ru": "Git tag to deploy (e.g. v0.28.1).", + "zh": "Git tag to deploy (e.g. v0.28.1)." + }, + "Write deploy-ref to $GITHUB_OUTPUT file.": { + "bg": "Write deploy-ref to $GITHUB_OUTPUT file.", + "de": "Write deploy-ref to $GITHUB_OUTPUT file.", + "en": "Write deploy-ref to $GITHUB_OUTPUT file.", + "pl": "Write deploy-ref to $GITHUB_OUTPUT file.", + "ru": "Write deploy-ref to $GITHUB_OUTPUT file.", + "zh": "Write deploy-ref to $GITHUB_OUTPUT file." } } diff --git a/src/devx/utils/confirm.py b/src/devx/utils/confirm.py new file mode 100644 index 0000000..a85ec41 --- /dev/null +++ b/src/devx/utils/confirm.py @@ -0,0 +1,27 @@ +"""Typed confirmation validation for destructive operations. + +Ensures the user typed an exact confirmation phrase before proceeding +with dangerous operations (e.g. production deploys, database migrations). + +Usage:: + + from devx.utils.confirm import validate_confirmation + + if not validate_confirmation(user_input, expected="deploy-production"): + raise SystemExit("Confirmation does not match") +""" + +from __future__ import annotations + + +def validate_confirmation(confirm: str, expected: str) -> bool: + """Check if confirmation text matches the expected phrase. + + Args: + confirm: The confirmation text entered by the user. + expected: The exact phrase that must be matched. + + Returns: + True if confirmation matches exactly, False otherwise. + """ + return confirm == expected diff --git a/src/devx/utils/crypto.py b/src/devx/utils/crypto.py new file mode 100644 index 0000000..5d2df9f --- /dev/null +++ b/src/devx/utils/crypto.py @@ -0,0 +1,73 @@ +"""Cryptographic secret generation helpers. + +Provides safe secret/password generators that avoid shell-option +interpretation issues (e.g. leading ``-`` being parsed as a flag by +``su -c`` in Docker entrypoints). + +Usage:: + + from devx.utils.crypto import generate_secret, generate_password + + api_key = generate_secret() + db_password = generate_password(length=32) +""" + +from __future__ import annotations + +import secrets + +_SYMBOLS = "!@#$%^&*()-_=+[]{}|;:,.<>?" +_UPPER = "ABCDEFGHIJKLMNOPQRSTUVWXYZ" +_LOWER = "abcdefghijklmnopqrstuvwxyz" +_DIGITS = "0123456789" + + +def generate_secret() -> str: + """Generate a URL-safe secret that never starts with ``-``. + + A leading ``-`` causes passwords to be interpreted as command-line + options when passed through shell expansion chains (e.g. Nextcloud's + Docker entrypoint uses ``su -c`` which strips quoting). + + Returns: + A 43-character URL-safe base64 secret. + """ + value = secrets.token_urlsafe(32) + while value.startswith("-"): + value = secrets.token_urlsafe(32) + return value + + +def generate_password(length: int = 32) -> str: + """Generate a password guaranteed to contain upper, lower, digit, and symbol. + + The first character is always alphanumeric to avoid being interpreted + as a command-line option when passed through shell expansion chains. + + Args: + length: Desired password length (minimum 4). + + Returns: + A password string with guaranteed character class coverage. + """ + pools = [_UPPER, _LOWER, _DIGITS, _SYMBOLS] + chars = [secrets.choice(p) for p in pools] + all_chars = "".join(pools) + chars += [secrets.choice(all_chars) for _ in range(length - len(pools))] + secrets.SystemRandom().shuffle(chars) + while chars[0] in _SYMBOLS: + secrets.SystemRandom().shuffle(chars) + return "".join(chars) + + +def generate_hex_secret(length: int = 32) -> str: + """Generate a hexadecimal secret of the given length. + + Args: + length: Desired number of hex characters (doubled internally + since ``token_hex`` produces pairs). + + Returns: + A hexadecimal string. + """ + return secrets.token_hex(length // 2) diff --git a/src/devx/utils/json_registry.py b/src/devx/utils/json_registry.py new file mode 100644 index 0000000..187c4ad --- /dev/null +++ b/src/devx/utils/json_registry.py @@ -0,0 +1,128 @@ +"""File-locked JSON registry for local state management. + +Provides a simple JSON-backed key-value store with ``fcntl`` file +locking for safe concurrent access. Useful for CLI tools that need +to track remote resources (runners, VMs, deployments) on the local +machine. + +Usage:: + + from devx.utils.json_registry import JsonRegistry + + registry = JsonRegistry(Path("~/.local/share/myapp/state.json")) + registry.add("item1", host="10.0.0.1", user="deploy") + info = registry.get("item1") + registry.remove("item1") +""" + +from __future__ import annotations + +import copy +import fcntl +import json +from datetime import UTC, datetime +from pathlib import Path +from typing import Any, cast + + +class JsonRegistry: + """Manages a local JSON file mapping names to arbitrary metadata. + + Uses ``fcntl`` for file locking (shared lock for reads, exclusive + lock for writes) to prevent race conditions in concurrent scenarios. + """ + + def __init__(self, path: Path | None = None) -> None: + """Initialise the registry. + + Args: + path: Path to the JSON file. Defaults to + ``~/.local/share/devx/registry.json``. + """ + self._path = path or Path.home() / ".local" / "share" / "devx" / "registry.json" + self._data: dict[str, dict[str, Any]] = self._load() + + def _load(self) -> dict[str, dict[str, Any]]: + 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) + 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: + 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, name: str, **fields: Any) -> None: + """Register or overwrite an entry in the registry. + + Args: + name: Unique key for the entry. + **fields: Arbitrary metadata fields to store. + """ + self._data[name] = { + **fields, + "created_at": datetime.now(UTC).isoformat(), + } + self._save() + + def get(self, name: str) -> dict[str, Any] | None: + """Retrieve entry metadata by name. + + Args: + name: Key to look up. + + Returns: + A copy of the entry's metadata, or None if not found. + """ + info = self._data.get(name) + if info: + return copy.deepcopy(info) + return None + + def remove(self, name: str) -> None: + """Remove an entry from the registry. + + Args: + name: Key to remove. No-op if not found. + """ + if name in self._data: + del self._data[name] + self._save() + + def list(self) -> dict[str, dict[str, Any]]: + """Return a copy of all registered entries. + + Returns: + Dict mapping names to metadata copies. + """ + return {name: copy.deepcopy(info) for name, info in self._data.items()} + + def update(self, name: str, **fields: Any) -> None: + """Update fields for an existing entry. + + Args: + name: Key to update. + **fields: Fields to update (None values are skipped). + + Raises: + KeyError: If the entry doesn't exist. + """ + if name not in self._data: + raise KeyError(name) + self._data[name].update({k: v for k, v in fields.items() if v is not None}) + self._save() diff --git a/src/devx/utils/logging.py b/src/devx/utils/logging.py new file mode 100644 index 0000000..57f835f --- /dev/null +++ b/src/devx/utils/logging.py @@ -0,0 +1,48 @@ +"""XDG-compliant logging configuration for CLI tools. + +Provides a standardised logging setup that writes to +``~/.local/state//logs/.log`` following the XDG state +directory specification. Console output is handled separately by +the application (e.g. via ``click.echo``). + +Usage:: + + from devx.utils.logging import get_logger + + logger = get_logger("myapp") + logger.info("Application started") +""" + +from __future__ import annotations + +import logging +from pathlib import Path + + +def get_logger(name: str = "devx") -> logging.Logger: + """Return a configured logger that writes to an XDG state directory. + + All messages (including DEBUG) are written to + ``~/.local/state//logs/.log``. Console output is + expected to be handled by the application via ``click.echo``. + + Args: + name: Logger name and subdirectory name for log files. + + Returns: + A configured :class:`logging.Logger` instance. + """ + logger = logging.getLogger(name) + if logger.handlers: + return logger + + logger.setLevel(logging.DEBUG) + + log_dir = Path.home() / ".local" / "state" / name / "logs" + log_dir.mkdir(parents=True, exist_ok=True) + file_handler = logging.FileHandler(log_dir / f"{name}.log") + file_handler.setLevel(logging.DEBUG) + file_handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s")) + logger.addHandler(file_handler) + + return logger diff --git a/src/devx/utils/network.py b/src/devx/utils/network.py new file mode 100644 index 0000000..afb8c0e --- /dev/null +++ b/src/devx/utils/network.py @@ -0,0 +1,102 @@ +"""Network connectivity helpers. + +Provides retry-aware HTTP connectivity checks and SSH availability +checks for deployment workflows. Uses ``tenacity`` for exponential +backoff retry logic. + +Usage:: + + from devx.utils.network import check_http_connectivity, wait_for_ssh + + check_http_connectivity("https://auth.example.com") + wait_for_ssh("178.105.254.83") +""" + +from __future__ import annotations + +import logging +import socket +import time +from collections.abc import Callable + +import requests +from tenacity import ( + Retrying, + before_sleep_log, + retry_if_exception_type, + stop_after_attempt, + wait_exponential, +) + + +def check_http_connectivity( + base_url: str, + max_attempts: int = 30, + *, + verify: bool = True, + sleep: Callable[[float], None] | None = None, +) -> None: + """Verify HTTP reachability of *base_url* with retry. + + Uses tenacity for retry with exponential backoff (2 s min, 10 s max). + + Args: + base_url: URL to check via GET request. + max_attempts: Maximum retry attempts. + verify: Whether to verify TLS certificates. + sleep: Custom sleep function for testing (defaults to ``time.sleep``). + + Raises: + requests.exceptions.ConnectionError: If the URL is not reachable + after *max_attempts*. + """ + retrying = Retrying( + stop=stop_after_attempt(max_attempts), + wait=wait_exponential(multiplier=2, min=2, max=10), + retry=retry_if_exception_type(requests.exceptions.ConnectionError), + before_sleep=before_sleep_log(logging.getLogger("devx.utils.network"), logging.WARNING), + sleep=sleep if sleep is not None else time.sleep, + reraise=True, + ) + + def _check() -> None: + requests.get(base_url, timeout=10, verify=verify) # nosec B501 + + retrying(_check) + + +def wait_for_ssh( + host: str, + port: int = 22, + max_attempts: int = 30, + interval: int = 10, + *, + sleep: Callable[[float], None] | None = None, +) -> None: + """Wait for SSH to be available on a host using a pure-Python socket check. + + Uses socket instead of ``nc(1)`` so it works on CI runners without + netcat. Uses exponential backoff: starts at 2 s, doubles each + attempt up to 10 s max. + + Args: + host: VM IP address or hostname. + port: SSH port (default 22). + max_attempts: Maximum number of connection attempts. + interval: Base interval for backoff calculation (seconds). + sleep: Custom sleep function for testing (defaults to ``time.sleep``). + + Raises: + RuntimeError: If SSH is not available after *max_attempts*. + """ + _sleep = sleep if sleep is not None else time.sleep + for i in range(max_attempts): + try: + with socket.create_connection((host, port), timeout=5): + return + except OSError: + pass + if i < max_attempts - 1: + wait = min(2 * (2**i), 10) + _sleep(wait) + raise RuntimeError(f"SSH not available on {host}:{port} after {max_attempts} attempts") diff --git a/src/devx/utils/ssh.py b/src/devx/utils/ssh.py new file mode 100644 index 0000000..aa27295 --- /dev/null +++ b/src/devx/utils/ssh.py @@ -0,0 +1,132 @@ +"""SSH helpers for running commands on remote hosts. + +Provides a simple wrapper around the ``ssh`` CLI for executing commands +on remote machines (e.g. customer VMs, CI runners) without requiring +Ansible. Includes a pure-Python ``wait_for_ssh`` that uses socket +instead of ``nc(1)`` so it works on minimal CI containers. + +Usage:: + + from devx.utils.ssh import ssh_exec, wait_for_ssh + + wait_for_ssh("178.105.254.83") + result = ssh_exec("178.105.254.83", "uname -a") + print(result.stdout) +""" + +from __future__ import annotations + +import socket +import subprocess # nosec B404 +import sys +import time + +SSH_CONNECT_TIMEOUT = "10" +SSH_HOST_KEY_CHECKING = "no" + + +def ssh_exec( + host: str, + command: str, + *, + user: str = "deploy", + timeout: int = 30, + check: bool = True, +) -> subprocess.CompletedProcess[str]: + """Run *command* on *host* via SSH and return the result. + + Args: + host: VM IP address or hostname. + command: Shell command to execute on the remote host. + user: SSH user (default ``deploy``). + timeout: Subprocess timeout in seconds. + check: If True, raise ``CalledProcessError`` on non-zero exit. + + Returns: + The completed process result with stdout/stderr captured. + """ + result = subprocess.run( # nosec B603, B607, B607 + [ + "ssh", + "-o", + f"StrictHostKeyChecking={SSH_HOST_KEY_CHECKING}", + "-o", + f"ConnectTimeout={SSH_CONNECT_TIMEOUT}", + f"{user}@{host}", + command, + ], + capture_output=True, + text=True, + check=False, + timeout=timeout, + ) + if check and result.returncode != 0: + print(f"SSH command failed on {host}: {command}", file=sys.stderr) + print(f" stdout: {result.stdout.strip()}", file=sys.stderr) + print(f" stderr: {result.stderr.strip()}", file=sys.stderr) + result.check_returncode() + return result + + +def docker_exec_on_vm( + host: str, + container: str, + command: str, + *, + user: str = "deploy", + db_user: str | None = None, + db_name: str | None = None, + timeout: int = 30, +) -> str: + """Run a command inside a Docker container on a remote VM via SSH. + + For PostgreSQL commands, set *db_user* and *db_name* to run + ``psql -U -d -c `` inside the container. + + Args: + host: VM IP address or hostname. + container: Docker container name on the remote host. + command: Command to execute inside the container (or SQL if db_user/db_name set). + user: SSH user (default ``deploy``). + db_user: PostgreSQL user name (enables psql mode). + db_name: PostgreSQL database name (enables psql mode). + timeout: Subprocess timeout in seconds. + + Returns: + Stripped stdout from the command. + """ + if db_user and db_name: + escaped_sql = command.replace("'", "'\"'\"'") + remote_cmd = f'docker exec {container} psql -U {db_user} -d {db_name} -t -A -c "{escaped_sql}"' + else: + remote_cmd = f"docker exec {container} {command}" + result = ssh_exec(host, remote_cmd, user=user, timeout=timeout) + return result.stdout.strip() + + +def wait_for_ssh(host: str, port: int = 22, max_attempts: int = 30, interval: int = 10) -> None: + """Wait for SSH to be available on a host using a pure-Python socket check. + + Uses socket instead of ``nc(1)`` so it works on CI runners without + netcat. Uses exponential backoff: starts at 2 s, doubles each + attempt up to 10 s max. + + Args: + host: VM IP address or hostname. + port: SSH port (default 22). + max_attempts: Maximum number of connection attempts. + interval: Base interval for backoff calculation (seconds). + + Raises: + RuntimeError: If SSH is not available after *max_attempts*. + """ + for i in range(max_attempts): + try: + with socket.create_connection((host, port), timeout=5): + return + except OSError: + pass + if i < max_attempts - 1: + wait = min(2 * (2**i), 10) + time.sleep(wait) + raise RuntimeError(f"SSH not available on {host}:{port} after {max_attempts} attempts") diff --git a/src/devx/utils/step_tracker.py b/src/devx/utils/step_tracker.py new file mode 100644 index 0000000..f657778 --- /dev/null +++ b/src/devx/utils/step_tracker.py @@ -0,0 +1,102 @@ +"""Operation step tracking with translated reports. + +Provides a context manager that tracks multi-step operations and prints +a status report on exit. Steps are marked as pending, in_progress, +completed, or failed. On exception, the last in-progress step is +marked as failed. + +Usage:: + + from devx.utils.step_tracker import track_steps + + with track_steps() as tracker: + tracker.begin("Install dependencies") + install_deps() + tracker.done() + + tracker.begin("Run tests") + run_tests() + tracker.done() +""" + +from __future__ import annotations + +from collections.abc import Generator +from contextlib import contextmanager + +import click + +_STATUS_ICONS = { + "completed": "✓", + "failed": "✗", + "pending": "○", + "in_progress": "◌", +} + +_STATUS_COLORS = { + "completed": "green", + "failed": "red", + "in_progress": "yellow", + "pending": "white", +} + + +class Step: + """A single tracked step in an operation.""" + + def __init__(self, name: str) -> None: + self.name = name + self.status = "pending" + + +class StepTracker: + """Tracks steps of an operation and prints a report on exit.""" + + def __init__(self) -> None: + self.steps: list[Step] = [] + + def begin(self, name: str) -> None: + """Start a new step. + + Args: + name: Human-readable step name. + """ + step = Step(name) + self.steps.append(step) + step.status = "in_progress" + + def done(self) -> None: + """Mark the most recent in-progress step as completed.""" + if self.steps and self.steps[-1].status == "in_progress": + self.steps[-1].status = "completed" + + +@contextmanager +def track_steps() -> Generator[StepTracker, None, None]: + """Context manager that tracks steps and prints a report on exit. + + On exception the last in-progress step is marked as failed. + The report is printed in the ``finally`` block so it always appears. + + Yields: + A :class:`StepTracker` instance to track steps with. + """ + tracker = StepTracker() + try: + yield tracker + except Exception: + for step in reversed(tracker.steps): + if step.status == "in_progress": + step.status = "failed" + raise + finally: + _print_report(tracker.steps) + + +def _print_report(steps: list[Step]) -> None: + """Print an operation report to stdout.""" + click.secho("=== Operation Report ===", fg="bright_cyan") + for step in steps: + icon = _STATUS_ICONS.get(step.status, "?") + color = _STATUS_COLORS.get(step.status) + click.secho(f" {icon} {step.name} ({step.status})", fg=color) diff --git a/src/devx/utils/vault.py b/src/devx/utils/vault.py new file mode 100644 index 0000000..5396fd5 --- /dev/null +++ b/src/devx/utils/vault.py @@ -0,0 +1,135 @@ +"""Ansible Vault helpers for encrypting and decrypting YAML files. + +Wraps ``ansible-vault`` to provide a convenient API for loading and +saving vault-encrypted YAML files. Falls back to plain YAML when no +vault-password file is available, making it safe to use in both +local (with vault) and CI (without vault) environments. + +Usage:: + + from devx.utils.vault import load_vault_yaml, save_vault_yaml + + data = load_vault_yaml(Path("secrets.yml"), vault_pass=Path("vault-password")) + data["new_key"] = "value" + save_vault_yaml(Path("secrets.yml"), data, vault_pass=Path("vault-password")) +""" + +from __future__ import annotations + +import subprocess # nosec B404 +from pathlib import Path + +import yaml + + +def encrypt_file(path: Path, vault_pass: Path) -> None: + """Encrypt a file in-place using ansible-vault. + + Args: + path: File to encrypt. + vault_pass: Path to the vault-password file. + """ + subprocess.run( # nosec B603, B607 + [ + "ansible-vault", + "encrypt", + str(path), + "--vault-password-file", + str(vault_pass), + "--encrypt-vault-id", + "default", + ], + check=True, + ) + + +def decrypt_file(path: Path, vault_pass: Path) -> None: + """Decrypt a file in-place using ansible-vault. + + Args: + path: File to decrypt. + vault_pass: Path to the vault-password file. + """ + subprocess.run( # nosec B603, B607 + [ + "ansible-vault", + "decrypt", + str(path), + "--vault-password-file", + str(vault_pass), + ], + check=True, + ) + + +def load_vault_yaml(path: Path, vault_pass: Path | None = None) -> dict: + """Load a YAML file, decrypting with ansible-vault if vault-password exists. + + If *vault_pass* is None or doesn't exist, the file is read as plain + YAML. If decryption fails (file not vault-encrypted), it falls back + to plain YAML. + + Args: + path: YAML file path. + vault_pass: Path to the vault-password file (optional). + + Returns: + Parsed YAML content as a dict (empty dict if file is empty). + """ + if vault_pass is None or not vault_pass.exists(): + with open(path, encoding="utf-8") as f: + return yaml.safe_load(f) or {} + result = subprocess.run( # nosec B603, B607 + ["ansible-vault", "view", str(path), "--vault-password-file", str(vault_pass)], + capture_output=True, + text=True, + check=False, + ) + if result.returncode == 0: + return yaml.safe_load(result.stdout) or {} + if "is not vault encrypted" in result.stderr: + with open(path, encoding="utf-8") as f: + return yaml.safe_load(f) or {} + result.check_returncode() # pragma: no cover + return {} # pragma: no cover + + +def save_vault_yaml(path: Path, data: dict, vault_pass: Path | None = None) -> None: + """Write YAML data, encrypting with ansible-vault if vault-password exists. + + Args: + path: Destination YAML file path. + data: Data to serialize. + vault_pass: Path to the vault-password file (optional). + """ + plain = yaml.dump(data, default_flow_style=False, sort_keys=False) + with open(path, "w", encoding="utf-8") as f: + f.write(plain) + if vault_pass is not None and vault_pass.exists(): + subprocess.run( # nosec B603, B607 + [ + "ansible-vault", + "encrypt", + str(path), + "--vault-password-file", + str(vault_pass), + "--encrypt-vault-id", + "default", + ], + capture_output=True, + check=True, + ) + + +def is_encrypted(path: Path) -> bool: + """Check if a file is ansible-vault encrypted. + + Args: + path: File to check. + + Returns: + True if the file starts with the ``$ANSIBLE_VAULT`` marker. + """ + with open(path, encoding="utf-8") as f: + first_line = f.readline() + return "$ANSIBLE_VAULT" in first_line diff --git a/tests/unit/test_record_deployed_tag.py b/tests/unit/test_record_deployed_tag.py new file mode 100644 index 0000000..b3106d5 --- /dev/null +++ b/tests/unit/test_record_deployed_tag.py @@ -0,0 +1,62 @@ +"""Unit tests for devx.ci.record_deployed_tag.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +from click.testing import CliRunner + +from devx.ci.record_deployed_tag import main + + +class TestRecordDeployedTag: + @patch("devx.ci.record_deployed_tag.GiteaClient") + @patch("devx.ci.record_deployed_tag.get_ci_token") + def test_records_production_tag(self, mock_token: MagicMock, mock_client: MagicMock) -> None: + mock_token.return_value = "fake-token" + client_instance = MagicMock() + mock_client.return_value = client_instance + + runner = CliRunner() + result = runner.invoke(main, ["--env", "production", "--tag", "v1.0.0"]) + + assert result.exit_code == 0 + assert "PRODUCTION_DEPLOY_TAG" in result.output + assert "v1.0.0" in result.output + client_instance.set_repo_variable.assert_called_once_with("PRODUCTION_DEPLOY_TAG", "v1.0.0") + + @patch("devx.ci.record_deployed_tag.GiteaClient") + @patch("devx.ci.record_deployed_tag.get_ci_token") + def test_records_staging_tag(self, mock_token: MagicMock, mock_client: MagicMock) -> None: + mock_token.return_value = "fake-token" + client_instance = MagicMock() + mock_client.return_value = client_instance + + runner = CliRunner() + result = runner.invoke(main, ["--env", "staging", "--tag", "master-abc123"]) + + assert result.exit_code == 0 + assert "STAGING_DEPLOY_TAG" in result.output + client_instance.set_repo_variable.assert_called_once_with("STAGING_DEPLOY_TAG", "master-abc123") + + @patch("devx.ci.record_deployed_tag.get_ci_token") + def test_token_error_exits_nonzero(self, mock_token: MagicMock) -> None: + import click + + mock_token.side_effect = click.ClickException("No token available") + + runner = CliRunner() + result = runner.invoke(main, ["--env", "production", "--tag", "v1.0.0"]) + + assert result.exit_code == 1 + assert "No token available" in result.output + + def test_invalid_env_choice(self) -> None: + runner = CliRunner() + result = runner.invoke(main, ["--env", "invalid", "--tag", "v1.0.0"]) + assert result.exit_code != 0 + + def test_missing_tag_option(self) -> None: + runner = CliRunner() + result = runner.invoke(main, ["--env", "production"]) + assert result.exit_code != 0 diff --git a/tests/unit/test_utils_confirm.py b/tests/unit/test_utils_confirm.py new file mode 100644 index 0000000..115e468 --- /dev/null +++ b/tests/unit/test_utils_confirm.py @@ -0,0 +1,28 @@ +"""Unit tests for devx.utils.confirm.""" + +from __future__ import annotations + +from devx.utils.confirm import validate_confirmation + + +class TestValidateConfirmation: + def test_exact_match(self) -> None: + assert validate_confirmation("deploy-production", "deploy-production") is True + + def test_mismatch(self) -> None: + assert validate_confirmation("deploy-staging", "deploy-production") is False + + def test_empty_string(self) -> None: + assert validate_confirmation("", "deploy-production") is False + + def test_case_sensitive(self) -> None: + assert validate_confirmation("Deploy-Production", "deploy-production") is False + + def test_partial_match(self) -> None: + assert validate_confirmation("deploy", "deploy-production") is False + + def test_extra_whitespace(self) -> None: + assert validate_confirmation("deploy-production ", "deploy-production") is False + + def test_custom_expected(self) -> None: + assert validate_confirmation("yes-delete-all", "yes-delete-all") is True diff --git a/tests/unit/test_utils_crypto.py b/tests/unit/test_utils_crypto.py new file mode 100644 index 0000000..bb53f8c --- /dev/null +++ b/tests/unit/test_utils_crypto.py @@ -0,0 +1,69 @@ +"""Unit tests for devx.utils.crypto.""" + +from __future__ import annotations + +import re + +from devx.utils.crypto import ( + _DIGITS, + _LOWER, + _SYMBOLS, + _UPPER, + generate_hex_secret, + generate_password, + generate_secret, +) + + +class TestGenerateSecret: + def test_returns_url_safe_string(self) -> None: + secret = generate_secret() + assert isinstance(secret, str) + assert len(secret) > 0 + # URL-safe base64 characters only + assert re.match(r"^[A-Za-z0-9_-]+$", secret) + + def test_never_starts_with_dash(self) -> None: + for _ in range(1000): + secret = generate_secret() + assert not secret.startswith("-") + + +class TestGeneratePassword: + def test_default_length(self) -> None: + pw = generate_password() + assert len(pw) == 32 + + def test_custom_length(self) -> None: + pw = generate_password(length=64) + assert len(pw) == 64 + + def test_contains_all_char_classes(self) -> None: + pw = generate_password(length=32) + assert any(c in _UPPER for c in pw), "Missing uppercase" + assert any(c in _LOWER for c in pw), "Missing lowercase" + assert any(c in _DIGITS for c in pw), "Missing digits" + assert any(c in _SYMBOLS for c in pw), "Missing symbols" + + def test_first_char_alphanumeric(self) -> None: + for _ in range(1000): + pw = generate_password() + assert pw[0] not in _SYMBOLS, f"First char '{pw[0]}' is a symbol" + + def test_minimum_length_4(self) -> None: + pw = generate_password(length=4) + assert len(pw) == 4 + + +class TestGenerateHexSecret: + def test_returns_hex_string(self) -> None: + secret = generate_hex_secret(length=32) + assert re.match(r"^[0-9a-f]+$", secret) + + def test_correct_length(self) -> None: + secret = generate_hex_secret(length=20) + assert len(secret) == 20 + + def test_empty_for_zero(self) -> None: + secret = generate_hex_secret(length=0) + assert secret == "" diff --git a/tests/unit/test_utils_json_registry.py b/tests/unit/test_utils_json_registry.py new file mode 100644 index 0000000..820e445 --- /dev/null +++ b/tests/unit/test_utils_json_registry.py @@ -0,0 +1,110 @@ +"""Unit tests for devx.utils.json_registry.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from devx.utils.json_registry import JsonRegistry + + +class TestJsonRegistry: + def test_add_and_get(self, tmp_path: Path) -> None: + reg = JsonRegistry(tmp_path / "state.json") + reg.add("item1", host="10.0.0.1", user="deploy") + info = reg.get("item1") + assert info is not None + assert info["host"] == "10.0.0.1" + assert info["user"] == "deploy" + assert "created_at" in info + + def test_get_nonexistent(self, tmp_path: Path) -> None: + reg = JsonRegistry(tmp_path / "state.json") + assert reg.get("nope") is None + + def test_remove(self, tmp_path: Path) -> None: + reg = JsonRegistry(tmp_path / "state.json") + reg.add("item1", host="10.0.0.1") + reg.remove("item1") + assert reg.get("item1") is None + + def test_remove_nonexistent_is_noop(self, tmp_path: Path) -> None: + reg = JsonRegistry(tmp_path / "state.json") + reg.remove("nonexistent") # should not raise + + def test_list(self, tmp_path: Path) -> None: + reg = JsonRegistry(tmp_path / "state.json") + reg.add("a", host="1.1.1.1") + reg.add("b", host="2.2.2.2") + items = reg.list() + assert set(items.keys()) == {"a", "b"} + assert items["a"]["host"] == "1.1.1.1" + + def test_list_empty(self, tmp_path: Path) -> None: + reg = JsonRegistry(tmp_path / "state.json") + assert reg.list() == {} + + def test_update_existing(self, tmp_path: Path) -> None: + reg = JsonRegistry(tmp_path / "state.json") + reg.add("item", host="1.1.1.1", status="active") + reg.update("item", status="inactive") + info = reg.get("item") + assert info["status"] == "inactive" + assert info["host"] == "1.1.1.1" # unchanged + + def test_update_nonexistent_raises(self, tmp_path: Path) -> None: + reg = JsonRegistry(tmp_path / "state.json") + with pytest.raises(KeyError): + reg.update("nonexistent", host="1.1.1.1") + + def test_update_skips_none_values(self, tmp_path: Path) -> None: + reg = JsonRegistry(tmp_path / "state.json") + reg.add("item", host="1.1.1.1") + reg.update("item", host=None, status="active") + info = reg.get("item") + assert info["host"] == "1.1.1.1" # not overwritten by None + assert info["status"] == "active" + + def test_persistence_across_instances(self, tmp_path: Path) -> None: + path = tmp_path / "state.json" + reg1 = JsonRegistry(path) + reg1.add("item", host="10.0.0.1") + reg2 = JsonRegistry(path) + info = reg2.get("item") + assert info is not None + assert info["host"] == "10.0.0.1" + + def test_overwrite_existing(self, tmp_path: Path) -> None: + reg = JsonRegistry(tmp_path / "state.json") + reg.add("item", host="1.1.1.1") + reg.add("item", host="2.2.2.2") + info = reg.get("item") + assert info["host"] == "2.2.2.2" + + def test_corrupt_json_returns_empty(self, tmp_path: Path) -> None: + path = tmp_path / "state.json" + path.write_text("{invalid json") + reg = JsonRegistry(path) + assert reg.list() == {} + + def test_nonexistent_file_returns_empty(self, tmp_path: Path) -> None: + reg = JsonRegistry(tmp_path / "nonexistent.json") + assert reg.list() == {} + + def test_creates_parent_dirs(self, tmp_path: Path) -> None: + path = tmp_path / "subdir" / "deeper" / "state.json" + reg = JsonRegistry(path) + reg.add("item", host="1.1.1.1") + assert path.exists() + + def test_get_returns_copy(self, tmp_path: Path) -> None: + reg = JsonRegistry(tmp_path / "state.json") + reg.add("item", host="1.1.1.1", tags=["a", "b"]) + info = reg.get("item") + assert info is not None + info["tags"].append("c") + # Original should be unchanged + info2 = reg.get("item") + assert info2 is not None + assert info2["tags"] == ["a", "b"] diff --git a/tests/unit/test_utils_logging.py b/tests/unit/test_utils_logging.py new file mode 100644 index 0000000..d18f096 --- /dev/null +++ b/tests/unit/test_utils_logging.py @@ -0,0 +1,53 @@ +"""Unit tests for devx.utils.logging.""" + +from __future__ import annotations + +import logging +from pathlib import Path +from unittest.mock import patch + +from devx.utils.logging import get_logger + + +class TestGetLogger: + def test_returns_logger_with_handlers(self) -> None: + logger = get_logger("test_devx_unit_1") + assert logger.handlers + assert isinstance(logger.handlers[0], logging.FileHandler) + + def test_idempotent(self) -> None: + logger1 = get_logger("test_devx_unit_2") + initial_count = len(logger1.handlers) + logger2 = get_logger("test_devx_unit_2") + assert logger1 is logger2 + assert len(logger2.handlers) == initial_count + + def test_log_level_is_debug(self) -> None: + logger = get_logger("test_devx_unit_3") + assert logger.level == logging.DEBUG + + def test_file_handler_level_is_debug(self) -> None: + logger = get_logger("test_devx_unit_4") + file_handler = logger.handlers[0] + assert file_handler.level == logging.DEBUG + + def test_default_name(self) -> None: + logger = get_logger() + assert logger.name == "devx" + + def test_creates_log_directory(self, tmp_path: Path) -> None: + with patch.object(Path, "home", return_value=tmp_path): + get_logger("test_app_creates_dir") + log_dir = tmp_path / ".local" / "state" / "test_app_creates_dir" / "logs" + assert log_dir.exists() + assert (log_dir / "test_app_creates_dir.log").exists() + + def test_formatter_includes_timestamp(self) -> None: + logger = get_logger("test_devx_unit_5") + file_handler = logger.handlers[0] + fmt = file_handler.formatter + assert fmt is not None + assert "%(asctime)s" in fmt._fmt + assert "%(levelname)s" in fmt._fmt + assert "%(name)s" in fmt._fmt + assert "%(message)s" in fmt._fmt diff --git a/tests/unit/test_utils_network.py b/tests/unit/test_utils_network.py new file mode 100644 index 0000000..7f2934e --- /dev/null +++ b/tests/unit/test_utils_network.py @@ -0,0 +1,71 @@ +"""Unit tests for devx.utils.network.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from devx.utils.network import check_http_connectivity, wait_for_ssh + +_no_sleep = MagicMock() + + +class TestCheckHttpConnectivity: + @patch("devx.utils.network.requests.get") + def test_success(self, mock_get: MagicMock) -> None: + mock_get.return_value = MagicMock(status_code=200) + check_http_connectivity("https://example.com", max_attempts=3) + mock_get.assert_called_once() + + @patch("devx.utils.network.requests.get") + def test_retries_on_connection_error(self, mock_get: MagicMock) -> None: + mock_get.side_effect = [ + requests.exceptions.ConnectionError("refused"), + requests.exceptions.ConnectionError("refused"), + MagicMock(status_code=200), + ] + check_http_connectivity("https://example.com", max_attempts=5, sleep=_no_sleep) + assert mock_get.call_count == 3 + + @patch("devx.utils.network.requests.get") + def test_raises_after_max_attempts(self, mock_get: MagicMock) -> None: + mock_get.side_effect = requests.exceptions.ConnectionError("refused") + with pytest.raises(requests.exceptions.ConnectionError): + check_http_connectivity("https://example.com", max_attempts=2, sleep=_no_sleep) + assert mock_get.call_count == 2 + + @patch("devx.utils.network.requests.get") + def test_verify_false(self, mock_get: MagicMock) -> None: + mock_get.return_value = MagicMock(status_code=200) + check_http_connectivity("https://example.com", verify=False) + mock_get.assert_called_once_with("https://example.com", timeout=10, verify=False) + + +class TestWaitForSsh: + @patch("devx.utils.network.socket.create_connection") + def test_immediate_success(self, mock_conn: MagicMock) -> None: + mock_conn.return_value.__enter__ = MagicMock() + mock_conn.return_value.__exit__ = MagicMock(return_value=False) + wait_for_ssh("10.0.0.1") + mock_conn.assert_called_once() + + @patch("devx.utils.network.socket.create_connection") + def test_retries_until_success(self, mock_conn: MagicMock) -> None: + mock_conn.side_effect = [ + OSError("refused"), + OSError("refused"), + MagicMock(), + ] + mock_conn.return_value.__enter__ = MagicMock() + mock_conn.return_value.__exit__ = MagicMock(return_value=False) + wait_for_ssh("10.0.0.1", max_attempts=5, sleep=_no_sleep) + assert mock_conn.call_count == 3 + + @patch("devx.utils.network.socket.create_connection") + def test_timeout_after_max_attempts(self, mock_conn: MagicMock) -> None: + mock_conn.side_effect = OSError("refused") + with pytest.raises(RuntimeError, match="SSH not available"): + wait_for_ssh("10.0.0.1", max_attempts=3, sleep=_no_sleep) + assert mock_conn.call_count == 3 diff --git a/tests/unit/test_utils_ssh.py b/tests/unit/test_utils_ssh.py new file mode 100644 index 0000000..dea0a88 --- /dev/null +++ b/tests/unit/test_utils_ssh.py @@ -0,0 +1,96 @@ +"""Unit tests for devx.utils.ssh.""" + +from __future__ import annotations + +import subprocess +from unittest.mock import MagicMock, patch + +import pytest + +from devx.utils.ssh import docker_exec_on_vm, ssh_exec, wait_for_ssh + + +class TestSshExec: + @patch("devx.utils.ssh.subprocess.run") + def test_success(self, mock_run: MagicMock) -> None: + mock_run.return_value = MagicMock(returncode=0, stdout="ok", stderr="") + result = ssh_exec("10.0.0.1", "uname -a") + assert result.returncode == 0 + mock_run.assert_called_once() + + @patch("devx.utils.ssh.subprocess.run") + def test_failure_with_check(self, mock_run: MagicMock) -> None: + mock_result = MagicMock(returncode=1, stdout="", stderr="error") + mock_result.check_returncode.side_effect = subprocess.CalledProcessError(1, "ssh") + mock_run.return_value = mock_result + with pytest.raises(subprocess.CalledProcessError): + ssh_exec("10.0.0.1", "false") + + @patch("devx.utils.ssh.subprocess.run") + def test_failure_without_check(self, mock_run: MagicMock) -> None: + mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="error") + result = ssh_exec("10.0.0.1", "false", check=False) + assert result.returncode == 1 + + @patch("devx.utils.ssh.subprocess.run") + def test_custom_user(self, mock_run: MagicMock) -> None: + mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="") + ssh_exec("10.0.0.1", "whoami", user="root") + cmd = mock_run.call_args[0][0] + assert "root@10.0.0.1" in cmd + + +class TestDockerExecOnVm: + @patch("devx.utils.ssh.ssh_exec") + def test_simple_command(self, mock_ssh: MagicMock) -> None: + mock_ssh.return_value = MagicMock(stdout="output\n") + result = docker_exec_on_vm("10.0.0.1", "mycontainer", "ls /") + assert result == "output" + mock_ssh.assert_called_once_with("10.0.0.1", "docker exec mycontainer ls /", user="deploy", timeout=30) + + @patch("devx.utils.ssh.ssh_exec") + def test_psql_mode(self, mock_ssh: MagicMock) -> None: + mock_ssh.return_value = MagicMock(stdout="result\n") + result = docker_exec_on_vm("10.0.0.1", "db", "SELECT 1", db_user="postgres", db_name="mydb") + assert result == "result" + call_args = mock_ssh.call_args[0][1] + assert "psql -U postgres -d mydb" in call_args + assert "SELECT 1" in call_args + + @patch("devx.utils.ssh.ssh_exec") + def test_psql_escapes_single_quotes(self, mock_ssh: MagicMock) -> None: + mock_ssh.return_value = MagicMock(stdout="\n") + docker_exec_on_vm("10.0.0.1", "db", "SELECT 'it''s ok'", db_user="pg", db_name="db") + call_args = mock_ssh.call_args[0][1] + assert "'\"'\"'" in call_args + + +class TestWaitForSsh: + @patch("devx.utils.ssh.socket.create_connection") + def test_immediate_success(self, mock_conn: MagicMock) -> None: + mock_conn.return_value.__enter__ = MagicMock() + mock_conn.return_value.__exit__ = MagicMock(return_value=False) + wait_for_ssh("10.0.0.1") + mock_conn.assert_called_once() + + @patch("devx.utils.ssh.socket.create_connection") + @patch("devx.utils.ssh.time.sleep") + def test_retries_until_success(self, mock_sleep: MagicMock, mock_conn: MagicMock) -> None: + # Fail twice, then succeed + mock_conn.side_effect = [ + OSError("refused"), + OSError("refused"), + MagicMock(), + ] + mock_conn.return_value.__enter__ = MagicMock() + mock_conn.return_value.__exit__ = MagicMock(return_value=False) + wait_for_ssh("10.0.0.1", max_attempts=5) + assert mock_conn.call_count == 3 + + @patch("devx.utils.ssh.socket.create_connection") + @patch("devx.utils.ssh.time.sleep") + def test_timeout_after_max_attempts(self, mock_sleep: MagicMock, mock_conn: MagicMock) -> None: + mock_conn.side_effect = OSError("refused") + with pytest.raises(RuntimeError, match="SSH not available"): + wait_for_ssh("10.0.0.1", max_attempts=3) + assert mock_conn.call_count == 3 diff --git a/tests/unit/test_utils_step_tracker.py b/tests/unit/test_utils_step_tracker.py new file mode 100644 index 0000000..f5cf53e --- /dev/null +++ b/tests/unit/test_utils_step_tracker.py @@ -0,0 +1,124 @@ +"""Unit tests for devx.utils.step_tracker.""" + +from __future__ import annotations + +import click +import pytest +from click.testing import CliRunner + +from devx.utils.step_tracker import Step, StepTracker, track_steps + + +class TestStep: + def test_initial_status_is_pending(self) -> None: + step = Step("install") + assert step.status == "pending" + assert step.name == "install" + + +class TestStepTracker: + def test_begin_adds_step_as_in_progress(self) -> None: + tracker = StepTracker() + tracker.begin("install deps") + assert len(tracker.steps) == 1 + assert tracker.steps[0].status == "in_progress" + + def test_done_marks_last_in_progress_as_completed(self) -> None: + tracker = StepTracker() + tracker.begin("step1") + tracker.done() + assert tracker.steps[0].status == "completed" + + def test_done_no_op_if_no_in_progress(self) -> None: + tracker = StepTracker() + tracker.begin("step1") + tracker.done() + tracker.done() # should not raise, no-op + assert tracker.steps[0].status == "completed" + + def test_done_no_op_if_empty(self) -> None: + tracker = StepTracker() + tracker.done() # should not raise + + def test_multiple_steps(self) -> None: + tracker = StepTracker() + tracker.begin("step1") + tracker.done() + tracker.begin("step2") + tracker.done() + assert len(tracker.steps) == 2 + assert tracker.steps[0].status == "completed" + assert tracker.steps[1].status == "completed" + + +class TestTrackSteps: + def test_successful_operation(self) -> None: + runner = CliRunner() + with runner.isolation(): + with track_steps() as tracker: + tracker.begin("step1") + tracker.done() + tracker.begin("step2") + tracker.done() + assert len(tracker.steps) == 2 + assert all(s.status == "completed" for s in tracker.steps) + + def test_exception_marks_in_progress_as_failed(self) -> None: + runner = CliRunner() + with runner.isolation(): + with pytest.raises(ValueError, match="boom"): + with track_steps() as tracker: + tracker.begin("step1") + tracker.done() + tracker.begin("step2") + raise ValueError("boom") + assert tracker.steps[0].status == "completed" + assert tracker.steps[1].status == "failed" + + def test_pending_step_stays_pending_on_exception(self) -> None: + runner = CliRunner() + with runner.isolation(): + with pytest.raises(ValueError): + with track_steps() as tracker: + tracker.begin("step1") + tracker.done() + tracker.begin("step2") + tracker.done() + tracker.begin("step3") # in_progress + # step4 is pending (not started) + raise ValueError("oops") + assert tracker.steps[2].status == "failed" + + def test_empty_operation(self) -> None: + runner = CliRunner() + with runner.isolation(): + with track_steps() as tracker: + pass + assert tracker.steps == [] + + def test_report_printed_on_success(self) -> None: + runner = CliRunner() + result = runner.invoke(_cmd_success, [], color=False) + assert result.exit_code == 0 + assert "Operation Report" in result.output + assert "step1" in result.output + + def test_report_printed_on_failure(self) -> None: + runner = CliRunner() + result = runner.invoke(_cmd_failure, [], color=False) + assert result.exit_code != 0 + assert "Operation Report" in result.output + + +@click.command() +def _cmd_success() -> None: + with track_steps() as tracker: + tracker.begin("step1") + tracker.done() + + +@click.command() +def _cmd_failure() -> None: + with track_steps() as tracker: + tracker.begin("step1") + raise ValueError("oops") diff --git a/tests/unit/test_utils_vault.py b/tests/unit/test_utils_vault.py new file mode 100644 index 0000000..0a08370 --- /dev/null +++ b/tests/unit/test_utils_vault.py @@ -0,0 +1,134 @@ +"""Unit tests for devx.utils.vault.""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import MagicMock, patch + +from devx.utils.vault import ( + decrypt_file, + encrypt_file, + is_encrypted, + load_vault_yaml, + save_vault_yaml, +) + + +class TestIsEncrypted: + def test_encrypted_file(self, tmp_path: Path) -> None: + f = tmp_path / "secret.yml" + f.write_text("$ANSIBLE_VAULT;1.1;AES256\n9382928...\n") + assert is_encrypted(f) is True + + def test_plain_file(self, tmp_path: Path) -> None: + f = tmp_path / "plain.yml" + f.write_text("key: value\n") + assert is_encrypted(f) is False + + +class TestLoadVaultYaml: + def test_plain_yaml_no_vault_pass(self, tmp_path: Path) -> None: + f = tmp_path / "data.yml" + f.write_text("key: value\nlist:\n - a\n - b\n") + data = load_vault_yaml(f) + assert data == {"key": "value", "list": ["a", "b"]} + + def test_empty_file(self, tmp_path: Path) -> None: + f = tmp_path / "empty.yml" + f.write_text("") + data = load_vault_yaml(f) + assert data == {} + + def test_vault_pass_not_exists(self, tmp_path: Path) -> None: + f = tmp_path / "data.yml" + f.write_text("key: value\n") + data = load_vault_yaml(f, vault_pass=tmp_path / "nonexistent") + assert data == {"key": "value"} + + @patch("devx.utils.vault.subprocess.run") + def test_encrypted_file_success(self, mock_run: MagicMock, tmp_path: Path) -> None: + f = tmp_path / "secret.yml" + f.write_text("$ANSIBLE_VAULT\n...") + vp = tmp_path / "vault-password" + vp.write_text("secret") + + mock_run.return_value = MagicMock(returncode=0, stdout="key: decrypted\n", stderr="") + data = load_vault_yaml(f, vault_pass=vp) + assert data == {"key": "decrypted"} + + @patch("devx.utils.vault.subprocess.run") + def test_not_vault_encrypted_fallback(self, mock_run: MagicMock, tmp_path: Path) -> None: + f = tmp_path / "plain.yml" + f.write_text("key: value\n") + vp = tmp_path / "vault-password" + vp.write_text("secret") + + mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="is not vault encrypted") + data = load_vault_yaml(f, vault_pass=vp) + assert data == {"key": "value"} + + +class TestSaveVaultYaml: + def test_save_plain(self, tmp_path: Path) -> None: + f = tmp_path / "output.yml" + save_vault_yaml(f, {"key": "value"}) + content = f.read_text() + assert "key: value" in content + + def test_save_with_vault_pass_not_exists(self, tmp_path: Path) -> None: + f = tmp_path / "output.yml" + vp = tmp_path / "nonexistent" + save_vault_yaml(f, {"key": "value"}, vault_pass=vp) + # Should save as plain YAML + content = f.read_text() + assert "key: value" in content + assert "$ANSIBLE_VAULT" not in content + + @patch("devx.utils.vault.subprocess.run") + def test_save_and_encrypt(self, mock_run: MagicMock, tmp_path: Path) -> None: + f = tmp_path / "output.yml" + vp = tmp_path / "vault-password" + vp.write_text("secret") + + save_vault_yaml(f, {"key": "value"}, vault_pass=vp) + # File should be written + assert f.exists() + # ansible-vault encrypt should be called + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + assert "ansible-vault" in cmd + assert "encrypt" in cmd + + +class TestEncryptFile: + @patch("devx.utils.vault.subprocess.run") + def test_calls_ansible_vault(self, mock_run: MagicMock, tmp_path: Path) -> None: + f = tmp_path / "file.yml" + f.write_text("key: value") + vp = tmp_path / "vault-password" + vp.write_text("secret") + + encrypt_file(f, vp) + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + assert "ansible-vault" in cmd + assert "encrypt" in cmd + assert str(f) in cmd + assert str(vp) in cmd + + +class TestDecryptFile: + @patch("devx.utils.vault.subprocess.run") + def test_calls_ansible_vault(self, mock_run: MagicMock, tmp_path: Path) -> None: + f = tmp_path / "file.yml" + f.write_text("$ANSIBLE_VAULT\n...") + vp = tmp_path / "vault-password" + vp.write_text("secret") + + decrypt_file(f, vp) + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + assert "ansible-vault" in cmd + assert "decrypt" in cmd + assert str(f) in cmd + assert str(vp) in cmd diff --git a/tests/unit/test_validate_deploy_ref.py b/tests/unit/test_validate_deploy_ref.py new file mode 100644 index 0000000..1eb861c --- /dev/null +++ b/tests/unit/test_validate_deploy_ref.py @@ -0,0 +1,70 @@ +"""Unit tests for devx.ci.validate_deploy_ref.""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import MagicMock, patch + +from click.testing import CliRunner + +from devx.ci.validate_deploy_ref import main + + +class TestValidateDeployRef: + def test_valid_tag_prints_ref(self, tmp_path: Path) -> None: + runner = CliRunner() + with patch("devx.ci.validate_deploy_ref.subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0, stdout="abcdef1234567890\n", stderr="") + result = runner.invoke(main, ["--tag", "v1.0.0"]) + assert result.exit_code == 0 + assert "v1.0.0" in result.output + + def test_invalid_tag_exits_nonzero(self) -> None: + runner = CliRunner() + with patch("devx.ci.validate_deploy_ref.subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="error") + result = runner.invoke(main, ["--tag", "nonexistent"]) + assert result.exit_code == 1 + assert "does not exist" in result.output + + def test_no_tag_without_allow_empty_exits_nonzero(self) -> None: + runner = CliRunner() + result = runner.invoke(main, []) + assert result.exit_code == 1 + assert "No tag specified" in result.output + + def test_allow_empty_prints_pr_mode(self) -> None: + runner = CliRunner() + result = runner.invoke(main, ["--allow-empty"]) + assert result.exit_code == 0 + assert "PR mode" in result.output + + def test_github_output_writes_ref(self, tmp_path: Path) -> None: + runner = CliRunner() + gh_output = tmp_path / "github_output" + gh_output.write_text("") + with patch("devx.ci.validate_deploy_ref.subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0, stdout="abcdef12\n", stderr="") + with runner.isolation(env={"GITHUB_OUTPUT": str(gh_output)}): + result = runner.invoke(main, ["--tag", "v1.0.0", "--github-output"]) + assert result.exit_code == 0 + content = gh_output.read_text() + assert "deploy-ref=v1.0.0" in content + + def test_github_output_without_env_var_exits_nonzero(self) -> None: + runner = CliRunner() + with patch("devx.ci.validate_deploy_ref.subprocess.run") as mock_run: + mock_run.return_value = MagicMock(returncode=0, stdout="abcdef12\n", stderr="") + with runner.isolation(env={"GITHUB_OUTPUT": ""}): + result = runner.invoke(main, ["--tag", "v1.0.0", "--github-output"]) + assert result.exit_code == 1 + assert "GITHUB_OUTPUT" in result.output + + def test_allow_empty_with_github_output(self, tmp_path: Path) -> None: + runner = CliRunner() + gh_output = tmp_path / "github_output" + gh_output.write_text("") + with runner.isolation(env={"GITHUB_OUTPUT": str(gh_output)}): + result = runner.invoke(main, ["--allow-empty", "--github-output"]) + assert result.exit_code == 0 + assert "deploy-ref=" in gh_output.read_text()