diff --git a/src/devx/make/devx.mak b/src/devx/make/devx.mak index e5c8139..d6177dc 100644 --- a/src/devx/make/devx.mak +++ b/src/devx/make/devx.mak @@ -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) diff --git a/src/devx/tools/check_api_identity_checks.py b/src/devx/tools/check_api_identity_checks.py new file mode 100644 index 0000000..e3a8a56 --- /dev/null +++ b/src/devx/tools/check_api_identity_checks.py @@ -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 diff --git a/src/devx/tools/setup_ssh_key.py b/src/devx/tools/setup_ssh_key.py new file mode 100644 index 0000000..6e559c8 --- /dev/null +++ b/src/devx/tools/setup_ssh_key.py @@ -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 diff --git a/src/devx/translations.json b/src/devx/translations.json index 263e2bf..b037e03 100644 --- a/src/devx/translations.json +++ b/src/devx/translations.json @@ -3278,5 +3278,69 @@ "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）。可重复。" } } diff --git a/src/devx/utils/__init__.py b/src/devx/utils/__init__.py new file mode 100644 index 0000000..314d713 --- /dev/null +++ b/src/devx/utils/__init__.py @@ -0,0 +1,3 @@ +"""Shared utility functions for devx and consumer projects.""" + +from __future__ import annotations diff --git a/src/devx/utils/api.py b/src/devx/utils/api.py new file mode 100644 index 0000000..ab87dd7 --- /dev/null +++ b/src/devx/utils/api.py @@ -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" diff --git a/tests/unit/test_api_utils.py b/tests/unit/test_api_utils.py new file mode 100644 index 0000000..d1d275c --- /dev/null +++ b/tests/unit/test_api_utils.py @@ -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 diff --git a/tests/unit/test_check_api_identity_checks.py b/tests/unit/test_check_api_identity_checks.py new file mode 100644 index 0000000..43ec3a7 --- /dev/null +++ b/tests/unit/test_check_api_identity_checks.py @@ -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 diff --git a/tests/unit/test_setup_ssh_key.py b/tests/unit/test_setup_ssh_key.py new file mode 100644 index 0000000..91981e7 --- /dev/null +++ b/tests/unit/test_setup_ssh_key.py @@ -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