Public Access
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3406639f13 | ||
|
|
9f02ccb40d | ||
|
|
5206158603 | ||
|
|
489cc8343a | ||
|
|
20ea80135c | ||
|
|
53b1d300aa | ||
|
|
5b9e92f324 | ||
|
|
2c0118111d | ||
|
|
333641f862 | ||
|
|
f21b01dce2 | ||
|
|
ff80745eea | ||
|
|
319807f41c |
@@ -5,7 +5,9 @@ name: Build Images
|
||||
# devx and all dependencies into the image.
|
||||
#
|
||||
# Triggers:
|
||||
# - On push to master (after post-merge release completes)
|
||||
# - After post-merge workflow completes successfully (workflow_run)
|
||||
# This ensures images are only rebuilt AFTER the release is published
|
||||
# to PyPI, so the image always has the latest released version.
|
||||
# - Manually via workflow_dispatch
|
||||
#
|
||||
# The workflow builds 3 tier images in sequence:
|
||||
@@ -15,12 +17,10 @@ name: Build Images
|
||||
# After pushing, a cleanup job removes old versions (keeps last 2 + latest).
|
||||
|
||||
on:
|
||||
push:
|
||||
workflow_run:
|
||||
workflows: ["Post-merge"]
|
||||
types: [completed]
|
||||
branches: [master]
|
||||
paths:
|
||||
- docker/**
|
||||
- pyproject.toml
|
||||
- src/devx/**
|
||||
workflow_dispatch:
|
||||
|
||||
concurrency:
|
||||
@@ -49,7 +49,11 @@ jobs:
|
||||
|
||||
build-and-push:
|
||||
needs: [detect-type]
|
||||
if: needs.detect-type.outputs.is-release == 'false'
|
||||
if: >-
|
||||
needs.detect-type.outputs.is-release == 'false' && (
|
||||
github.event_name == 'workflow_dispatch' ||
|
||||
(github.event_name == 'workflow_run' && github.event.workflow_run.conclusion == 'success')
|
||||
)
|
||||
runs-on: docker
|
||||
timeout-minutes: 30
|
||||
steps:
|
||||
|
||||
@@ -49,6 +49,44 @@ repos:
|
||||
pass_filenames: false
|
||||
stages: [pre-commit]
|
||||
|
||||
- id: checkmake
|
||||
name: checkmake Makefile linter
|
||||
entry: make checkmake
|
||||
language: system
|
||||
files: (Makefile|\.mak)$
|
||||
pass_filenames: false
|
||||
stages: [pre-commit]
|
||||
|
||||
- id: check-test-speed
|
||||
name: unit test speed check
|
||||
entry: .venv/bin/python -m devx.tools.check_test_speed --max-seconds 6 --max-single-seconds 0.5
|
||||
language: system
|
||||
types: [python]
|
||||
pass_filenames: false
|
||||
stages: [pre-commit]
|
||||
|
||||
- id: check-translations
|
||||
name: translation completeness check
|
||||
entry: env PYTHONPATH=src .venv/bin/python -m devx.ci.check_translations
|
||||
language: system
|
||||
files: ^src/devx/translations\.json$
|
||||
pass_filenames: false
|
||||
stages: [pre-commit]
|
||||
|
||||
- id: doc-coverage
|
||||
name: documentation coverage check
|
||||
entry: env PYTHONPATH=src .venv/bin/python -m devx.ci.doc_coverage --fail-on-missing
|
||||
language: system
|
||||
pass_filenames: false
|
||||
stages: [pre-commit]
|
||||
|
||||
- id: lint-docs
|
||||
name: documentation lint check
|
||||
entry: env PYTHONPATH=src .venv/bin/python -m devx.ci.lint_docs --root .
|
||||
language: system
|
||||
pass_filenames: false
|
||||
stages: [pre-commit]
|
||||
|
||||
- id: pytest-cov
|
||||
name: pytest with 100% coverage
|
||||
entry: make pytest-cov
|
||||
|
||||
@@ -2,6 +2,30 @@
|
||||
|
||||
All notable changes to this project will be documented in this file.
|
||||
|
||||
## [0.33.2] - 2026-07-05
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
- Abort sync_wiki when list_wiki_pages fails
|
||||
|
||||
## [0.33.1] - 2026-07-05
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
- Build images after post-merge publish, not on push
|
||||
|
||||
## [0.33.0] - 2026-07-05
|
||||
|
||||
### Features
|
||||
|
||||
- Add check_api_identity_checks, setup_ssh_key, and api utils
|
||||
|
||||
## [0.32.1] - 2026-07-01
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
- Add missing i18n translations for new tools
|
||||
|
||||
## [0.32.0] - 2026-07-01
|
||||
|
||||
### Features
|
||||
|
||||
@@ -115,10 +115,7 @@ lint-all: lint workflow-lint lint-dockerfiles
|
||||
# devx's own CI images may have an older devx.mak. Consumer repos can safely alias.
|
||||
lint-dockerfiles:
|
||||
@echo "[lint-dockerfiles] Linting Dockerfiles with hadolint..."
|
||||
@if ! command -v hadolint >/dev/null 2>&1; then \
|
||||
echo "[lint-dockerfiles] ERROR: hadolint not found. Install from https://github.com/hadolint/hadolint/releases" >&2; \
|
||||
exit 1; \
|
||||
fi
|
||||
@command -v hadolint >/dev/null 2>&1 || { echo "hadolint not found" >&2; exit 1; }
|
||||
@find docker -name 'Dockerfile*' -exec hadolint {} +
|
||||
@echo "[lint-dockerfiles] All Dockerfiles passed."
|
||||
|
||||
|
||||
@@ -16,12 +16,12 @@ quality badges.
|
||||
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/src/branch/master/LICENSE)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
|
||||
[](https://www.python.org/downloads/)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
|
||||
[](https://www.python.org/downloads/)
|
||||
|
||||
## Why devx?
|
||||
|
||||
|
||||
+6
-6
@@ -12,12 +12,12 @@ project to be reusable across all oblachno-oss repositories.
|
||||
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/src/branch/master/LICENSE)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
|
||||
[](https://www.python.org/downloads/)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/wiki)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/actions)
|
||||
[](https://git.oblachno.oblachno.fyi/oblachno-oss/devx/releases)
|
||||
[](https://www.python.org/downloads/)
|
||||
|
||||
## Overview
|
||||
|
||||
|
||||
@@ -1,3 +1,3 @@
|
||||
"""devx — reusable development and CI/CD tools for oblachno-oss projects."""
|
||||
|
||||
__version__ = "0.32.0"
|
||||
__version__ = "0.33.2"
|
||||
|
||||
@@ -292,8 +292,10 @@ def main(dry_run: bool, repo: str | None, verify: bool, strict: bool) -> None:
|
||||
|
||||
try:
|
||||
existing_pages = list_wiki_pages(client)
|
||||
except APIError:
|
||||
existing_pages = {}
|
||||
except APIError as e:
|
||||
raise click.ClickException(
|
||||
_("Failed to list existing wiki pages: {error}. Aborting to avoid creating duplicate pages.", error=e)
|
||||
) from e
|
||||
if existing_pages:
|
||||
click.echo(_("Found {count} existing wiki pages.", count=len(existing_pages)))
|
||||
|
||||
|
||||
@@ -110,6 +110,7 @@ devx-ensure-venv:
|
||||
.PHONY: devx-lint-ruff devx-lint-format devx-typecheck devx-lint-bandit devx-lint-deps devx-lint
|
||||
.PHONY: devx-clean devx-pre-push
|
||||
.PHONY: devx-check-mutable-globals devx-check-dep-docs devx-check-test-coverage devx-check-docs devx-check-test-speed
|
||||
.PHONY: devx-check-api-identity-checks devx-setup-ssh-key
|
||||
.PHONY: devx-test-unit devx-pytest-cov
|
||||
.PHONY: devx-setup-image devx-lint-dockerfiles
|
||||
|
||||
@@ -327,6 +328,14 @@ devx-check-docs:
|
||||
devx-check-test-speed:
|
||||
@$(DEVX_PYTHON) -m devx.tools.check_test_speed
|
||||
|
||||
# Scan integration tests for unsafe is True/is False identity checks
|
||||
devx-check-api-identity-checks:
|
||||
@$(DEVX_PYTHON) -m devx.tools.check_api_identity_checks
|
||||
|
||||
# Set up SSH private key from SSH_PRIVATE_KEY env var
|
||||
devx-setup-ssh-key:
|
||||
@$(DEVX_PYTHON) -m devx.tools.setup_ssh_key
|
||||
|
||||
# ── Pre-push validation ───────────────────────────────────────────────────────
|
||||
|
||||
# Run lint + tests before push (projects can override with project-specific targets)
|
||||
|
||||
@@ -0,0 +1,137 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Scan integration tests for unsafe ``is True``/``is False`` identity checks.
|
||||
|
||||
Many APIs (e.g. Mattermost) return boolean values as strings (``"true"``,
|
||||
``"false"``) rather than native JSON booleans. Using ``is True`` or
|
||||
``is not False`` on such responses silently fails because ``"true" is True``
|
||||
evaluates to ``False`` in Python.
|
||||
|
||||
This tool scans ``tests/integration/test_*.py`` files for identity checks
|
||||
on API response values and reports them as errors.
|
||||
|
||||
Configuration (``[tool.devx.check_api_identity_checks]`` in pyproject.toml):
|
||||
|
||||
``scan_dirs`` — list of directories to scan (default: ``["tests/integration"]``)
|
||||
``skip_patterns`` — list of filename patterns to skip (default: ``["test_*_helpers.py"]``)
|
||||
``noqa_marker`` — comment to suppress individual lines (default: ``# noqa``)
|
||||
|
||||
Usage::
|
||||
|
||||
python3 -m devx.tools.check_api_identity_checks
|
||||
python3 -m devx.tools.check_api_identity_checks --scan-dir tests/integration
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
import click
|
||||
|
||||
from devx.config import _load_pyproject_devx
|
||||
from devx.i18n import _
|
||||
|
||||
DEFAULT_SCAN_DIRS = ["tests/integration"]
|
||||
DEFAULT_SKIP_PATTERNS = ["test_*_helpers.py"]
|
||||
DEFAULT_NOQA_MARKER = "# noqa"
|
||||
|
||||
# Matches: x is True, x is False, x is not True, x is not False
|
||||
_IDENTITY_CHECK_RE = re.compile(r"\bis\s+(not\s+)?(True|False)\b")
|
||||
|
||||
|
||||
def _load_config() -> tuple[list[str], list[str], str]:
|
||||
"""Load configuration from pyproject.toml [tool.devx.check_api_identity_checks]."""
|
||||
devx_cfg = _load_pyproject_devx()
|
||||
cfg_raw = devx_cfg.get("check_api_identity_checks", {})
|
||||
if not isinstance(cfg_raw, dict):
|
||||
return DEFAULT_SCAN_DIRS, DEFAULT_SKIP_PATTERNS, DEFAULT_NOQA_MARKER
|
||||
cfg: dict[str, object] = cfg_raw # type: ignore[assignment]
|
||||
|
||||
scan_dirs_raw = cfg.get("scan_dirs", DEFAULT_SCAN_DIRS)
|
||||
scan_dirs: list[str] = [str(d) for d in scan_dirs_raw] if isinstance(scan_dirs_raw, list) else DEFAULT_SCAN_DIRS
|
||||
|
||||
skip_raw = cfg.get("skip_patterns", DEFAULT_SKIP_PATTERNS)
|
||||
skip_patterns: list[str] = [str(p) for p in skip_raw] if isinstance(skip_raw, list) else DEFAULT_SKIP_PATTERNS
|
||||
|
||||
noqa_marker = str(cfg.get("noqa_marker", DEFAULT_NOQA_MARKER))
|
||||
|
||||
return scan_dirs, skip_patterns, noqa_marker
|
||||
|
||||
|
||||
def _matches_skip_pattern(path: Path, skip_patterns: list[str]) -> bool:
|
||||
"""Check if a file path matches any skip pattern."""
|
||||
name = path.name
|
||||
return any(Path(name).match(pattern) for pattern in skip_patterns)
|
||||
|
||||
|
||||
def find_identity_checks(
|
||||
file_path: Path,
|
||||
repo_root: Path,
|
||||
noqa_marker: str,
|
||||
) -> list[str]:
|
||||
"""Return a list of issue strings for unsafe identity checks in *file_path*."""
|
||||
issues: list[str] = []
|
||||
try:
|
||||
source = file_path.read_text(encoding="utf-8")
|
||||
except (OSError, UnicodeDecodeError):
|
||||
return issues
|
||||
|
||||
rel = str(file_path.relative_to(repo_root))
|
||||
for lineno, line in enumerate(source.splitlines(), 1):
|
||||
if noqa_marker in line:
|
||||
continue
|
||||
match = _IDENTITY_CHECK_RE.search(line)
|
||||
if match:
|
||||
issues.append(
|
||||
f"{rel}:{lineno}: unsafe identity check '{match.group()}' "
|
||||
f"— APIs may return string 'true'/'false'. "
|
||||
f"Use string comparison or _is_truthy()/_is_falsy() helpers."
|
||||
)
|
||||
|
||||
return issues
|
||||
|
||||
|
||||
@click.command()
|
||||
@click.option(
|
||||
"--scan-dir",
|
||||
multiple=True,
|
||||
help=_("Directory to scan (default: tests/integration). Can be repeated."),
|
||||
)
|
||||
def cli(scan_dir: tuple[str, ...]) -> None:
|
||||
"""Scan integration tests for unsafe ``is True``/``is False`` identity checks."""
|
||||
repo_root = Path.cwd()
|
||||
config_scan_dirs, skip_patterns, noqa_marker = _load_config()
|
||||
|
||||
scan_dirs = list(scan_dir) if scan_dir else config_scan_dirs
|
||||
|
||||
all_issues: list[str] = []
|
||||
|
||||
for scan_dir_name in scan_dirs:
|
||||
scan_path = repo_root / scan_dir_name
|
||||
if not scan_path.exists():
|
||||
continue
|
||||
for py_file in scan_path.rglob("test_*.py"):
|
||||
if _matches_skip_pattern(py_file, skip_patterns):
|
||||
continue
|
||||
all_issues.extend(find_identity_checks(py_file, repo_root, noqa_marker))
|
||||
|
||||
if all_issues:
|
||||
click.echo(
|
||||
_("Found {count} unsafe identity check(s) in integration tests.", count=len(all_issues)),
|
||||
err=True,
|
||||
)
|
||||
for issue in all_issues:
|
||||
click.echo(f" {issue}", err=True)
|
||||
raise click.ClickException(
|
||||
_(
|
||||
"Use string comparison or _is_truthy()/_is_falsy() helpers instead. "
|
||||
"Add '{marker}' to suppress individual lines.",
|
||||
marker=noqa_marker,
|
||||
)
|
||||
)
|
||||
|
||||
click.echo(_("[check-api-identity-checks] Passed: no unsafe identity checks found"))
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
cli() # pragma: no cover
|
||||
@@ -80,7 +80,7 @@ def _check_python_version(venv_bin: Path) -> None:
|
||||
@click.option("--checkmake-bin", default=None, help="Path to checkmake binary (fallback if not on PATH).")
|
||||
def cli(venv: str, checkmake_bin: str | None) -> None:
|
||||
"""Verify that required development tools are present."""
|
||||
click.echo("[check-deps] Verifying tools...")
|
||||
click.echo(_("[check-deps] Verifying tools..."))
|
||||
|
||||
all_required = True
|
||||
for tool in REQUIRED_TOOLS:
|
||||
@@ -90,15 +90,15 @@ def cli(venv: str, checkmake_bin: str | None) -> None:
|
||||
for tool in OPTIONAL_TOOLS:
|
||||
if not _check_tool(tool, optional=True):
|
||||
if checkmake_bin and Path(checkmake_bin).exists():
|
||||
click.echo(f" {tool}: found at {checkmake_bin}")
|
||||
click.echo(_(" {tool}: found at {path}", tool=tool, path=checkmake_bin))
|
||||
else:
|
||||
click.echo(" Run 'make install-checkmake' to install the Makefile linter.")
|
||||
click.echo(_(" Run 'make install-checkmake' to install the Makefile linter."))
|
||||
|
||||
_check_python_version(Path(venv) / "bin")
|
||||
|
||||
if not all_required:
|
||||
raise click.ClickException("Required tools missing.")
|
||||
click.echo("[check-deps] All core tools present.")
|
||||
raise click.ClickException(_("Required tools missing."))
|
||||
click.echo(_("[check-deps] All core tools present."))
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
|
||||
@@ -32,7 +32,10 @@ _TIMING_RE = re.compile(r"(\d+) passed.* in ([0-9.]+)s")
|
||||
|
||||
# Matches per-test duration lines from --durations=0:
|
||||
# 0.51s call tests/test_foo.py::test_bar
|
||||
_DURATION_LINE_RE = re.compile(r"^(\d+\.?\d*)s\s+(?:setup|call|teardown)\s+(.+)$")
|
||||
# Only "call" duration is counted — "setup" includes import/collection
|
||||
# overhead (coverage init, module imports) which is environment-dependent
|
||||
# and not a test quality signal.
|
||||
_DURATION_LINE_RE = re.compile(r"^(\d+\.?\d*)s\s+call\s+(.+)$")
|
||||
|
||||
|
||||
def run_tests() -> tuple[str, str]:
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Set up SSH private key for CI jobs that need SSH access to remote hosts.
|
||||
|
||||
Writes the ``SSH_PRIVATE_KEY`` env var to ``~/.ssh/id_rsa``, starts
|
||||
``ssh-agent``, and adds the key. Replaces the repeated inline shell
|
||||
pattern in CI workflow files.
|
||||
|
||||
Usage::
|
||||
|
||||
python3 -m devx.tools.setup_ssh_key
|
||||
|
||||
Reads ``SSH_PRIVATE_KEY`` from the environment. Exits 0 on success,
|
||||
1 on missing key.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess # nosec B404
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import click
|
||||
|
||||
from devx.i18n import _
|
||||
|
||||
|
||||
def setup_ssh_key(private_key: str | None = None) -> bool:
|
||||
"""Set up SSH private key and start ssh-agent.
|
||||
|
||||
Args:
|
||||
private_key: The SSH private key content. If None, reads from
|
||||
``SSH_PRIVATE_KEY`` environment variable.
|
||||
|
||||
Returns:
|
||||
True if setup succeeded, False if key is missing.
|
||||
"""
|
||||
key = private_key or os.environ.get("SSH_PRIVATE_KEY", "")
|
||||
if not key:
|
||||
click.echo(_("SSH_PRIVATE_KEY not set — skipping SSH key setup"), err=True)
|
||||
return False
|
||||
|
||||
ssh_dir = Path.home() / ".ssh"
|
||||
ssh_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
key_path = ssh_dir / "id_rsa"
|
||||
key_path.write_text(f"{key}\n", encoding="utf-8")
|
||||
key_path.chmod(0o600)
|
||||
|
||||
# Start ssh-agent and add the key
|
||||
agent_result = subprocess.run( # nosec B603, B607
|
||||
["ssh-agent", "-s"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
if agent_result.returncode != 0:
|
||||
click.echo(_("Failed to start ssh-agent: {error}", error=agent_result.stderr), err=True)
|
||||
return False
|
||||
|
||||
# Parse ssh-agent output to set env vars
|
||||
for raw_line in agent_result.stdout.splitlines():
|
||||
stripped = raw_line.strip()
|
||||
if "=" in stripped and ";" in stripped:
|
||||
var, val = stripped.split("=", 1)
|
||||
val = val.rstrip(";")
|
||||
os.environ[var] = val
|
||||
|
||||
# Add the key (non-fatal if it fails — key may already be loaded)
|
||||
subprocess.run( # nosec B603, B607
|
||||
["ssh-add", str(key_path)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
@click.command()
|
||||
def cli() -> None:
|
||||
"""Set up SSH private key from SSH_PRIVATE_KEY env var."""
|
||||
if setup_ssh_key():
|
||||
click.echo(_("SSH key set up successfully"))
|
||||
sys.exit(0)
|
||||
click.echo(_("SSH key setup skipped (no key provided)"), err=True)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
cli() # pragma: no cover
|
||||
@@ -33,7 +33,7 @@ DEFAULT_VALIDATE_DIRS = [
|
||||
|
||||
def _run_tofu(cmd: list[str], cwd: Path) -> None:
|
||||
"""Run a tofu command in the given directory, raising on failure."""
|
||||
click.echo(f" -> {cwd}")
|
||||
click.echo(_(" -> {dir}", dir=cwd))
|
||||
result = subprocess.run( # nosec B603, B607
|
||||
cmd,
|
||||
cwd=str(cwd),
|
||||
@@ -60,9 +60,9 @@ def tofu_init(env: str, root: str = ".", extra_dirs: list[str] | None = None) ->
|
||||
for dir_pattern in dirs:
|
||||
dir_path = root_path / dir_pattern
|
||||
if dir_path.is_dir():
|
||||
click.echo(f"[tofu-init] Initializing {dir_path}...")
|
||||
click.echo(_("[tofu-init] Initializing {dir}...", dir=dir_path))
|
||||
_run_tofu(["tofu", "init"], dir_path)
|
||||
click.echo("[tofu-init] Done.")
|
||||
click.echo(_("[tofu-init] Done."))
|
||||
|
||||
|
||||
def tofu_validate(
|
||||
@@ -83,7 +83,7 @@ def tofu_validate(
|
||||
root_path = Path(root)
|
||||
target_dirs = dirs or DEFAULT_VALIDATE_DIRS
|
||||
mode = "ci" if ci else "validate"
|
||||
click.echo(f"[tofu-{mode}] Validating OpenTofu configurations...")
|
||||
click.echo(_("[tofu-{mode}] Validating OpenTofu configurations...", mode=mode))
|
||||
for dir_rel in target_dirs:
|
||||
dir_path = root_path / dir_rel
|
||||
if not dir_path.is_dir():
|
||||
@@ -91,7 +91,7 @@ def tofu_validate(
|
||||
if ci:
|
||||
_run_tofu(["tofu", "init", "-backend=false", "-input=false"], dir_path)
|
||||
_run_tofu(["tofu", "validate"], dir_path)
|
||||
click.echo(f"[tofu-{mode}] All configurations valid.")
|
||||
click.echo(_("[tofu-{mode}] All configurations valid.", mode=mode))
|
||||
|
||||
|
||||
@click.group()
|
||||
|
||||
@@ -3198,5 +3198,157 @@
|
||||
"pl": "OSTRZEŻENIE: .venv ma Python {version}, ale wymagane jest >={req}.",
|
||||
"ru": "ПРЕДУПРЕЖДЕНИЕ: в .venv установлен Python {version}, но требуется >={req}.",
|
||||
"zh": "警告: .venv 的 Python 版本为 {version},但要求 >={req}。"
|
||||
},
|
||||
" -> {dir}": {
|
||||
"en": " -> {dir}",
|
||||
"bg": " -> {dir}",
|
||||
"de": " -> {dir}",
|
||||
"pl": " -> {dir}",
|
||||
"ru": " -> {dir}",
|
||||
"zh": " -> {dir}"
|
||||
},
|
||||
"[tofu-init] Initializing {dir}...": {
|
||||
"en": "[tofu-init] Initializing {dir}...",
|
||||
"bg": "[tofu-init] Инициализиране на {dir}...",
|
||||
"de": "[tofu-init] Initialisiere {dir}...",
|
||||
"pl": "[tofu-init] Inicjalizacja {dir}...",
|
||||
"ru": "[tofu-init] Инициализация {dir}...",
|
||||
"zh": "[tofu-init] 正在初始化 {dir}..."
|
||||
},
|
||||
"[tofu-init] Done.": {
|
||||
"en": "[tofu-init] Done.",
|
||||
"bg": "[tofu-init] Готово.",
|
||||
"de": "[tofu-init] Fertig.",
|
||||
"pl": "[tofu-init] Gotowe.",
|
||||
"ru": "[tofu-init] Готово.",
|
||||
"zh": "[tofu-init] 完成。"
|
||||
},
|
||||
"[tofu-{mode}] Validating OpenTofu configurations...": {
|
||||
"en": "[tofu-{mode}] Validating OpenTofu configurations...",
|
||||
"bg": "[tofu-{mode}] Проверка на OpenTofu конфигурациите...",
|
||||
"de": "[tofu-{mode}] Validiere OpenTofu-Konfigurationen...",
|
||||
"pl": "[tofu-{mode}] Sprawdzanie konfiguracji OpenTofu...",
|
||||
"ru": "[tofu-{mode}] Проверка конфигураций OpenTofu...",
|
||||
"zh": "[tofu-{mode}] 正在验证 OpenTofu 配置..."
|
||||
},
|
||||
"[tofu-{mode}] All configurations valid.": {
|
||||
"en": "[tofu-{mode}] All configurations valid.",
|
||||
"bg": "[tofu-{mode}] Всички конфигурации са валидни.",
|
||||
"de": "[tofu-{mode}] Alle Konfigurationen gültig.",
|
||||
"pl": "[tofu-{mode}] Wszystkie konfiguracje są poprawne.",
|
||||
"ru": "[tofu-{mode}] Все конфигурации валидны.",
|
||||
"zh": "[tofu-{mode}] 所有配置有效。"
|
||||
},
|
||||
"[check-deps] Verifying tools...": {
|
||||
"en": "[check-deps] Verifying tools...",
|
||||
"bg": "[check-deps] Проверка на инструментите...",
|
||||
"de": "[check-deps] Werkzeuge werden überprüft...",
|
||||
"pl": "[check-deps] Sprawdzanie narzędzi...",
|
||||
"ru": "[check-deps] Проверка инструментов...",
|
||||
"zh": "[check-deps] 正在验证工具..."
|
||||
},
|
||||
" {tool}: found at {path}": {
|
||||
"en": " {tool}: found at {path}",
|
||||
"bg": " {tool}: намерен на {path}",
|
||||
"de": " {tool}: gefunden unter {path}",
|
||||
"pl": " {tool}: znaleziono w {path}",
|
||||
"ru": " {tool}: найден в {path}",
|
||||
"zh": " {tool}: 在 {path} 找到"
|
||||
},
|
||||
" Run 'make install-checkmake' to install the Makefile linter.": {
|
||||
"en": " Run 'make install-checkmake' to install the Makefile linter.",
|
||||
"bg": " Изпълнете 'make install-checkmake' за инсталиране на Makefile линтера.",
|
||||
"de": " Führen Sie 'make install-checkmake' aus, um den Makefile-Linter zu installieren.",
|
||||
"pl": " Uruchom 'make install-checkmake', aby zainstalować linter Makefile.",
|
||||
"ru": " Выполните 'make install-checkmake' для установки линтера Makefile.",
|
||||
"zh": " 运行 'make install-checkmake' 来安装 Makefile 检查器。"
|
||||
},
|
||||
"Required tools missing.": {
|
||||
"en": "Required tools missing.",
|
||||
"bg": "Липсват задължителни инструменти.",
|
||||
"de": "Erforderliche Werkzeuge fehlen.",
|
||||
"pl": "Brak wymaganych narzędzi.",
|
||||
"ru": "Отсутствуют обязательные инструменты.",
|
||||
"zh": "缺少必需的工具。"
|
||||
},
|
||||
"[check-deps] All core tools present.": {
|
||||
"en": "[check-deps] All core tools present.",
|
||||
"bg": "[check-deps] Всички основни инструменти са налични.",
|
||||
"de": "[check-deps] Alle Kernwerkzeuge vorhanden.",
|
||||
"pl": "[check-deps] Wszystkie podstawowe narzędzia są dostępne.",
|
||||
"ru": "[check-deps] Все основные инструменты доступны.",
|
||||
"zh": "[check-deps] 所有核心工具均已就绪。"
|
||||
},
|
||||
"SSH_PRIVATE_KEY not set — skipping SSH key setup": {
|
||||
"en": "SSH_PRIVATE_KEY not set — skipping SSH key setup",
|
||||
"bg": "SSH_PRIVATE_KEY не е зададен — пропускане на SSH ключ настройката",
|
||||
"de": "SSH_PRIVATE_KEY nicht gesetzt — SSH-Schlüssel-Setup übersprungen",
|
||||
"pl": "SSH_PRIVATE_KEY nie ustawione — pomijanie konfiguracji klucza SSH",
|
||||
"ru": "SSH_PRIVATE_KEY не задан — пропуск настройки SSH-ключа",
|
||||
"zh": "SSH_PRIVATE_KEY 未设置 — 跳过 SSH 密钥设置"
|
||||
},
|
||||
"Failed to start ssh-agent: {error}": {
|
||||
"en": "Failed to start ssh-agent: {error}",
|
||||
"bg": "Неуспешно стартиране на ssh-agent: {error}",
|
||||
"de": "Starten von ssh-agent fehlgeschlagen: {error}",
|
||||
"pl": "Nie udało się uruchomić ssh-agent: {error}",
|
||||
"ru": "Не удалось запустить ssh-agent: {error}",
|
||||
"zh": "启动 ssh-agent 失败: {error}"
|
||||
},
|
||||
"SSH key set up successfully": {
|
||||
"en": "SSH key set up successfully",
|
||||
"bg": "SSH ключът е настроен успешно",
|
||||
"de": "SSH-Schlüssel erfolgreich eingerichtet",
|
||||
"pl": "Klucz SSH skonfigurowany pomyślnie",
|
||||
"ru": "SSH-ключ успешно настроен",
|
||||
"zh": "SSH 密钥设置成功"
|
||||
},
|
||||
"SSH key setup skipped (no key provided)": {
|
||||
"en": "SSH key setup skipped (no key provided)",
|
||||
"bg": "Настройката на SSH ключ е пропусната (не е предоставен ключ)",
|
||||
"de": "SSH-Schlüssel-Setup übersprungen (kein Schlüssel bereitgestellt)",
|
||||
"pl": "Pominięto konfigurację klucza SSH (brak klucza)",
|
||||
"ru": "Настройка SSH-ключа пропущена (ключ не предоставлен)",
|
||||
"zh": "SSH 密钥设置已跳过(未提供密钥)"
|
||||
},
|
||||
"Found {count} unsafe identity check(s) in integration tests.": {
|
||||
"en": "Found {count} unsafe identity check(s) in integration tests.",
|
||||
"bg": "Намерени са {count} небрежни проверки за идентичност в интеграционните тестове.",
|
||||
"de": "{count} unsichere Identitätsprüfung(en) in Integrationstests gefunden.",
|
||||
"pl": "Znaleziono {count} niebezpiecznych sprawdzeń tożsamości w testach integracyjnych.",
|
||||
"ru": "Найдено {count} небезопасных проверок идентичности в интеграционных тестах.",
|
||||
"zh": "在集成测试中发现 {count} 个不安全的身份检查。"
|
||||
},
|
||||
"Use string comparison or _is_truthy()/_is_falsy() helpers instead. Add '{marker}' to suppress individual lines.": {
|
||||
"en": "Use string comparison or _is_truthy()/_is_falsy() helpers instead. Add '{marker}' to suppress individual lines.",
|
||||
"bg": "Използвайте сравнение на низове или _is_truthy()/_is_falsy() помощници. Добавете '{marker}' за потискане на отделни редове.",
|
||||
"de": "Verwenden Sie String-Vergleich oder _is_truthy()/_is_falsy() Hilfsfunktionen. Fügen Sie '{marker}' hinzu, um einzelne Zeilen zu unterdrücken.",
|
||||
"pl": "Użyj porównania ciągów lub pomocników _is_truthy()/_is_falsy(). Dodaj '{marker}', aby pominąć pojedyncze linie.",
|
||||
"ru": "Используйте строковое сравнение или помощники _is_truthy()/_is_falsy(). Добавьте '{marker}' для подавления отдельных строк.",
|
||||
"zh": "使用字符串比较或 _is_truthy()/_is_falsy() 辅助函数。添加 '{marker}' 以抑制个别行。"
|
||||
},
|
||||
"[check-api-identity-checks] Passed: no unsafe identity checks found": {
|
||||
"en": "[check-api-identity-checks] Passed: no unsafe identity checks found",
|
||||
"bg": "[check-api-identity-checks] Мина: не са намерени небрежни проверки за идентичност",
|
||||
"de": "[check-api-identity-checks] Bestanden: keine unsicheren Identitätsprüfungen gefunden",
|
||||
"pl": "[check-api-identity-checks] Passed: nie znaleziono niebezpiecznych sprawdzeń tożsamości",
|
||||
"ru": "[check-api-identity-checks] Пройдено: небезопасных проверок идентичности не найдено",
|
||||
"zh": "[check-api-identity-checks] 通过:未发现不安全的身份检查"
|
||||
},
|
||||
"Directory to scan (default: tests/integration). Can be repeated.": {
|
||||
"en": "Directory to scan (default: tests/integration). Can be repeated.",
|
||||
"bg": "Директория за сканиране (по подразбиране: tests/integration). Може да се повтаря.",
|
||||
"de": "Zu scannendes Verzeichnis (Standard: tests/integration). Kann wiederholt werden.",
|
||||
"pl": "Katalog do skanowania (domyślnie: tests/integration). Można powtarzać.",
|
||||
"ru": "Директория для сканирования (по умолчанию: tests/integration). Можно повторять.",
|
||||
"zh": "要扫描的目录(默认:tests/integration)。可重复。"
|
||||
},
|
||||
"Failed to list existing wiki pages: {error}. Aborting to avoid creating duplicate pages.": {
|
||||
"bg": "Неуспешно извличане на съществуващи wiki страници: {error}. Прекратяване, за да се избегне създаване на дублирани страници.",
|
||||
"de": "Abrufen bestehender Wiki-Seiten fehlgeschlagen: {error}. Abbruch, um doppelte Seiten zu vermeiden.",
|
||||
"en": "Failed to list existing wiki pages: {error}. Aborting to avoid creating duplicate pages.",
|
||||
"pl": "Nie udało się wylistować istniejących stron wiki: {error}. Przerywanie, aby uniknąć tworzenia zduplikowanych stron.",
|
||||
"ru": "Не удалось получить список существующих wiki-страниц: {error}. Прерывание, чтобы избежать создания дубликатов страниц.",
|
||||
"zh": "列出现有 wiki 页面失败:{error}。正在中止以避免创建重复页面。"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
"""Shared utility functions for devx and consumer projects."""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -0,0 +1,51 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Utilities for handling API response values.
|
||||
|
||||
Many APIs return boolean values as strings (``"true"``, ``"false"``)
|
||||
rather than native JSON booleans. The Mattermost ``/api/v4/config/client``
|
||||
endpoint is a notable example. These helpers handle both string and
|
||||
boolean responses safely.
|
||||
|
||||
Usage::
|
||||
|
||||
from devx.utils.api import is_truthy, is_falsy
|
||||
|
||||
if not is_truthy(config.get("EnableOpenServer")):
|
||||
raise ValueError("EnableOpenServer not enabled")
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
||||
def is_truthy(value: str | bool | None) -> bool:
|
||||
"""Check if an API config value is truthy.
|
||||
|
||||
The API may return strings (``"true"``/``"false"``) or native
|
||||
booleans. This helper handles both.
|
||||
|
||||
Args:
|
||||
value: The value to check (string, bool, or None).
|
||||
|
||||
Returns:
|
||||
True if the value represents a truthy boolean.
|
||||
"""
|
||||
if isinstance(value, bool):
|
||||
return value
|
||||
return str(value).lower() == "true"
|
||||
|
||||
|
||||
def is_falsy(value: str | bool | None) -> bool:
|
||||
"""Check if an API config value is falsy.
|
||||
|
||||
The API may return strings (``"true"``/``"false"``) or native
|
||||
booleans. This helper handles both.
|
||||
|
||||
Args:
|
||||
value: The value to check (string, bool, or None).
|
||||
|
||||
Returns:
|
||||
True if the value represents a falsy boolean.
|
||||
"""
|
||||
if isinstance(value, bool):
|
||||
return not value
|
||||
return str(value).lower() == "false"
|
||||
@@ -0,0 +1,54 @@
|
||||
"""Unit tests for devx.utils.api."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from devx.utils.api import is_falsy, is_truthy
|
||||
|
||||
|
||||
class TestIsTruthy:
|
||||
def test_string_true(self) -> None:
|
||||
assert is_truthy("true") is True
|
||||
|
||||
def test_string_true_uppercase(self) -> None:
|
||||
assert is_truthy("True") is True
|
||||
|
||||
def test_boolean_true(self) -> None:
|
||||
assert is_truthy(True) is True
|
||||
|
||||
def test_string_false(self) -> None:
|
||||
assert is_truthy("false") is False
|
||||
|
||||
def test_boolean_false(self) -> None:
|
||||
assert is_truthy(False) is False
|
||||
|
||||
def test_none(self) -> None:
|
||||
assert is_truthy(None) is False
|
||||
|
||||
def test_empty_string(self) -> None:
|
||||
assert is_truthy("") is False
|
||||
|
||||
def test_random_string(self) -> None:
|
||||
assert is_truthy("random") is False
|
||||
|
||||
|
||||
class TestIsFalsy:
|
||||
def test_string_false(self) -> None:
|
||||
assert is_falsy("false") is True
|
||||
|
||||
def test_string_false_uppercase(self) -> None:
|
||||
assert is_falsy("False") is True
|
||||
|
||||
def test_boolean_false(self) -> None:
|
||||
assert is_falsy(False) is True
|
||||
|
||||
def test_string_true(self) -> None:
|
||||
assert is_falsy("true") is False
|
||||
|
||||
def test_boolean_true(self) -> None:
|
||||
assert is_falsy(True) is False
|
||||
|
||||
def test_none(self) -> None:
|
||||
assert is_falsy(None) is False
|
||||
|
||||
def test_empty_string(self) -> None:
|
||||
assert is_falsy("") is False
|
||||
@@ -0,0 +1,176 @@
|
||||
"""Unit tests for devx.tools.check_api_identity_checks."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from click.testing import CliRunner
|
||||
|
||||
from devx.tools.check_api_identity_checks import (
|
||||
DEFAULT_NOQA_MARKER,
|
||||
DEFAULT_SCAN_DIRS,
|
||||
DEFAULT_SKIP_PATTERNS,
|
||||
_load_config,
|
||||
_matches_skip_pattern,
|
||||
cli,
|
||||
find_identity_checks,
|
||||
)
|
||||
|
||||
|
||||
class TestFindIdentityChecks:
|
||||
def test_detects_is_true(self, tmp_path: Path) -> None:
|
||||
f = tmp_path / "test_foo.py"
|
||||
f.write_text("assert config.get('x') is True\n")
|
||||
issues = find_identity_checks(f, tmp_path, DEFAULT_NOQA_MARKER)
|
||||
assert len(issues) == 1
|
||||
assert "is True" in issues[0]
|
||||
|
||||
def test_detects_is_false(self, tmp_path: Path) -> None:
|
||||
f = tmp_path / "test_foo.py"
|
||||
f.write_text("if config.get('x') is False:\n pass\n")
|
||||
issues = find_identity_checks(f, tmp_path, DEFAULT_NOQA_MARKER)
|
||||
assert len(issues) == 1
|
||||
assert "is False" in issues[0]
|
||||
|
||||
def test_detects_is_not_true(self, tmp_path: Path) -> None:
|
||||
f = tmp_path / "test_foo.py"
|
||||
f.write_text("if config.get('x') is not True:\n fail()\n")
|
||||
issues = find_identity_checks(f, tmp_path, DEFAULT_NOQA_MARKER)
|
||||
assert len(issues) == 1
|
||||
assert "is not True" in issues[0]
|
||||
|
||||
def test_detects_is_not_false(self, tmp_path: Path) -> None:
|
||||
f = tmp_path / "test_foo.py"
|
||||
f.write_text("if config.get('x') is not False:\n fail()\n")
|
||||
issues = find_identity_checks(f, tmp_path, DEFAULT_NOQA_MARKER)
|
||||
assert len(issues) == 1
|
||||
assert "is not False" in issues[0]
|
||||
|
||||
def test_noqa_suppresses(self, tmp_path: Path) -> None:
|
||||
f = tmp_path / "test_foo.py"
|
||||
f.write_text("assert config.get('x') is True # noqa\n")
|
||||
issues = find_identity_checks(f, tmp_path, DEFAULT_NOQA_MARKER)
|
||||
assert len(issues) == 0
|
||||
|
||||
def test_no_false_positives(self, tmp_path: Path) -> None:
|
||||
f = tmp_path / "test_foo.py"
|
||||
f.write_text("assert config.get('x') == 'true'\nassert config.get('y') == True\nx = True\nif x:\n pass\n")
|
||||
issues = find_identity_checks(f, tmp_path, DEFAULT_NOQA_MARKER)
|
||||
assert len(issues) == 0
|
||||
|
||||
def test_multiple_issues(self, tmp_path: Path) -> None:
|
||||
f = tmp_path / "test_foo.py"
|
||||
f.write_text("if config.get('a') is True:\n pass\nif config.get('b') is not False:\n pass\n")
|
||||
issues = find_identity_checks(f, tmp_path, DEFAULT_NOQA_MARKER)
|
||||
assert len(issues) == 2
|
||||
|
||||
def test_file_not_found(self, tmp_path: Path) -> None:
|
||||
f = tmp_path / "nonexistent.py"
|
||||
issues = find_identity_checks(f, tmp_path, DEFAULT_NOQA_MARKER)
|
||||
assert issues == []
|
||||
|
||||
|
||||
class TestMatchesSkipPattern:
|
||||
def test_matches_helpers(self) -> None:
|
||||
assert _matches_skip_pattern(Path("test_mattermost_helpers.py"), DEFAULT_SKIP_PATTERNS)
|
||||
|
||||
def test_does_not_match_regular(self) -> None:
|
||||
assert not _matches_skip_pattern(Path("test_mattermost.py"), DEFAULT_SKIP_PATTERNS)
|
||||
|
||||
def test_empty_patterns(self) -> None:
|
||||
assert not _matches_skip_pattern(Path("test_anything.py"), [])
|
||||
|
||||
|
||||
class TestLoadConfig:
|
||||
def test_defaults(self) -> None:
|
||||
with patch("devx.tools.check_api_identity_checks._load_pyproject_devx") as mock:
|
||||
mock.return_value = {}
|
||||
scan_dirs, skip_patterns, noqa = _load_config()
|
||||
assert scan_dirs == DEFAULT_SCAN_DIRS
|
||||
assert skip_patterns == DEFAULT_SKIP_PATTERNS
|
||||
assert noqa == DEFAULT_NOQA_MARKER
|
||||
|
||||
def test_custom_config(self) -> None:
|
||||
with patch("devx.tools.check_api_identity_checks._load_pyproject_devx") as mock:
|
||||
mock.return_value = {
|
||||
"check_api_identity_checks": {
|
||||
"scan_dirs": ["tests/api"],
|
||||
"skip_patterns": ["test_*_unit.py"],
|
||||
"noqa_marker": "# allow",
|
||||
}
|
||||
}
|
||||
scan_dirs, skip_patterns, noqa = _load_config()
|
||||
assert scan_dirs == ["tests/api"]
|
||||
assert skip_patterns == ["test_*_unit.py"]
|
||||
assert noqa == "# allow"
|
||||
|
||||
def test_invalid_config_returns_defaults(self) -> None:
|
||||
with patch("devx.tools.check_api_identity_checks._load_pyproject_devx") as mock:
|
||||
mock.return_value = {"check_api_identity_checks": "not a dict"}
|
||||
scan_dirs, _, _ = _load_config()
|
||||
assert scan_dirs == DEFAULT_SCAN_DIRS
|
||||
|
||||
|
||||
class TestCli:
|
||||
def test_no_issues(self, tmp_path: Path) -> None:
|
||||
runner = CliRunner()
|
||||
with (
|
||||
patch("devx.tools.check_api_identity_checks._load_config") as mock_cfg,
|
||||
patch("devx.tools.check_api_identity_checks.Path.cwd", return_value=tmp_path),
|
||||
):
|
||||
mock_cfg.return_value = (["tests/integration"], DEFAULT_SKIP_PATTERNS, DEFAULT_NOQA_MARKER)
|
||||
(tmp_path / "tests" / "integration").mkdir(parents=True)
|
||||
(tmp_path / "tests" / "integration" / "test_foo.py").write_text("assert config.get('x') == 'true'\n")
|
||||
result = runner.invoke(cli, [])
|
||||
assert result.exit_code == 0
|
||||
assert "Passed" in result.output
|
||||
|
||||
def test_with_issues(self, tmp_path: Path) -> None:
|
||||
runner = CliRunner()
|
||||
with (
|
||||
patch("devx.tools.check_api_identity_checks._load_config") as mock_cfg,
|
||||
patch("devx.tools.check_api_identity_checks.Path.cwd", return_value=tmp_path),
|
||||
):
|
||||
mock_cfg.return_value = (["tests/integration"], DEFAULT_SKIP_PATTERNS, DEFAULT_NOQA_MARKER)
|
||||
(tmp_path / "tests" / "integration").mkdir(parents=True)
|
||||
(tmp_path / "tests" / "integration" / "test_foo.py").write_text(
|
||||
"if config.get('x') is not True:\n fail()\n"
|
||||
)
|
||||
result = runner.invoke(cli, [])
|
||||
assert result.exit_code != 0
|
||||
assert "is not True" in result.output
|
||||
|
||||
def test_skips_helpers(self, tmp_path: Path) -> None:
|
||||
runner = CliRunner()
|
||||
with (
|
||||
patch("devx.tools.check_api_identity_checks._load_config") as mock_cfg,
|
||||
patch("devx.tools.check_api_identity_checks.Path.cwd", return_value=tmp_path),
|
||||
):
|
||||
mock_cfg.return_value = (["tests/integration"], DEFAULT_SKIP_PATTERNS, DEFAULT_NOQA_MARKER)
|
||||
(tmp_path / "tests" / "integration").mkdir(parents=True)
|
||||
(tmp_path / "tests" / "integration" / "test_foo_helpers.py").write_text("assert x is True\n")
|
||||
result = runner.invoke(cli, [])
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_nonexistent_dir(self, tmp_path: Path) -> None:
|
||||
runner = CliRunner()
|
||||
with (
|
||||
patch("devx.tools.check_api_identity_checks._load_config") as mock_cfg,
|
||||
patch("devx.tools.check_api_identity_checks.Path.cwd", return_value=tmp_path),
|
||||
):
|
||||
mock_cfg.return_value = (["nonexistent"], DEFAULT_SKIP_PATTERNS, DEFAULT_NOQA_MARKER)
|
||||
result = runner.invoke(cli, [])
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_custom_scan_dir(self, tmp_path: Path) -> None:
|
||||
runner = CliRunner()
|
||||
with (
|
||||
patch("devx.tools.check_api_identity_checks._load_config") as mock_cfg,
|
||||
patch("devx.tools.check_api_identity_checks.Path.cwd", return_value=tmp_path),
|
||||
):
|
||||
mock_cfg.return_value = (["other"], DEFAULT_SKIP_PATTERNS, DEFAULT_NOQA_MARKER)
|
||||
(tmp_path / "custom").mkdir()
|
||||
(tmp_path / "custom" / "test_foo.py").write_text("if x is True:\n pass\n")
|
||||
result = runner.invoke(cli, ["--scan-dir", "custom"])
|
||||
assert result.exit_code != 0
|
||||
@@ -69,16 +69,16 @@ class TestParsePerTestDurations:
|
||||
assert len(durations) == 1
|
||||
assert durations[0] == ("tests/test_foo.py::test_bar", 0.01)
|
||||
|
||||
def test_parses_setup_and_teardown(self) -> None:
|
||||
def test_ignores_setup_and_teardown(self) -> None:
|
||||
"""Only 'call' durations are counted — setup includes import overhead."""
|
||||
output = (
|
||||
"0.02s setup tests/test_foo.py::test_bar\n"
|
||||
"0.68s setup tests/test_foo.py::test_bar\n"
|
||||
"0.01s call tests/test_foo.py::test_bar\n"
|
||||
"0.00s teardown tests/test_foo.py::test_bar\n"
|
||||
)
|
||||
durations = parse_per_test_durations(output)
|
||||
assert len(durations) == 3
|
||||
names = [d[0] for d in durations]
|
||||
assert "tests/test_foo.py::test_bar" in names
|
||||
assert len(durations) == 1
|
||||
assert durations[0] == ("tests/test_foo.py::test_bar", 0.01)
|
||||
|
||||
def test_sorted_slowest_first(self) -> None:
|
||||
output = "0.01s call tests/test_a.py::test_slow\n0.50s call tests/test_b.py::test_fast\n"
|
||||
|
||||
@@ -0,0 +1,153 @@
|
||||
"""Unit tests for devx.tools.setup_ssh_key."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from click.testing import CliRunner
|
||||
|
||||
from devx.tools.setup_ssh_key import cli, setup_ssh_key
|
||||
|
||||
|
||||
class TestSetupSshKey:
|
||||
def test_success(self, tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
monkeypatch.setenv("SSH_PRIVATE_KEY", "-----BEGIN KEY-----\nfake\n-----END KEY-----")
|
||||
with (
|
||||
patch("subprocess.run") as mock_run,
|
||||
patch("pathlib.Path.chmod"),
|
||||
):
|
||||
agent_result = MagicMock()
|
||||
agent_result.returncode = 0
|
||||
agent_result.stdout = "SSH_AUTH_SOCK=/tmp/agent.sock;\nSSH_AGENT_PID=12345;\n"
|
||||
agent_result.stderr = ""
|
||||
add_result = MagicMock()
|
||||
add_result.returncode = 0
|
||||
add_result.stdout = ""
|
||||
add_result.stderr = ""
|
||||
mock_run.side_effect = [agent_result, add_result]
|
||||
assert setup_ssh_key() is True
|
||||
assert mock_run.call_count == 2
|
||||
|
||||
def test_missing_key(self, monkeypatch) -> None:
|
||||
monkeypatch.delenv("SSH_PRIVATE_KEY", raising=False)
|
||||
assert setup_ssh_key() is False
|
||||
|
||||
def test_empty_key(self, monkeypatch) -> None:
|
||||
monkeypatch.setenv("SSH_PRIVATE_KEY", "")
|
||||
assert setup_ssh_key() is False
|
||||
|
||||
def test_explicit_key_param(self, tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
monkeypatch.delenv("SSH_PRIVATE_KEY", raising=False)
|
||||
with (
|
||||
patch("subprocess.run") as mock_run,
|
||||
patch("pathlib.Path.chmod"),
|
||||
):
|
||||
agent_result = MagicMock()
|
||||
agent_result.returncode = 0
|
||||
agent_result.stdout = "SSH_AUTH_SOCK=/tmp/agent.sock;\n"
|
||||
agent_result.stderr = ""
|
||||
add_result = MagicMock()
|
||||
add_result.returncode = 0
|
||||
add_result.stdout = ""
|
||||
add_result.stderr = ""
|
||||
mock_run.side_effect = [agent_result, add_result]
|
||||
assert setup_ssh_key("-----BEGIN KEY-----\nfake\n-----END KEY-----") is True
|
||||
|
||||
def test_ssh_agent_failure(self, tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
monkeypatch.setenv("SSH_PRIVATE_KEY", "fake-key")
|
||||
with (
|
||||
patch("subprocess.run") as mock_run,
|
||||
patch("pathlib.Path.chmod"),
|
||||
):
|
||||
agent_result = MagicMock()
|
||||
agent_result.returncode = 1
|
||||
agent_result.stdout = ""
|
||||
agent_result.stderr = "ssh-agent failed"
|
||||
mock_run.return_value = agent_result
|
||||
assert setup_ssh_key() is False
|
||||
|
||||
def test_key_file_written(self, tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
monkeypatch.setenv("SSH_PRIVATE_KEY", "my-secret-key")
|
||||
with (
|
||||
patch("subprocess.run") as mock_run,
|
||||
patch("pathlib.Path.chmod") as mock_chmod,
|
||||
):
|
||||
agent_result = MagicMock()
|
||||
agent_result.returncode = 0
|
||||
agent_result.stdout = "SSH_AUTH_SOCK=/tmp/agent.sock;\n"
|
||||
agent_result.stderr = ""
|
||||
add_result = MagicMock()
|
||||
add_result.returncode = 0
|
||||
add_result.stdout = ""
|
||||
add_result.stderr = ""
|
||||
mock_run.side_effect = [agent_result, add_result]
|
||||
setup_ssh_key()
|
||||
key_file = tmp_path / ".ssh" / "id_rsa"
|
||||
assert key_file.exists()
|
||||
assert "my-secret-key" in key_file.read_text()
|
||||
mock_chmod.assert_called_with(0o600)
|
||||
|
||||
def test_env_vars_set_from_agent(self, tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
monkeypatch.setenv("SSH_PRIVATE_KEY", "fake-key")
|
||||
with (
|
||||
patch("subprocess.run") as mock_run,
|
||||
patch("pathlib.Path.chmod"),
|
||||
):
|
||||
agent_result = MagicMock()
|
||||
agent_result.returncode = 0
|
||||
agent_result.stdout = "SSH_AUTH_SOCK=/tmp/agent.sock;\nSSH_AGENT_PID=999;\n"
|
||||
agent_result.stderr = ""
|
||||
add_result = MagicMock()
|
||||
add_result.returncode = 0
|
||||
add_result.stdout = ""
|
||||
add_result.stderr = ""
|
||||
mock_run.side_effect = [agent_result, add_result]
|
||||
setup_ssh_key()
|
||||
assert os.environ.get("SSH_AUTH_SOCK") == "/tmp/agent.sock"
|
||||
assert os.environ.get("SSH_AGENT_PID") == "999"
|
||||
|
||||
def test_agent_output_without_env_vars(self, tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
monkeypatch.setenv("SSH_PRIVATE_KEY", "fake-key")
|
||||
monkeypatch.delenv("SSH_AUTH_SOCK", raising=False)
|
||||
with (
|
||||
patch("subprocess.run") as mock_run,
|
||||
patch("pathlib.Path.chmod"),
|
||||
):
|
||||
agent_result = MagicMock()
|
||||
agent_result.returncode = 0
|
||||
agent_result.stdout = "Agent started\nsome message without equals\n"
|
||||
agent_result.stderr = ""
|
||||
add_result = MagicMock()
|
||||
add_result.returncode = 0
|
||||
add_result.stdout = ""
|
||||
add_result.stderr = ""
|
||||
mock_run.side_effect = [agent_result, add_result]
|
||||
assert setup_ssh_key() is True
|
||||
assert os.environ.get("SSH_AUTH_SOCK") is None
|
||||
|
||||
|
||||
class TestCli:
|
||||
def test_success(self, tmp_path, monkeypatch) -> None:
|
||||
monkeypatch.setenv("HOME", str(tmp_path))
|
||||
monkeypatch.setenv("SSH_PRIVATE_KEY", "fake-key")
|
||||
runner = CliRunner()
|
||||
with patch("devx.tools.setup_ssh_key.setup_ssh_key") as mock_setup:
|
||||
mock_setup.return_value = True
|
||||
result = runner.invoke(cli, [])
|
||||
assert result.exit_code == 0
|
||||
assert "successfully" in result.output
|
||||
|
||||
def test_no_key(self, monkeypatch) -> None:
|
||||
monkeypatch.delenv("SSH_PRIVATE_KEY", raising=False)
|
||||
runner = CliRunner()
|
||||
with patch("devx.tools.setup_ssh_key.setup_ssh_key") as mock_setup:
|
||||
mock_setup.return_value = False
|
||||
result = runner.invoke(cli, [])
|
||||
assert result.exit_code == 1
|
||||
@@ -537,8 +537,8 @@ class TestMain:
|
||||
|
||||
@patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True)
|
||||
@patch("devx.ci.sync_wiki.GiteaClient")
|
||||
def test_initial_list_api_error_treated_as_empty(self, mock_client_cls: MagicMock) -> None:
|
||||
"""When the initial page list fails, sync proceeds treating wiki as empty."""
|
||||
def test_initial_list_api_error_aborts(self, mock_client_cls: MagicMock) -> None:
|
||||
"""When the initial page list fails, sync aborts to avoid duplicate pages."""
|
||||
mock_client = MagicMock()
|
||||
mock_client_cls.return_value = mock_client
|
||||
with patch("devx.ci.sync_wiki.MAPPING_FILE") as mock_mapping:
|
||||
@@ -549,8 +549,9 @@ class TestMain:
|
||||
with patch("devx.ci.sync_wiki.sync_page", return_value="created"):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["--repo", "owner/repo"])
|
||||
assert result.exit_code == 0
|
||||
assert "Created: Home" in result.output
|
||||
assert result.exit_code != 0
|
||||
assert "Failed to list existing wiki pages" in result.output
|
||||
assert "Aborting" in result.output
|
||||
|
||||
@patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True)
|
||||
@patch("devx.ci.sync_wiki.GiteaClient")
|
||||
|
||||
Reference in New Issue
Block a user