Files
devx/src/devx/molecule/molecule_changed.py
T

159 lines
5.0 KiB
Python

"""Detect which Ansible roles changed and output their molecule scenarios.
Usage::
python -m devx.molecule.molecule_changed --print-targets
python -m devx.molecule.molecule_changed --base origin/master --print-roles
Outputs the list of make targets (e.g. molecule-docker-base) for roles
that have changed files vs the base ref. Used by ``make molecule-changed``
to run only the molecule scenarios affected by the current diff.
Role-to-target mapping is derived from the directory structure:
ansible/roles/<role>/ → molecule-<role>
For roles with multiple scenarios (e.g. app_container has customer-apps,
nextcloud, postgres-upgrade, simple-app), the base target runs all
scenarios for that role.
Playbooks that change also trigger molecule for the roles they include.
Shared infrastructure changes (ansible.cfg, requirements.yml, molecule/)
trigger all scenarios.
"""
from __future__ import annotations
import subprocess # nosec B404 — used to run git, a trusted binary
from pathlib import Path
import click
REPO_ROOT = Path.cwd()
# Map role names to make targets.
ROLE_TARGET_MAP: dict[str, str] = {
"app_container": "molecule-app-container",
"app_hardening": "molecule-app-hardening",
"crowdsec": "molecule-crowdsec",
"disk_cleanup": "molecule-disk-cleanup",
"docker_base": "molecule-docker-base",
"observability": "molecule-observability",
"restore": "molecule-restore",
"sso_config": "molecule-sso-config",
"storage": "molecule-storage",
"zitadel": "molecule-zitadel",
}
# Playbooks that map to molecule scenarios (via roles they include).
PLAYBOOK_ROLE_MAP: dict[str, list[str]] = {
"ansible/playbooks/deploy-observability.yml": ["observability", "docker_base", "zitadel", "crowdsec"],
"ansible/playbooks/deploy-customer.yml": ["app_container", "docker_base", "app_hardening", "sso_config"],
"ansible/playbooks/configure-oidc.yml": ["sso_config", "app_container"],
"ansible/playbooks/prepare-vms.yml": ["docker_base", "app_hardening", "storage", "disk_cleanup", "crowdsec"],
}
# Shared infrastructure that affects all molecule tests.
SHARED_PATHS = (
"ansible/ansible.cfg",
"ansible/requirements.yml",
"ansible/molecule/",
)
# Minimum path parts for a role file: ansible/roles/<role> (3 parts).
# Files inside the role have more parts, but we only need the role name.
_MIN_ROLE_PATH_PARTS = 3
def _run_git(args: list[str]) -> str: # pragma: no cover
"""Run a git command and return stdout."""
result = subprocess.run( # nosec
["git", *args],
cwd=REPO_ROOT,
capture_output=True,
text=True,
check=False,
)
return result.stdout
def get_changed_files(base: str) -> list[str]:
"""Get list of changed files vs base ref."""
for ref in [base, "master"]:
output = _run_git(["diff", "--name-only", f"{ref}...HEAD"])
if output.strip():
return sorted(output.strip().splitlines())
return []
def detect_changed_roles(changed_files: list[str]) -> set[str]:
"""Detect which roles have changed files."""
roles: set[str] = set()
for filepath in changed_files:
# Check if file is in a role directory
if filepath.startswith("ansible/roles/"):
parts = filepath.split("/")
if len(parts) >= _MIN_ROLE_PATH_PARTS:
roles.add(parts[2])
# Check if file is a playbook that maps to roles
if filepath in PLAYBOOK_ROLE_MAP:
roles.update(PLAYBOOK_ROLE_MAP[filepath])
# Check shared infrastructure — triggers all roles
for shared in SHARED_PATHS:
if filepath.startswith(shared):
return set(ROLE_TARGET_MAP.keys())
return roles
def roles_to_targets(roles: set[str]) -> list[str]:
"""Convert role names to make targets."""
targets = []
for role in sorted(roles):
target = ROLE_TARGET_MAP.get(role)
if target:
targets.append(target)
return targets
@click.command()
@click.option(
"--base",
default="origin/master",
help="Base ref to compare against (default: origin/master).",
)
@click.option(
"--print-targets",
is_flag=True,
help="Print make targets (e.g. molecule-docker-base).",
)
@click.option(
"--print-roles",
is_flag=True,
help="Print role names (default if no --print-targets).",
)
def main(base: str, print_targets: bool, print_roles: bool) -> None:
"""Detect which Ansible roles changed and output molecule scenarios."""
changed_files = get_changed_files(base)
if not changed_files:
click.echo("No changed files detected.", err=True)
return
roles = detect_changed_roles(changed_files)
if not roles:
click.echo("No molecule scenarios affected by changes.", err=True)
return
if print_targets:
for target in roles_to_targets(roles):
click.echo(target)
else:
for role in sorted(roles):
click.echo(role)
if __name__ == "__main__": # pragma: no cover
main()