Public Access
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 devx-ci-bot
|
bdfe2c561b | ||
 emil
|
02b27dd343 | ||
|
gitea-actions-bot
|
e5488fcfbd | ||
|
devx-ci-bot
|
1a60739b5a | ||
|
emil
|
50dcb67083 | ||
|
gitea-actions-bot
|
7b624b0525 |
@@ -2,6 +2,18 @@
|
||||
|
||||
All notable changes to this project will be documented in this file.
|
||||
|
||||
## [0.42.0] - 2026-07-13
|
||||
|
||||
### Features
|
||||
|
||||
- Add I/O function isolation check and skip integration tests
|
||||
|
||||
## [0.41.2] - 2026-07-13
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
- Auto-discover molecule root instead of hardcoding gitea-runner
|
||||
|
||||
## [0.41.1] - 2026-07-13
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
@@ -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?
|
||||
|
||||
@@ -87,7 +87,7 @@ extra index and list devx in your dependencies:
|
||||
```toml
|
||||
[project]
|
||||
dependencies = [
|
||||
"devx>=0.41.1",
|
||||
"devx>=0.42.0",
|
||||
]
|
||||
|
||||
[tool.pip]
|
||||
@@ -101,8 +101,8 @@ pip install -e .
|
||||
```
|
||||
|
||||
> **Note:** If your project requires a specific devx version, pin it in
|
||||
> `dependencies` (for example, `"devx==0.41.1"`) or use a version constraint
|
||||
> (for example, `"devx>=0.41.1,<0.42"`).
|
||||
> `dependencies` (for example, `"devx==0.42.0"`) or use a version constraint
|
||||
> (for example, `"devx>=0.42.0,<0.43"`).
|
||||
|
||||
### Optional extras
|
||||
|
||||
|
||||
+8
-8
@@ -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
|
||||
|
||||
@@ -74,14 +74,14 @@ Add devx to your `pyproject.toml` dependencies and configure the registry:
|
||||
```toml
|
||||
[project]
|
||||
dependencies = [
|
||||
"devx>=0.41.1",
|
||||
"devx>=0.42.0",
|
||||
]
|
||||
|
||||
[tool.pip]
|
||||
extra-index-url = "https://git.oblachno.oblachno.fyi/api/packages/oblachno-oss/pypi/simple"
|
||||
```
|
||||
|
||||
Pin a specific version if needed: `"devx==0.41.1"` or `"devx>=0.41.1,<0.42"`.
|
||||
Pin a specific version if needed: `"devx==0.42.0"` or `"devx>=0.42.0,<0.43"`.
|
||||
|
||||
### Optional extras
|
||||
|
||||
|
||||
@@ -48,12 +48,12 @@ Add devx to your `pyproject.toml`:
|
||||
```toml
|
||||
[project]
|
||||
dependencies = [
|
||||
"devx>=0.41.1",
|
||||
"devx>=0.42.0",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
dev = [
|
||||
"devx>=0.41.1",
|
||||
"devx>=0.42.0",
|
||||
]
|
||||
```
|
||||
|
||||
|
||||
@@ -1,3 +1,3 @@
|
||||
"""devx — reusable development and CI/CD tools for oblachno-oss projects."""
|
||||
|
||||
__version__ = "0.41.1"
|
||||
__version__ = "0.42.0"
|
||||
|
||||
@@ -30,10 +30,24 @@ from devx.i18n import _
|
||||
from devx.molecule.platforms import PLATFORMS, load_platforms
|
||||
|
||||
DEFAULT_MAX_RUNNERS = 3
|
||||
MOLECULE_ROOT = Path("ansible/roles/gitea-runner/molecule")
|
||||
DEFAULT_ROLES_ROOT = Path("ansible/roles")
|
||||
|
||||
|
||||
def _default_molecule_root() -> Path:
|
||||
"""Auto-discover the single molecule directory under ansible/roles/.
|
||||
|
||||
If exactly one role has a molecule/ subdirectory, return it.
|
||||
Otherwise, fall back to the first role with a molecule/ directory.
|
||||
"""
|
||||
roles_root = DEFAULT_ROLES_ROOT
|
||||
if not roles_root.is_dir():
|
||||
return roles_root / "gitea_runner" / "molecule" # sensible default for error message
|
||||
mol_dirs = sorted(d / "molecule" for d in roles_root.iterdir() if (d / "molecule").is_dir())
|
||||
if mol_dirs:
|
||||
return mol_dirs[0]
|
||||
return roles_root / "molecule" # will produce a clear "not found" error
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TestPair:
|
||||
"""A (scenario, platform) combination to test."""
|
||||
@@ -83,7 +97,7 @@ class MultiRoleTestPair:
|
||||
def discover_scenarios(root: Path | None = None) -> list[str]:
|
||||
"""Return sorted list of molecule scenario directory names."""
|
||||
if root is None:
|
||||
root = MOLECULE_ROOT
|
||||
root = _default_molecule_root()
|
||||
if not root.is_dir():
|
||||
raise click.ClickException(_("Molecule directory not found: {path}", path=str(root)))
|
||||
scenarios = [d.name for d in root.iterdir() if d.is_dir() and not d.name.startswith("_") and d.name != "common"]
|
||||
@@ -318,7 +332,7 @@ def _write_github_env(key: str, value: str) -> None:
|
||||
"--molecule-root",
|
||||
type=click.Path(exists=True, file_okay=False, path_type=Path),
|
||||
default=None,
|
||||
help="Custom molecule directory (single-role mode). Default: ansible/roles/gitea-runner/molecule.",
|
||||
help="Custom molecule directory (single-role mode). Default: auto-discovered under ansible/roles/*/molecule.",
|
||||
)
|
||||
@click.option(
|
||||
"--roles-root",
|
||||
|
||||
@@ -21,7 +21,20 @@ import click
|
||||
|
||||
from devx.molecule.platforms import PLATFORMS
|
||||
|
||||
ROLE_DIR = Path("ansible/roles/gitea-runner")
|
||||
DEFAULT_ROLES_ROOT = Path("ansible/roles")
|
||||
|
||||
|
||||
def _default_role_dir() -> Path:
|
||||
"""Auto-discover the single role directory with molecule scenarios."""
|
||||
roles_root = DEFAULT_ROLES_ROOT
|
||||
if not roles_root.is_dir():
|
||||
return roles_root / "gitea_runner" # sensible default for error message
|
||||
role_dirs = sorted(d for d in roles_root.iterdir() if (d / "molecule").is_dir())
|
||||
if role_dirs:
|
||||
return role_dirs[0]
|
||||
return roles_root / "role" # will produce a clear error
|
||||
|
||||
|
||||
SCENARIOS = ["default", "multi-instance", "lifecycle", "template-content", "deregister", "update"]
|
||||
|
||||
|
||||
@@ -72,15 +85,16 @@ def main(bin_dir: str) -> None:
|
||||
if not Path(molecule_bin).exists():
|
||||
raise click.ClickException(f"molecule not found at {molecule_bin}. Run 'make setup' first.")
|
||||
|
||||
if not ROLE_DIR.exists():
|
||||
raise click.ClickException(f"Role directory not found: {ROLE_DIR}")
|
||||
role_dir = _default_role_dir()
|
||||
if not role_dir.exists():
|
||||
raise click.ClickException(f"Role directory not found: {role_dir}")
|
||||
|
||||
base_env = dict(os.environ)
|
||||
base_env["ANSIBLE_ALLOW_BROKEN_CONDITIONALS"] = "true"
|
||||
base_env["ANSIBLE_INJECT_INVOCATION"] = "1"
|
||||
|
||||
for platform in PLATFORMS:
|
||||
rc = _run_platform(molecule_bin, platform, ROLE_DIR, SCENARIOS, base_env)
|
||||
rc = _run_platform(molecule_bin, platform, role_dir, SCENARIOS, base_env)
|
||||
if rc != 0:
|
||||
click.echo(f"FAILED on platform {platform['name']}", err=True)
|
||||
sys.exit(rc)
|
||||
|
||||
@@ -145,13 +145,19 @@ def resolve_role_dir(role: str, roles_root: Path | None, repo_root: Path) -> Pat
|
||||
"""Resolve the working directory for a molecule pair.
|
||||
|
||||
For multi-role pairs (role non-empty), uses ``roles_root/role``.
|
||||
For single-role pairs, uses ``repo_root/ansible/roles/gitea-runner``.
|
||||
For single-role pairs, auto-discovers the first role with a molecule/
|
||||
subdirectory under ``repo_root/ansible/roles/``.
|
||||
"""
|
||||
if role:
|
||||
if roles_root is None:
|
||||
roles_root = repo_root / "ansible" / "roles"
|
||||
return roles_root / role
|
||||
return repo_root / "ansible" / "roles" / "gitea-runner"
|
||||
roles_dir = repo_root / "ansible" / "roles"
|
||||
if roles_dir.is_dir():
|
||||
role_dirs = sorted(d for d in roles_dir.iterdir() if (d / "molecule").is_dir())
|
||||
if role_dirs:
|
||||
return role_dirs[0]
|
||||
return roles_dir / "role" # will produce a clear "not found" error
|
||||
|
||||
|
||||
@click.command()
|
||||
|
||||
@@ -30,7 +30,10 @@ Patterns detected:
|
||||
without patching it.
|
||||
3. **Unpatched known-subprocess-helpers** — functions known to spawn
|
||||
subprocesses (e.g. ``update_doc_versions``) called without patching.
|
||||
4. **Excessive iteration loops** — ``for _ in range(N)`` where N > 100.
|
||||
4. **Unpatched I/O functions** — functions known to do filesystem or
|
||||
network I/O (e.g. ``get_pat``, ``load_secrets``, ``requests.get``)
|
||||
called without patching.
|
||||
5. **Excessive iteration loops** — ``for _ in range(N)`` where N > 100.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -57,6 +60,23 @@ KNOWN_SUBPROCESS_HELPERS: dict[str, str] = {
|
||||
"run_cmd": "calls subprocess.run for shell commands",
|
||||
}
|
||||
|
||||
# Functions known to do filesystem or network I/O that should be mocked in tests.
|
||||
# Maps function name → description of what I/O it does.
|
||||
# If a test calls one of these without a corresponding @patch, it's a violation.
|
||||
KNOWN_IO_FUNCTIONS: dict[str, str] = { # nosec B105 — descriptions, not passwords
|
||||
"get_pat": "reads ZITADEL PAT from filesystem/env (ZitadelAuth._iter_sources)",
|
||||
"load_secrets": "reads YAML config file from disk",
|
||||
"get_customer_secret": "reads customer-specific config from disk",
|
||||
"requests.get": "performs HTTP GET to a real server",
|
||||
"requests.post": "performs HTTP POST to a real server",
|
||||
"requests.put": "performs HTTP PUT to a real server",
|
||||
"requests.patch": "performs HTTP PATCH to a real server",
|
||||
"requests.delete": "performs HTTP DELETE to a real server",
|
||||
"urlopen": "performs HTTP request to a real server",
|
||||
"httpx.get": "performs HTTP GET to a real server",
|
||||
"httpx.post": "performs HTTP POST to a real server",
|
||||
}
|
||||
|
||||
# Transitive dependencies: if a helper calls another helper that is patched,
|
||||
# the call is safe. Maps helper → set of function names it internally calls.
|
||||
# If ANY of these are in the test's patches, the helper call is safe.
|
||||
@@ -126,6 +146,20 @@ def _is_test_function(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
|
||||
return node.name.startswith("test_")
|
||||
|
||||
|
||||
def _has_integration_marker(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
|
||||
"""Check if a test function has @pytest.mark.integration decorator."""
|
||||
for decorator in node.decorator_list:
|
||||
# @pytest.mark.integration → ast.Attribute(attr='integration')
|
||||
if isinstance(decorator, ast.Attribute) and decorator.attr == "integration":
|
||||
return True
|
||||
# @pytest.mark.integration(...) → ast.Call(func=ast.Attribute(attr='integration'))
|
||||
if isinstance(decorator, ast.Call):
|
||||
func = decorator.func
|
||||
if isinstance(func, ast.Attribute) and func.attr == "integration":
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _get_called_name(node: ast.Call) -> str | None:
|
||||
func = node.func
|
||||
if isinstance(func, ast.Name):
|
||||
@@ -208,6 +242,11 @@ class TestIsolationVisitor(ast.NodeVisitor):
|
||||
self.generic_visit(node)
|
||||
return
|
||||
|
||||
# Skip integration tests — they intentionally do real I/O
|
||||
if _has_integration_marker(node):
|
||||
self.generic_visit(node)
|
||||
return
|
||||
|
||||
patches = _extract_patch_targets(node)
|
||||
info = TestFunctionInfo(
|
||||
name=node.name,
|
||||
@@ -299,6 +338,34 @@ class TestIsolationVisitor(ast.NodeVisitor):
|
||||
)
|
||||
)
|
||||
|
||||
# Check 4: Known I/O functions (filesystem/network)
|
||||
# Match by short name (e.g. "get_pat") or full name (e.g. "requests.get")
|
||||
sn = short_name or ""
|
||||
io_key = sn if sn in KNOWN_IO_FUNCTIONS else None
|
||||
if io_key is None and full_name and full_name in KNOWN_IO_FUNCTIONS:
|
||||
io_key = full_name
|
||||
if io_key and not (
|
||||
io_key in all_patches
|
||||
or sn in all_patches
|
||||
or any(io_key in p or sn in p for p in all_patches)
|
||||
or any(p.endswith(f".{sn}") for p in all_patches)
|
||||
):
|
||||
self.violations.append(
|
||||
Violation(
|
||||
file=self.file_path,
|
||||
line=node.lineno,
|
||||
col=node.col_offset,
|
||||
category="unpatched-io",
|
||||
message=_(
|
||||
"{func} called in test '{test}' without @patch — "
|
||||
'this function {desc}. Add @patch("<module>.{func}").',
|
||||
func=io_key,
|
||||
test=self._current_function.name,
|
||||
desc=KNOWN_IO_FUNCTIONS[io_key],
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
self.generic_visit(node)
|
||||
|
||||
def visit_For(self, node: ast.For) -> None:
|
||||
@@ -334,7 +401,13 @@ def find_test_files(test_path: Path) -> list[Path]:
|
||||
|
||||
|
||||
def analyze_file(file_path: Path, max_loop_iterations: int = DEFAULT_MAX_LOOP_ITERATIONS) -> list[Violation]:
|
||||
"""Analyze a single test file for isolation violations."""
|
||||
"""Analyze a single test file for isolation violations.
|
||||
|
||||
Files in ``integration/`` directories are skipped — integration tests
|
||||
intentionally do real I/O (subprocess, network, filesystem).
|
||||
"""
|
||||
if "integration" in file_path.parts:
|
||||
return []
|
||||
try:
|
||||
source = file_path.read_text()
|
||||
tree = ast.parse(source, filename=str(file_path))
|
||||
|
||||
@@ -566,6 +566,147 @@ class TestKnownHelpers:
|
||||
assert "subprocess" in HELPER_INTERNAL_CALLS["update_doc_versions"]
|
||||
|
||||
|
||||
class TestIOFunctionChecks:
|
||||
"""Tests for unpatched I/O function detection."""
|
||||
|
||||
def test_unpatched_get_pat_violation(self, tmp_path: Path) -> None:
|
||||
file = _write_test_file(
|
||||
tmp_path,
|
||||
"""
|
||||
from mymodule import get_pat
|
||||
|
||||
class TestExample:
|
||||
def test_calls_get_pat(self) -> None:
|
||||
result = get_pat("staging")
|
||||
""",
|
||||
)
|
||||
violations = analyze_file(file)
|
||||
assert len(violations) == 1
|
||||
assert violations[0].category == "unpatched-io"
|
||||
assert "get_pat" in violations[0].message
|
||||
|
||||
def test_patched_get_pat_no_violation(self, tmp_path: Path) -> None:
|
||||
file = _write_test_file(
|
||||
tmp_path,
|
||||
"""
|
||||
from unittest.mock import patch
|
||||
from mymodule import get_pat
|
||||
|
||||
class TestExample:
|
||||
@patch("mymodule.get_pat", return_value="pat")
|
||||
def test_patched(self, mock) -> None:
|
||||
result = get_pat("staging")
|
||||
""",
|
||||
)
|
||||
violations = analyze_file(file)
|
||||
assert violations == []
|
||||
|
||||
def test_unpatched_load_secrets_violation(self, tmp_path: Path) -> None:
|
||||
file = _write_test_file(
|
||||
tmp_path,
|
||||
"""
|
||||
from mymodule import load_secrets
|
||||
|
||||
class TestExample:
|
||||
def test_calls_load_secrets(self) -> None:
|
||||
result = load_secrets("staging")
|
||||
""",
|
||||
)
|
||||
violations = analyze_file(file)
|
||||
assert len(violations) == 1
|
||||
assert violations[0].category == "unpatched-io"
|
||||
assert "load_secrets" in violations[0].message
|
||||
|
||||
def test_patched_load_secrets_no_violation(self, tmp_path: Path) -> None:
|
||||
file = _write_test_file(
|
||||
tmp_path,
|
||||
"""
|
||||
from unittest.mock import patch
|
||||
from mymodule import load_secrets
|
||||
|
||||
class TestExample:
|
||||
@patch("mymodule.load_secrets", return_value={})
|
||||
def test_patched(self, mock) -> None:
|
||||
result = load_secrets("staging")
|
||||
""",
|
||||
)
|
||||
violations = analyze_file(file)
|
||||
assert violations == []
|
||||
|
||||
def test_unpatched_requests_get_violation(self, tmp_path: Path) -> None:
|
||||
file = _write_test_file(
|
||||
tmp_path,
|
||||
"""
|
||||
import requests
|
||||
|
||||
class TestExample:
|
||||
def test_calls_requests(self) -> None:
|
||||
resp = requests.get("https://example.com")
|
||||
""",
|
||||
)
|
||||
violations = analyze_file(file)
|
||||
assert len(violations) == 1
|
||||
assert violations[0].category == "unpatched-io"
|
||||
assert "requests.get" in violations[0].message or "get" in violations[0].message
|
||||
|
||||
def test_integration_marker_skips_subprocess(self, tmp_path: Path) -> None:
|
||||
"""@pytest.mark.integration tests should not be flagged for subprocess.run."""
|
||||
file = _write_test_file(
|
||||
tmp_path,
|
||||
"""
|
||||
import subprocess
|
||||
import pytest
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_real_subprocess():
|
||||
subprocess.run(["echo", "hello"])
|
||||
""",
|
||||
)
|
||||
violations = analyze_file(file)
|
||||
assert violations == []
|
||||
|
||||
def test_integration_marker_skips_sleep(self, tmp_path: Path) -> None:
|
||||
"""@pytest.mark.integration tests should not be flagged for time.sleep."""
|
||||
file = _write_test_file(
|
||||
tmp_path,
|
||||
"""
|
||||
import time
|
||||
import pytest
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_real_sleep():
|
||||
time.sleep(1)
|
||||
""",
|
||||
)
|
||||
violations = analyze_file(file)
|
||||
assert violations == []
|
||||
|
||||
def test_integration_marker_with_args_skips(self, tmp_path: Path) -> None:
|
||||
"""@pytest.mark.integration(...) with args should also be skipped."""
|
||||
file = _write_test_file(
|
||||
tmp_path,
|
||||
"""
|
||||
import subprocess
|
||||
import pytest
|
||||
|
||||
@pytest.mark.integration(scope="module")
|
||||
def test_real_subprocess():
|
||||
subprocess.run(["echo", "hello"])
|
||||
""",
|
||||
)
|
||||
violations = analyze_file(file)
|
||||
assert violations == []
|
||||
|
||||
def test_integration_directory_skipped(self, tmp_path: Path) -> None:
|
||||
"""Files in integration/ directories should be skipped entirely."""
|
||||
integration_dir = tmp_path / "integration"
|
||||
integration_dir.mkdir()
|
||||
file = integration_dir / "test_real_io.py"
|
||||
file.write_text("import subprocess\ndef test_real_subprocess():\n subprocess.run(['echo', 'hello'])\n")
|
||||
violations = analyze_file(file)
|
||||
assert violations == []
|
||||
|
||||
|
||||
class TestCli:
|
||||
"""Tests for the standalone CLI interface."""
|
||||
|
||||
|
||||
@@ -9,10 +9,10 @@ from click.testing import CliRunner
|
||||
|
||||
from devx.molecule.distribute_molecule import (
|
||||
DEFAULT_ROLES_ROOT,
|
||||
MOLECULE_ROOT,
|
||||
PLATFORMS,
|
||||
MultiRoleTestPair,
|
||||
TestPair,
|
||||
_default_molecule_root,
|
||||
_load_molecule_weights,
|
||||
_lpt_distribute,
|
||||
_scenario_weight,
|
||||
@@ -43,8 +43,25 @@ class TestDiscoverScenarios:
|
||||
discover_scenarios(tmp_path / "nonexistent")
|
||||
assert "not found" in str(exc.value)
|
||||
|
||||
def test_default_root_constant(self) -> None:
|
||||
assert Path("ansible/roles/gitea-runner/molecule") == MOLECULE_ROOT
|
||||
def test_default_root_auto_discovery(self, tmp_path: Path) -> None:
|
||||
"""_default_molecule_root auto-discovers first role with molecule/ dir."""
|
||||
# When no roles exist, returns a fallback path
|
||||
with patch("devx.molecule.distribute_molecule.DEFAULT_ROLES_ROOT", tmp_path / "roles"):
|
||||
result = _default_molecule_root()
|
||||
assert "molecule" in str(result)
|
||||
|
||||
# When a role has molecule/, it's discovered
|
||||
(tmp_path / "roles" / "my_role" / "molecule").mkdir(parents=True)
|
||||
with patch("devx.molecule.distribute_molecule.DEFAULT_ROLES_ROOT", tmp_path / "roles"):
|
||||
result = _default_molecule_root()
|
||||
assert result == tmp_path / "roles" / "my_role" / "molecule"
|
||||
|
||||
def test_default_root_no_molecule_dirs(self, tmp_path: Path) -> None:
|
||||
"""When roles exist but none have molecule/, returns fallback path."""
|
||||
(tmp_path / "roles" / "role_without_molecule").mkdir(parents=True)
|
||||
with patch("devx.molecule.distribute_molecule.DEFAULT_ROLES_ROOT", tmp_path / "roles"):
|
||||
result = _default_molecule_root()
|
||||
assert result == tmp_path / "roles" / "molecule"
|
||||
|
||||
|
||||
class TestPairEncoding:
|
||||
@@ -150,7 +167,7 @@ class TestCli:
|
||||
root = tmp_path / "molecule"
|
||||
(root / "alpha").mkdir(parents=True)
|
||||
(root / "beta").mkdir(parents=True)
|
||||
with patch("devx.molecule.distribute_molecule.MOLECULE_ROOT", root):
|
||||
with patch("devx.molecule.distribute_molecule._default_molecule_root", return_value=root):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, ["--list"])
|
||||
assert result.exit_code == 0
|
||||
@@ -176,7 +193,7 @@ class TestCli:
|
||||
root = tmp_path / "molecule"
|
||||
for s in ["a", "b", "c"]:
|
||||
(root / s).mkdir(parents=True)
|
||||
with patch("devx.molecule.distribute_molecule.MOLECULE_ROOT", root):
|
||||
with patch("devx.molecule.distribute_molecule._default_molecule_root", return_value=root):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, ["--max-runners", "3"])
|
||||
assert result.exit_code == 0
|
||||
@@ -191,7 +208,7 @@ class TestCli:
|
||||
|
||||
root = tmp_path / "molecule"
|
||||
(root / "alpha").mkdir(parents=True)
|
||||
with patch("devx.molecule.distribute_molecule.MOLECULE_ROOT", root):
|
||||
with patch("devx.molecule.distribute_molecule._default_molecule_root", return_value=root):
|
||||
runner = CliRunner()
|
||||
# 1-based index: "1" maps to internal 0
|
||||
result = runner.invoke(cli, ["--runner-index", "1", "--max-runners", "3"])
|
||||
@@ -209,7 +226,7 @@ class TestGithubEnv:
|
||||
scenario = root / "alpha"
|
||||
scenario.mkdir(parents=True)
|
||||
(scenario / "molecule.yml").write_text("name: alpha\n")
|
||||
with patch("devx.molecule.distribute_molecule.MOLECULE_ROOT", root):
|
||||
with patch("devx.molecule.distribute_molecule._default_molecule_root", return_value=root):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, ["--runner-index", "1", "--max-runners", "3", "--github-env"])
|
||||
assert result.exit_code == 0
|
||||
@@ -224,7 +241,7 @@ class TestGithubEnv:
|
||||
scenario = root / "alpha"
|
||||
scenario.mkdir(parents=True)
|
||||
(scenario / "molecule.yml").write_text("name: alpha\n")
|
||||
with patch("devx.molecule.distribute_molecule.MOLECULE_ROOT", root):
|
||||
with patch("devx.molecule.distribute_molecule._default_molecule_root", return_value=root):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(
|
||||
cli, ["--runner-index", "5", "--max-runners", "3", "--github-env", "--skip-if-excess"]
|
||||
@@ -240,7 +257,7 @@ class TestGithubEnv:
|
||||
scenario = root / "alpha"
|
||||
scenario.mkdir(parents=True)
|
||||
(scenario / "molecule.yml").write_text("name: alpha\n")
|
||||
with patch("devx.molecule.distribute_molecule.MOLECULE_ROOT", root):
|
||||
with patch("devx.molecule.distribute_molecule._default_molecule_root", return_value=root):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, ["--runner-index", "1", "--max-runners", "3", "--github-env"])
|
||||
assert result.exit_code != 0
|
||||
|
||||
@@ -104,12 +104,24 @@ class TestMain:
|
||||
assert result.exit_code != 0
|
||||
assert "Role directory not found" in result.output
|
||||
|
||||
def test_role_dir_no_molecule(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Roles dir exists but no role has molecule/ — should error."""
|
||||
monkeypatch.chdir(tmp_path)
|
||||
bin_dir = tmp_path / ".venv" / "bin"
|
||||
bin_dir.mkdir(parents=True)
|
||||
(bin_dir / "molecule").touch()
|
||||
(tmp_path / "ansible" / "roles" / "role_without_molecule").mkdir(parents=True)
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(molecule_all.main, ["--bin", str(bin_dir)])
|
||||
assert result.exit_code != 0
|
||||
assert "Role directory not found" in result.output
|
||||
|
||||
def test_all_pass(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
bin_dir = tmp_path / ".venv" / "bin"
|
||||
bin_dir.mkdir(parents=True)
|
||||
(bin_dir / "molecule").touch()
|
||||
(tmp_path / "ansible" / "roles" / "gitea-runner").mkdir(parents=True)
|
||||
(tmp_path / "ansible" / "roles" / "gitea-runner" / "molecule").mkdir(parents=True)
|
||||
|
||||
runner = CliRunner()
|
||||
with patch("devx.molecule.molecule_all._run_platform", return_value=0):
|
||||
@@ -122,7 +134,7 @@ class TestMain:
|
||||
bin_dir = tmp_path / ".venv" / "bin"
|
||||
bin_dir.mkdir(parents=True)
|
||||
(bin_dir / "molecule").touch()
|
||||
(tmp_path / "ansible" / "roles" / "gitea-runner").mkdir(parents=True)
|
||||
(tmp_path / "ansible" / "roles" / "gitea-runner" / "molecule").mkdir(parents=True)
|
||||
|
||||
runner = CliRunner()
|
||||
with patch("devx.molecule.molecule_all._run_platform", return_value=1):
|
||||
|
||||
@@ -524,9 +524,17 @@ class TestResolveRoleDir:
|
||||
result = resolve_role_dir("docker-base", None, tmp_path)
|
||||
assert result == tmp_path / "ansible" / "roles" / "docker-base"
|
||||
|
||||
def test_single_role_uses_default(self, tmp_path: Path) -> None:
|
||||
def test_single_role_auto_discovers(self, tmp_path: Path) -> None:
|
||||
"""Single-role mode auto-discovers first role with molecule/ dir."""
|
||||
roles_dir = tmp_path / "ansible" / "roles"
|
||||
(roles_dir / "my_role" / "molecule").mkdir(parents=True)
|
||||
result = resolve_role_dir("", None, tmp_path)
|
||||
assert result == tmp_path / "ansible" / "roles" / "gitea-runner"
|
||||
assert result == roles_dir / "my_role"
|
||||
|
||||
def test_single_role_no_roles_returns_fallback(self, tmp_path: Path) -> None:
|
||||
"""When no roles exist, returns a fallback path (will error at runtime)."""
|
||||
result = resolve_role_dir("", None, tmp_path)
|
||||
assert "roles" in str(result)
|
||||
|
||||
|
||||
class TestCliMultiRole:
|
||||
|
||||
Reference in New Issue
Block a user