DEVX-152: feat: sync missing features from v0.49.x line to master

This commit is contained in:
2026-08-12 12:43:37 +00:00
parent 808e7a2e42
commit 1fc7a03c1a
13 changed files with 2584 additions and 20 deletions
+192
View File
@@ -0,0 +1,192 @@
"""Unit tests for devx.molecule.molecule_changed.
Verifies that the script correctly detects changed roles and maps
them to make targets.
"""
from __future__ import annotations
from unittest.mock import patch
from click.testing import CliRunner
from devx.molecule.molecule_changed import (
detect_changed_roles,
get_changed_files,
main,
roles_to_targets,
)
def test_detect_role_change():
"""A file in ansible/roles/<role>/ maps to that role."""
files = ["ansible/roles/docker_base/tasks/main.yml"]
roles = detect_changed_roles(files)
assert "docker_base" in roles
def test_detect_playbook_change():
"""A playbook change maps to its included roles."""
files = ["ansible/playbooks/deploy-observability.yml"]
roles = detect_changed_roles(files)
assert "observability" in roles
assert "docker_base" in roles
assert "zitadel" in roles
def test_detect_shared_infra_triggers_all():
"""ansible.cfg change triggers all roles."""
files = ["ansible/ansible.cfg"]
roles = detect_changed_roles(files)
assert len(roles) == 10 # all roles
def test_detect_no_ansible_changes():
"""Non-Ansible files don't trigger any roles."""
files = ["scripts/molecule_changed.py", "Makefile"]
roles = detect_changed_roles(files)
assert len(roles) == 0
def test_roles_to_targets():
"""Role names map to make targets."""
targets = roles_to_targets({"docker_base", "zitadel"})
assert "molecule-docker-base" in targets
assert "molecule-zitadel" in targets
def test_roles_to_targets_unknown_role():
"""Unknown roles are silently skipped."""
targets = roles_to_targets({"docker_base", "unknown_role"})
assert targets == ["molecule-docker-base"]
def test_main_no_changes():
"""When no files changed, outputs message to stderr."""
with patch("devx.molecule.molecule_changed.get_changed_files", return_value=[]):
runner = CliRunner()
result = runner.invoke(main, ["--print-targets"])
assert result.exit_code == 0
assert "No changed files" in result.output
def test_main_print_targets():
"""--print-targets outputs make targets."""
with patch(
"devx.molecule.molecule_changed.get_changed_files",
return_value=["ansible/roles/docker_base/tasks/main.yml"],
):
runner = CliRunner()
result = runner.invoke(main, ["--print-targets"])
assert result.exit_code == 0
assert "molecule-docker-base" in result.output
def test_main_print_roles():
"""--print-roles outputs role names."""
with patch(
"devx.molecule.molecule_changed.get_changed_files",
return_value=["ansible/roles/zitadel/tasks/main.yml"],
):
runner = CliRunner()
result = runner.invoke(main, ["--print-roles"])
assert result.exit_code == 0
assert "zitadel" in result.output
def test_main_no_ansible_changes():
"""When only non-Ansible files changed, outputs no scenarios message."""
with patch(
"devx.molecule.molecule_changed.get_changed_files",
return_value=["scripts/molecule_changed.py"],
):
runner = CliRunner()
result = runner.invoke(main, ["--print-targets"])
assert result.exit_code == 0
assert "No molecule scenarios" in result.output
def test_get_changed_files_with_mock():
"""get_changed_files returns files from git diff."""
with patch("devx.molecule.molecule_changed._run_git", return_value="file1\nfile2\n"):
files = get_changed_files("origin/master")
assert files == ["file1", "file2"]
def test_get_changed_files_falls_back_to_master():
"""When base ref has no diff, falls back to master."""
calls: list[list[str]] = []
def mock_git(args):
calls.append(args)
# First call (origin/master) returns empty, second (master) returns files
if "origin/master...HEAD" in args[2]:
return ""
return "ansible/roles/docker_base/tasks/main.yml\n"
with patch("devx.molecule.molecule_changed._run_git", side_effect=mock_git):
files = get_changed_files("origin/master")
assert files == ["ansible/roles/docker_base/tasks/main.yml"]
assert len(calls) == 2
def test_get_changed_files_empty():
"""When no changes in either ref, returns empty list."""
with patch("devx.molecule.molecule_changed._run_git", return_value=""):
files = get_changed_files("origin/master")
assert files == []
def test_detect_molecule_shared_path():
"""ansible/molecule/ change triggers all roles."""
files = ["ansible/molecule/Dockerfile"]
roles = detect_changed_roles(files)
assert len(roles) == 10
def test_detect_requirements_yml_triggers_all():
"""ansible/requirements.yml change triggers all roles."""
files = ["ansible/requirements.yml"]
roles = detect_changed_roles(files)
assert len(roles) == 10
def test_detect_configure_oidc_playbook():
"""configure-oidc.yml maps to sso_config and app_container."""
files = ["ansible/playbooks/configure-oidc.yml"]
roles = detect_changed_roles(files)
assert "sso_config" in roles
assert "app_container" in roles
def test_detect_prepare_vms_playbook():
"""prepare-vms.yml maps to all base roles."""
files = ["ansible/playbooks/prepare-vms.yml"]
roles = detect_changed_roles(files)
assert "docker_base" in roles
assert "app_hardening" in roles
assert "storage" in roles
assert "disk_cleanup" in roles
assert "crowdsec" in roles
def test_detect_deploy_customer_playbook():
"""deploy-customer.yml maps to its roles."""
files = ["ansible/playbooks/deploy-customer.yml"]
roles = detect_changed_roles(files)
assert "app_container" in roles
assert "docker_base" in roles
assert "app_hardening" in roles
assert "sso_config" in roles
def test_main_default_base():
"""main() with no --base uses origin/master."""
with patch(
"devx.molecule.molecule_changed.get_changed_files",
return_value=["ansible/roles/zitadel/tasks/main.yml"],
) as mock:
runner = CliRunner()
result = runner.invoke(main, ["--print-roles"])
assert result.exit_code == 0
mock.assert_called_once_with("origin/master")
@@ -0,0 +1,367 @@
"""Unit tests for devx.tools.check_ansible_no_log."""
from __future__ import annotations
from pathlib import Path
from click.testing import CliRunner
from devx.tools.check_ansible_no_log import _check_task, check_directory, main
def _make_task(name: str, action: str, value: str, **extra: object) -> dict:
"""Build a minimal task dict for testing."""
task: dict = {"name": name, action: value}
task.update(extra)
return task
class TestCheckTask:
def test_task_with_secret_and_no_log_passes(self):
task = _make_task(
"Safe task",
"ansible.builtin.shell",
"echo {{ _secrets.mattermost_admin_password }}",
no_log=True,
)
assert _check_task(task, Path("test.yml"), 1) == []
def test_task_with_secret_and_no_no_log_fails(self):
task = _make_task(
"Unsafe task",
"ansible.builtin.shell",
"echo {{ _secrets.mattermost_admin_password }}",
)
violations = _check_task(task, Path("test.yml"), 1)
assert len(violations) == 1
assert "no_log" in violations[0]
def test_task_without_secret_passes(self):
task = _make_task(
"Normal task",
"ansible.builtin.shell",
"echo hello world",
)
assert _check_task(task, Path("test.yml"), 1) == []
def test_task_with_jinja_no_log_passes(self):
task = _make_task(
"Safe task with jinja no_log",
"ansible.builtin.shell",
"echo {{ _secrets.mattermost_admin_password }}",
no_log="{{ not (debug_mode | default(false) | bool) }}",
)
assert _check_task(task, Path("test.yml"), 1) == []
def test_task_with_password_in_name_only_no_false_positive(self):
"""Task name contains 'password' but no secret value — should not flag."""
task = _make_task(
"Configure passwdqc in common-password",
"ansible.builtin.lineinfile",
"password required pam_passwdqc.so min=disabled,disabled,16,12,8",
)
assert _check_task(task, Path("test.yml"), 1) == []
def test_task_with_password_in_module_param_no_false_positive(self):
"""Module param named 'password' but value is a literal — no Jinja."""
task = {
"name": "Set user password",
"ansible.builtin.user": {
"name": "deploy",
"password_lock": True,
},
}
assert _check_task(task, Path("test.yml"), 1) == []
def test_task_with_vault_password_variable_fails(self):
task = _make_task(
"Unsafe vault task",
"ansible.builtin.shell",
"echo {{ vault_zitadel_db_password }}",
)
violations = _check_task(task, Path("test.yml"), 1)
assert len(violations) == 1
def test_task_with_nested_dict_secret_fails(self):
"""Secrets in nested dict values (e.g. set_fact) should be caught."""
task = {
"name": "Set secrets",
"ansible.builtin.set_fact": {
"db_password": "{{ vault_db_password }}",
"api_key": "{{ vault_api_key }}",
},
}
violations = _check_task(task, Path("test.yml"), 1)
assert len(violations) == 1
def test_task_with_no_log_none_passes(self):
"""no_log: None should count as not set (flagged)."""
task = _make_task(
"Unsafe task",
"ansible.builtin.shell",
"echo {{ _secrets.db_password }}",
no_log=None,
)
violations = _check_task(task, Path("test.yml"), 1)
assert len(violations) == 1
def test_task_with_secret_in_list_value_fails(self):
"""Secrets inside list values should be caught."""
task = {
"name": "Task with list secret",
"ansible.builtin.set_fact": {
"items": ["{{ _secrets.api_key }}", "normal_value"],
},
}
violations = _check_task(task, Path("test.yml"), 1)
assert len(violations) == 1
def test_task_with_api_key_secret_fails(self):
"""api_key in Jinja expression should be caught."""
task = _make_task(
"Unsafe task",
"ansible.builtin.shell",
"echo {{ my_api_key }}",
)
violations = _check_task(task, Path("test.yml"), 1)
assert len(violations) == 1
def test_task_with_secret_in_jinja_fails(self):
"""_secret in Jinja expression should be caught."""
task = _make_task(
"Unsafe task",
"ansible.builtin.shell",
"echo {{ my_secret }}",
)
violations = _check_task(task, Path("test.yml"), 1)
assert len(violations) == 1
def test_task_with_access_token_fails(self):
"""access_token in Jinja expression should be caught."""
task = _make_task(
"Unsafe task",
"ansible.builtin.shell",
"echo {{ my_access_token }}",
)
violations = _check_task(task, Path("test.yml"), 1)
assert len(violations) == 1
def test_task_with_non_secret_non_dict_non_list_value(self):
"""Non-str, non-dict, non-list values (e.g. int) should not crash."""
task = _make_task(
"Task with int",
"ansible.builtin.shell",
"echo hello",
some_int=42,
)
assert _check_task(task, Path("test.yml"), 1) == []
class TestCheckDirectory:
def test_clean_directory_passes(self, tmp_path: Path):
"""A directory with no secret-handling tasks should pass."""
role_dir = tmp_path / "roles" / "test_role"
(role_dir / "tasks").mkdir(parents=True)
(role_dir / "tasks" / "main.yml").write_text(
"- name: Normal task\n ansible.builtin.shell: echo hello\n changed_when: false\n"
)
assert check_directory(role_dir) == []
def test_unsafe_task_is_caught(self, tmp_path: Path):
"""A task with secrets but no no_log should be flagged."""
role_dir = tmp_path / "roles" / "test_role"
(role_dir / "tasks").mkdir(parents=True)
(role_dir / "tasks" / "main.yml").write_text(
"- name: Unsafe task\n ansible.builtin.shell: echo {{ _secrets.db_password }}\n changed_when: false\n"
)
violations = check_directory(role_dir)
assert len(violations) == 1
assert "Unsafe task" in violations[0]
def test_molecule_files_are_skipped(self, tmp_path: Path):
"""Molecule test files should not be scanned."""
role_dir = tmp_path / "roles" / "test_role"
mol_dir = role_dir / "molecule" / "default" / "tasks"
mol_dir.mkdir(parents=True)
(mol_dir / "main.yml").write_text(
"- name: Unsafe task in molecule\n ansible.builtin.shell: echo {{ _secrets.db_password }}\n"
)
assert check_directory(role_dir) == []
def test_playbook_format_is_parsed(self, tmp_path: Path):
"""Playbook files (list of plays with 'hosts') should be parsed."""
pb_dir = tmp_path / "playbooks"
pb_dir.mkdir(parents=True)
(pb_dir / "test.yml").write_text(
"---\n"
"- name: Test play\n"
" hosts: all\n"
" tasks:\n"
" - name: Unsafe task\n"
" ansible.builtin.shell: echo {{ _secrets.db_password }}\n"
)
violations = check_directory(tmp_path)
assert len(violations) == 1
assert "Unsafe task" in violations[0]
def test_invalid_yaml_is_skipped(self, tmp_path: Path):
"""Invalid YAML files should be skipped, not crash."""
role_dir = tmp_path / "roles" / "test_role"
(role_dir / "tasks").mkdir(parents=True)
(role_dir / "tasks" / "main.yml").write_text("{{ invalid yaml: [")
assert check_directory(role_dir) == []
def test_empty_yaml_doc_is_skipped(self, tmp_path: Path):
"""Empty YAML documents (None) should be skipped."""
role_dir = tmp_path / "roles" / "test_role"
(role_dir / "tasks").mkdir(parents=True)
(role_dir / "tasks" / "main.yml").write_text("---\n")
assert check_directory(role_dir) == []
def test_non_dict_non_list_doc_is_skipped(self, tmp_path: Path):
"""YAML docs that are neither dict nor list should be skipped."""
role_dir = tmp_path / "roles" / "test_role"
(role_dir / "tasks").mkdir(parents=True)
(role_dir / "tasks" / "main.yml").write_text("just a string\n")
assert check_directory(role_dir) == []
def test_task_file_with_non_dict_task_skipped(self, tmp_path: Path):
"""Non-dict items in a task list should be skipped."""
role_dir = tmp_path / "roles" / "test_role"
(role_dir / "tasks").mkdir(parents=True)
(role_dir / "tasks" / "main.yml").write_text(
"- just a string\n- name: Safe task\n ansible.builtin.shell: echo hello\n"
)
assert check_directory(role_dir) == []
def test_secret_in_list_value_is_caught(self, tmp_path: Path):
"""Secrets inside list values should be caught."""
role_dir = tmp_path / "roles" / "test_role"
(role_dir / "tasks").mkdir(parents=True)
(role_dir / "tasks" / "main.yml").write_text(
"- name: Task with list secret\n"
" ansible.builtin.set_fact:\n"
" items:\n"
' - "{{ _secrets.api_key }}"\n'
" - normal_value\n"
)
violations = check_directory(role_dir)
assert len(violations) == 1
def test_single_play_dict_format(self, tmp_path: Path):
"""A playbook that's a bare dict (not list of plays) should be parsed."""
pb_dir = tmp_path / "playbooks"
pb_dir.mkdir(parents=True)
(pb_dir / "test.yml").write_text(
"---\n"
"name: Single play\n"
"hosts: all\n"
"tasks:\n"
" - name: Unsafe task\n"
" ansible.builtin.shell: echo {{ _secrets.db_password }}\n"
)
violations = check_directory(tmp_path)
assert len(violations) == 1
def test_play_with_non_dict_play_skipped(self, tmp_path: Path):
"""Non-dict plays in a playbook list should be skipped."""
pb_dir = tmp_path / "playbooks"
pb_dir.mkdir(parents=True)
# First play is valid (makes is_plays=True), second is a non-dict
(pb_dir / "test.yml").write_text(
"---\n"
"- name: Safe play\n"
" hosts: all\n"
" tasks:\n"
" - name: Safe task\n"
" ansible.builtin.shell: echo hello\n"
'- "just a string as second play"\n'
)
assert check_directory(tmp_path) == []
def test_play_with_non_list_tasks_skipped(self, tmp_path: Path):
"""Plays where tasks is not a list should be skipped."""
pb_dir = tmp_path / "playbooks"
pb_dir.mkdir(parents=True)
(pb_dir / "test.yml").write_text('---\n- name: Play with bad tasks\n hosts: all\n tasks: "not a list"\n')
assert check_directory(tmp_path) == []
def test_play_with_non_dict_task_in_playbook(self, tmp_path: Path):
"""Non-dict tasks in a playbook should be skipped."""
pb_dir = tmp_path / "playbooks"
pb_dir.mkdir(parents=True)
(pb_dir / "test.yml").write_text(
"---\n"
"- name: Play\n"
" hosts: all\n"
" tasks:\n"
' - "just a string"\n'
" - name: Safe task\n"
" ansible.builtin.shell: echo hello\n"
)
assert check_directory(tmp_path) == []
def test_yaml_file_with_oserror_skipped(self, tmp_path: Path):
"""YAML files that can't be opened should be skipped."""
role_dir = tmp_path / "roles" / "test_role"
(role_dir / "tasks").mkdir(parents=True)
# Create a file that will cause OSError when opened
# (use a directory with .yml extension)
bad_file = role_dir / "tasks" / "main.yml"
bad_file.mkdir()
assert check_directory(role_dir) == []
class TestMain:
def test_main_passes_on_clean_dir(self, tmp_path: Path):
"""main() should exit 0 on a clean directory."""
role_dir = tmp_path / "roles" / "test_role"
(role_dir / "tasks").mkdir(parents=True)
(role_dir / "tasks" / "main.yml").write_text(
"- name: Normal task\n ansible.builtin.shell: echo hello\n changed_when: false\n"
)
runner = CliRunner()
result = runner.invoke(main, ["--path", str(role_dir)])
assert result.exit_code == 0
assert "OK" in result.output or "no_log" in result.output
def test_main_fails_on_unsafe_dir(self, tmp_path: Path):
"""main() should exit 1 when violations are found."""
role_dir = tmp_path / "roles" / "test_role"
(role_dir / "tasks").mkdir(parents=True)
(role_dir / "tasks" / "main.yml").write_text(
"- name: Unsafe task\n ansible.builtin.shell: echo {{ _secrets.db_password }}\n"
)
runner = CliRunner()
result = runner.invoke(main, ["--path", str(role_dir)])
assert result.exit_code == 1
assert "Unsafe task" in result.output
def test_main_returns_2_on_missing_dir(self, tmp_path: Path):
"""main() should exit 2 when the directory doesn't exist."""
runner = CliRunner()
result = runner.invoke(main, ["--path", str(tmp_path / "nonexistent")])
assert result.exit_code == 2
def test_main_with_ansible_dir_option(self, tmp_path: Path):
"""main() --ansible-dir should work like --path."""
role_dir = tmp_path / "roles" / "test_role"
(role_dir / "tasks").mkdir(parents=True)
(role_dir / "tasks" / "main.yml").write_text(
"- name: Unsafe task\n ansible.builtin.shell: echo {{ _secrets.db_password }}\n"
)
runner = CliRunner()
result = runner.invoke(main, ["--ansible-dir", str(tmp_path)])
assert result.exit_code == 1
def test_main_no_path_no_ansible_dir_uses_default(self, tmp_path: Path, monkeypatch):
"""main() with no args uses DEFAULT_ANSIBLE_DIR."""
import devx.tools.check_ansible_no_log as mod
role_dir = tmp_path / "roles" / "test_role"
(role_dir / "tasks").mkdir(parents=True)
(role_dir / "tasks" / "main.yml").write_text("- name: Normal task\n ansible.builtin.shell: echo hello\n")
monkeypatch.setattr(mod, "DEFAULT_ANSIBLE_DIR", tmp_path)
runner = CliRunner()
result = runner.invoke(main, [])
assert result.exit_code == 0
@@ -0,0 +1,198 @@
"""Unit tests for devx.tools.check_ansible_no_state_absent_on_db."""
from __future__ import annotations
from pathlib import Path
from click.testing import CliRunner
from devx.tools.check_ansible_no_state_absent_on_db import _check_file, _find_task_files, main
class TestCheckFile:
def test_clean_file_no_db_paths(self, tmp_path: Path):
"""A file with no DB paths should produce no violations."""
p = tmp_path / "test.yml"
p.write_text("- name: Safe task\n ansible.builtin.file:\n path: /opt/app/data\n state: directory\n")
assert _check_file(p, tmp_path) == []
def test_state_absent_on_zitadel_db_fails(self, tmp_path: Path):
"""state: absent on zitadel-db path should be flagged."""
p = tmp_path / "test.yml"
p.write_text(
"- name: Dangerous wipe\n ansible.builtin.file:\n path: /opt/postgres/zitadel-db\n state: absent\n"
)
violations = _check_file(p, tmp_path)
assert len(violations) >= 1
assert "state" in violations[0].lower() or "absent" in violations[0].lower()
def test_state_absent_on_var_lib_postgresql_fails(self, tmp_path: Path):
"""state: absent on /var/lib/postgresql/data should be flagged."""
p = tmp_path / "test.yml"
p.write_text(
"- name: Dangerous wipe\n ansible.builtin.file:\n path: /var/lib/postgresql/data\n state: absent\n"
)
violations = _check_file(p, tmp_path)
assert len(violations) >= 1
def test_state_absent_on_app_db_fails(self, tmp_path: Path):
"""state: absent on any *-db path should be flagged."""
p = tmp_path / "test.yml"
p.write_text(
"- name: Dangerous wipe\n ansible.builtin.file:\n path: /opt/postgres/gitea-db\n state: absent\n"
)
violations = _check_file(p, tmp_path)
assert len(violations) >= 1
def test_state_absent_with_pg_upgrade_context_passes(self, tmp_path: Path):
"""state: absent near DB path with upgrade-postgres context should pass."""
p = tmp_path / "test.yml"
p.write_text(
"- name: PG upgrade — remove old data\n"
" ansible.builtin.file:\n"
" path: /opt/postgres/zitadel-db\n"
" state: absent\n"
" when: pg_version_changed | default(false)\n"
)
violations = _check_file(p, tmp_path)
assert violations == []
def test_state_absent_with_pg_version_context_passes(self, tmp_path: Path):
"""state: absent near DB path with PG_VERSION context should pass."""
p = tmp_path / "test.yml"
p.write_text(
"- name: PG upgrade\n"
" ansible.builtin.file:\n"
" path: /opt/postgres/zitadel-db\n"
" state: absent\n"
" when: PG_VERSION is defined\n"
)
violations = _check_file(p, tmp_path)
assert violations == []
def test_state_absent_with_allow_marker_passes(self, tmp_path: Path):
"""state: absent with lint:allow-state-absent comment should pass."""
p = tmp_path / "test.yml"
p.write_text(
"# lint:allow-state-absent\n"
"- name: Intentional wipe\n"
" ansible.builtin.file:\n"
" path: /opt/postgres/zitadel-db\n"
" state: absent\n"
)
violations = _check_file(p, tmp_path)
assert violations == []
def test_state_present_on_db_path_passes(self, tmp_path: Path):
"""state: present (not absent) on DB path should pass."""
p = tmp_path / "test.yml"
p.write_text(
"- name: Safe task\n ansible.builtin.file:\n path: /opt/postgres/zitadel-db\n state: directory\n"
)
assert _check_file(p, tmp_path) == []
def test_nonexistent_file_returns_empty(self):
"""A nonexistent file should return no violations."""
assert _check_file(Path("/nonexistent/path/file.yml"), Path.cwd()) == []
def test_rm_rf_db_fails(self, tmp_path: Path):
"""rm -rf on a DB path should be flagged."""
p = tmp_path / "test.yml"
p.write_text("- name: Dangerous wipe\n ansible.builtin.shell: rm -rf /opt/postgres/zitadel-db\n")
violations = _check_file(p, tmp_path)
assert len(violations) >= 1
def test_relative_path_outside_repo(self, tmp_path: Path):
"""Files outside repo_root use the full path in display."""
p = tmp_path / "test.yml"
p.write_text(
"- name: Dangerous wipe\n ansible.builtin.file:\n path: /opt/postgres/zitadel-db\n state: absent\n"
)
violations = _check_file(p, Path("/other/repo"))
assert len(violations) >= 1
assert str(tmp_path) in violations[0] or "test.yml" in violations[0]
class TestFindTaskFiles:
def test_find_yml_files_in_directory(self, tmp_path: Path):
"""Should find .yml files in a directory."""
(tmp_path / "tasks").mkdir()
(tmp_path / "tasks" / "main.yml").write_text("[]")
(tmp_path / "tasks" / "other.yaml").write_text("[]")
files = _find_task_files(tmp_path)
assert len(files) == 2
def test_skip_molecule_files(self, tmp_path: Path):
"""Should skip files in molecule directories."""
(tmp_path / "molecule").mkdir()
(tmp_path / "molecule" / "test.yml").write_text("[]")
(tmp_path / "main.yml").write_text("[]")
files = _find_task_files(tmp_path)
assert len(files) == 1
assert "molecule" not in files[0].parts
def test_single_file_input(self, tmp_path: Path):
"""Should return the file itself if it's a .yml file."""
f = tmp_path / "test.yml"
f.write_text("[]")
files = _find_task_files(f)
assert files == [f]
def test_nonexistent_path_returns_empty(self):
"""A path that is neither a file nor a dir should return []."""
files = _find_task_files(Path("/nonexistent/path/that/does/not/exist"))
assert files == []
def test_non_yaml_file_skipped(self, tmp_path: Path):
"""Non-YAML files should not be included."""
f = tmp_path / "readme.txt"
f.write_text("not yaml")
assert _find_task_files(f) == []
class TestMain:
def test_main_no_violations_exit_zero(self, tmp_path: Path):
"""main() with a clean file should exit 0."""
f = tmp_path / "clean.yml"
f.write_text("- name: Safe task\n ansible.builtin.file:\n path: /opt/app\n state: directory\n")
result = CliRunner().invoke(main, ["--path", str(f)])
assert result.exit_code == 0
assert "OK" in result.output
def test_main_with_violations_exit_one(self, tmp_path: Path):
"""main() with a state: absent on a DB path should exit 1."""
f = tmp_path / "dangerous.yml"
f.write_text(
"- name: Dangerous wipe\n ansible.builtin.file:\n path: /opt/postgres/zitadel-db\n state: absent\n"
)
result = CliRunner().invoke(main, ["--path", str(f)])
assert result.exit_code == 1
assert "FAIL" in result.output
def test_main_path_to_clean_file(self, tmp_path: Path):
"""main() --path pointing to a specific clean file should exit 0."""
f = tmp_path / "tasks.yml"
f.write_text("- name: Safe\n ansible.builtin.file:\n path: /opt/app\n state: directory\n")
result = CliRunner().invoke(main, ["--path", str(f)])
assert result.exit_code == 0
def test_main_default_dirs_no_violations(self, tmp_path: Path, monkeypatch):
"""main() with no --path scans default dirs and exits 0."""
import devx.tools.check_ansible_no_state_absent_on_db as mod
(tmp_path / "clean.yml").write_text(
"- name: Safe\n ansible.builtin.file:\n path: /opt/app\n state: directory\n"
)
monkeypatch.setattr(mod, "DEFAULT_ANSIBLE_DIRS", [tmp_path])
result = CliRunner().invoke(main, [])
assert result.exit_code == 0
assert "OK" in result.output
def test_main_custom_ansible_dirs(self, tmp_path: Path):
"""main() --ansible-dir should work."""
f = tmp_path / "dangerous.yml"
f.write_text(
"- name: Dangerous wipe\n ansible.builtin.file:\n path: /opt/postgres/zitadel-db\n state: absent\n"
)
result = CliRunner().invoke(main, ["--ansible-dir", str(tmp_path)])
assert result.exit_code == 1
@@ -0,0 +1,340 @@
"""Unit tests for devx.tools.check_ansible_patterns."""
from __future__ import annotations
from pathlib import Path
from click.testing import CliRunner
from devx.tools.check_ansible_patterns import (
_check_file,
_check_task,
_check_tasks,
_find_task_files,
_is_legitimate_devnull,
_is_legitimate_or_true,
main,
)
def _make_task(name: str, action: str, value: str, **extra: object) -> dict:
"""Build a minimal task dict for testing."""
task: dict = {"name": name, action: value}
task.update(extra)
return task
class TestIsLegitimateOrTrue:
def test_cleanup_task_name_is_legitimate(self):
assert _is_legitimate_or_true("docker rm old-container", "Remove old container")
def test_prune_task_name_is_legitimate(self):
assert _is_legitimate_or_true("docker image prune -f", "Prune unused images")
def test_docker_rm_command_is_legitimate(self):
assert _is_legitimate_or_true("docker rm -f mycontainer", "Some task")
def test_provision_task_is_not_legitimate(self):
assert not _is_legitimate_or_true("curl -X POST https://api/app || true", "Provision OIDC client")
def test_sync_task_name_is_legitimate(self):
assert _is_legitimate_or_true("psql -c 'ALTER USER' || true", "Sync PostgreSQL password")
class TestCheckTask:
def test_or_true_on_provision_task_fails(self, tmp_path: Path):
task = _make_task(
"Provision OIDC client",
"ansible.builtin.shell",
"curl -X POST https://zitadel/api || true",
)
violations = _check_task(task, tmp_path / "test.yml", 1, tmp_path)
assert len(violations) >= 1
assert "|| true" in violations[0]
def test_or_true_on_cleanup_task_passes(self, tmp_path: Path):
task = _make_task(
"Remove old container",
"ansible.builtin.shell",
"docker rm -f old-container || true",
)
assert _check_task(task, tmp_path / "test.yml", 1, tmp_path) == []
def test_failed_when_false_on_provision_fails(self, tmp_path: Path):
task = _make_task(
"Provision OIDC client",
"ansible.builtin.shell",
"curl -X POST https://zitadel/api",
failed_when=False,
)
violations = _check_task(task, tmp_path / "test.yml", 1, tmp_path)
assert any("failed_when" in v for v in violations)
def test_failed_when_false_on_stop_passes(self, tmp_path: Path):
task = _make_task(
"Stop ZITADEL containers",
"ansible.builtin.shell",
"docker stop zitadel",
failed_when=False,
)
assert _check_task(task, tmp_path / "test.yml", 1, tmp_path) == []
def test_failed_when_false_on_check_passes(self, tmp_path: Path):
task = _make_task(
"Check if ZITADEL is running",
"ansible.builtin.shell",
"docker inspect zitadel",
failed_when=False,
)
assert _check_task(task, tmp_path / "test.yml", 1, tmp_path) == []
def test_allow_marker_in_name_passes(self, tmp_path: Path):
task = _make_task(
"Provision OIDC #lint:allow-failure-masking",
"ansible.builtin.shell",
"curl -X POST https://zitadel/api || true",
failed_when=False,
)
assert _check_task(task, tmp_path / "test.yml", 1, tmp_path) == []
def test_safe_task_no_violations(self, tmp_path: Path):
task = _make_task(
"Create directory",
"ansible.builtin.file",
"path=/opt/app state=directory",
)
assert _check_task(task, tmp_path / "test.yml", 1, tmp_path) == []
def test_relative_path_outside_repo(self, tmp_path: Path):
"""Files outside repo_root use the full path in display."""
task = _make_task(
"Provision OIDC",
"ansible.builtin.shell",
"curl || true",
)
other_dir = Path("/tmp/other")
violations = _check_task(task, other_dir / "test.yml", 1, tmp_path)
assert len(violations) >= 1
class TestCheckFile:
def test_clean_file_passes(self, tmp_path: Path):
p = tmp_path / "test.yml"
p.write_text("- name: Safe task\n ansible.builtin.file:\n path: /opt/app\n state: directory\n")
assert _check_file(p, tmp_path) == []
def test_dangerous_pattern_detected(self, tmp_path: Path):
p = tmp_path / "test.yml"
p.write_text(
"- name: Provision OIDC\n"
" ansible.builtin.shell: |\n"
" curl -X POST https://api/app || true\n"
" failed_when: false\n"
)
violations = _check_file(p, tmp_path)
assert len(violations) >= 1
def test_file_level_allow_marker_passes(self, tmp_path: Path):
p = tmp_path / "test.yml"
p.write_text(
"# lint:allow-failure-masking\n"
"- name: Provision OIDC\n"
" ansible.builtin.shell: |\n"
" curl -X POST https://api/app || true\n"
" failed_when: false\n"
)
assert _check_file(p, tmp_path) == []
def test_nonexistent_file_returns_empty(self):
assert _check_file(Path("/nonexistent/path/file.yml"), Path.cwd()) == []
def test_yaml_parse_error_returns_empty(self, tmp_path: Path):
p = tmp_path / "test.yml"
p.write_text("name: Provision OIDC\n shell: curl || true\n: invalid: [")
assert _check_file(p, tmp_path) == []
def test_dict_doc_playbook_with_tasks(self, tmp_path: Path):
p = tmp_path / "playbook.yml"
p.write_text(
"- hosts: all\n"
" tasks:\n"
" - name: Provision OIDC\n"
" ansible.builtin.shell: curl -X POST https://api/app || true\n"
)
violations = _check_file(p, tmp_path)
assert any("|| true" in v for v in violations)
def test_dict_doc_with_pre_tasks_and_post_tasks(self, tmp_path: Path):
p = tmp_path / "playbook.yml"
p.write_text(
"- hosts: all\n"
" pre_tasks:\n"
" - name: Provision secret\n"
" ansible.builtin.shell: curl -X POST https://api/app || true\n"
" post_tasks:\n"
" - name: Provision OIDC\n"
" ansible.builtin.shell: curl -X POST https://api/app || true\n"
" handlers:\n"
" - name: Provision password\n"
" ansible.builtin.shell: curl -X POST https://api/app || true\n"
)
violations = _check_file(p, tmp_path)
assert len(violations) >= 3
def test_block_tasks_in_list_item(self, tmp_path: Path):
p = tmp_path / "tasks.yml"
p.write_text(
"- name: Outer task\n"
" block:\n"
" - name: Provision OIDC\n"
" ansible.builtin.shell: curl -X POST https://api/app || true\n"
" - name: Provision secret\n"
" ansible.builtin.shell: curl -X POST https://api/app || true\n"
)
violations = _check_file(p, tmp_path)
assert any("|| true" in v for v in violations)
def test_empty_doc_skipped(self, tmp_path: Path):
p = tmp_path / "test.yml"
p.write_text("---\nnull\n---\n- name: Provision OIDC\n ansible.builtin.shell: curl || true\n")
violations = _check_file(p, tmp_path)
assert any("|| true" in v for v in violations)
def test_pure_dict_doc_with_tasks(self, tmp_path: Path):
p = tmp_path / "playbook.yml"
p.write_text(
"hosts: all\n"
"tasks:\n"
" - name: Provision OIDC\n"
" ansible.builtin.shell: curl -X POST https://api/app || true\n"
)
violations = _check_file(p, tmp_path)
assert any("|| true" in v for v in violations)
class TestIsLegitimateDevnull:
def test_cleanup_task_is_legitimate(self):
assert _is_legitimate_devnull("docker rm old-container 2>/dev/null", "Remove old container")
def test_provision_task_is_not_legitimate(self):
assert not _is_legitimate_devnull("curl -X POST https://api/app 2>/dev/null", "Provision OIDC client")
class TestCheckTasks:
def test_tasks_section_checked(self, tmp_path: Path):
doc = {
"tasks": [
{"name": "Provision OIDC", "ansible.builtin.shell": "curl || true"},
],
}
errors: list[str] = []
_check_tasks(doc, tmp_path / "test.yml", errors, tmp_path)
assert any("|| true" in e for e in errors)
def test_block_inside_tasks_section(self, tmp_path: Path):
doc = {
"tasks": [
{
"name": "Outer",
"block": [
{"name": "Provision secret", "ansible.builtin.shell": "curl || true"},
],
},
],
}
errors: list[str] = []
_check_tasks(doc, tmp_path / "test.yml", errors, tmp_path)
assert any("|| true" in e for e in errors)
def test_non_list_section_ignored(self, tmp_path: Path):
doc = {"tasks": "not a list"}
errors: list[str] = []
_check_tasks(doc, tmp_path / "test.yml", errors, tmp_path)
assert errors == []
def test_non_dict_task_ignored(self, tmp_path: Path):
doc = {"tasks": ["just a string"]}
errors: list[str] = []
_check_tasks(doc, tmp_path / "test.yml", errors, tmp_path)
assert errors == []
class TestFindTaskFiles:
def test_single_file(self, tmp_path: Path):
p = tmp_path / "main.yml"
p.write_text("- name: test\n")
assert _find_task_files(p) == [p]
def test_single_yaml_file(self, tmp_path: Path):
p = tmp_path / "main.yaml"
p.write_text("- name: test\n")
assert _find_task_files(p) == [p]
def test_non_yaml_file_returns_empty(self, tmp_path: Path):
p = tmp_path / "main.txt"
p.write_text("hello\n")
assert _find_task_files(p) == []
def test_directory_finds_yaml_files(self, tmp_path: Path):
(tmp_path / "a.yml").write_text("- name: a\n")
(tmp_path / "sub").mkdir()
(tmp_path / "sub" / "b.yaml").write_text("- name: b\n")
(tmp_path / "ignore.txt").write_text("nope\n")
result = _find_task_files(tmp_path)
names = {f.name for f in result}
assert names == {"a.yml", "b.yaml"}
def test_directory_skips_molecule(self, tmp_path: Path):
(tmp_path / "a.yml").write_text("- name: a\n")
(tmp_path / "molecule").mkdir()
(tmp_path / "molecule" / "scenario.yml").write_text("- name: mol\n")
result = _find_task_files(tmp_path)
assert all("molecule" not in f.parts for f in result)
def test_nonexistent_path_returns_empty(self):
assert _find_task_files(Path("/nonexistent/path/xyz")) == []
class TestMain:
def test_main_clean_file_exit_zero(self, tmp_path: Path):
p = tmp_path / "clean.yml"
p.write_text("- name: Safe task\n ansible.builtin.file:\n path: /opt/app\n state: directory\n")
result = CliRunner().invoke(main, ["--path", str(p)])
assert result.exit_code == 0
assert "OK" in result.output
def test_main_violation_exit_one(self, tmp_path: Path):
p = tmp_path / "bad.yml"
p.write_text(
"- name: Provision OIDC\n"
" ansible.builtin.shell: curl -X POST https://api/app || true\n"
" failed_when: false\n"
)
result = CliRunner().invoke(main, ["--path", str(p)])
assert result.exit_code == 1
assert "FAIL" in result.output
def test_main_directory(self, tmp_path: Path):
(tmp_path / "clean.yml").write_text(
"- name: Safe task\n ansible.builtin.file:\n path: /opt\n state: directory\n"
)
result = CliRunner().invoke(main, ["--path", str(tmp_path)])
assert result.exit_code == 0
def test_main_default_dirs(self, tmp_path: Path, monkeypatch):
import devx.tools.check_ansible_patterns as mod
(tmp_path / "clean.yml").write_text(
"- name: Safe task\n ansible.builtin.file:\n path: /opt\n state: directory\n"
)
monkeypatch.setattr(mod, "DEFAULT_ANSIBLE_DIRS", [tmp_path])
result = CliRunner().invoke(main, [])
assert result.exit_code == 0
assert "OK" in result.output
def test_main_custom_ansible_dirs(self, tmp_path: Path):
(tmp_path / "bad.yml").write_text(
"- name: Provision OIDC\n ansible.builtin.shell: curl -X POST https://api/app || true\n"
)
result = CliRunner().invoke(main, ["--ansible-dir", str(tmp_path)])
assert result.exit_code == 1
+264
View File
@@ -0,0 +1,264 @@
"""Unit tests for devx.tools.check_jinja_expr.
Verifies that the check correctly validates Jinja2 expressions,
catches reversed strftime filter arguments (the OBL-INFRA-508 bug),
and passes on valid expressions.
"""
from __future__ import annotations
from pathlib import Path
from unittest.mock import patch
from click.testing import CliRunner
from devx.tools.check_jinja_expr import (
_check_file,
_default_ansible_dirs,
_extract_expressions,
_render_expression,
main,
)
def test_render_valid_expression():
"""Valid Jinja expression renders without error."""
ok, _ = _render_expression("'%Y-%m-%dT%H:%M:%S+00:00' | strftime(1735689600)")
assert ok
def test_render_reversed_strftime_args():
"""Reversed strftime filter args are detected as an error."""
ok, msg = _render_expression("(now().timestamp() | int + 3600) | strftime('%Y-%m-%dT%H:%M:%S+00:00')")
assert not ok
assert "reversed" in msg.lower()
def test_render_correct_strftime_args():
"""Correct strftime filter args pass."""
ok, _ = _render_expression("'%Y-%m-%dT%H:%M:%S+00:00' | strftime((now().timestamp() | int) + 3600)")
assert ok
def test_render_unknown_filter():
"""Unknown filter is reported as an error."""
ok, msg = _render_expression("'test' | nonexistent_filter")
assert not ok
assert "filter" in msg.lower()
def test_extract_skips_go_templates():
"""Go template syntax ({{.Field}}) is not extracted."""
content = "cmd: docker inspect --format '{{.State.Running}}' container"
expressions = _extract_expressions(content)
assert len(expressions) == 0
def test_extract_skips_single_char():
"""Single-character fragments are not extracted."""
content = 'value: "{{ \' }}"'
expressions = _extract_expressions(content)
assert len(expressions) == 0
def test_extract_skips_multiline():
"""Multi-line expressions are skipped."""
content = 'value: "{{\n something\n}}"'
expressions = _extract_expressions(content)
assert len(expressions) == 0
def test_extract_skips_unbalanced():
"""Expressions with unbalanced braces (from partial capture) are skipped."""
content = "value: \"{{ default({'k': {}}, true) }}\""
expressions = _extract_expressions(content)
# The regex captures {{ default({'k': {}} — unbalanced parens
# because the inner }} terminates the match early.
# All extracted expressions should have balanced braces.
for expr in expressions:
assert expr.count("{") == expr.count("}")
def test_extract_valid_expression():
"""Valid Jinja expressions are extracted."""
content = "value: \"{{ my_var | default('x') }}\""
expressions = _extract_expressions(content)
assert "my_var | default('x')" in expressions
def test_main_passes_on_clean_file(tmp_path: Path) -> None:
"""A file with valid expressions passes."""
test_file = tmp_path / "tasks.yml"
test_file.write_text("value: \"{{ my_var | default('x') }}\"\nother: \"{{ '%Y' | strftime(1735689600) }}\"\n")
runner = CliRunner()
result = runner.invoke(main, ["--path", str(test_file)])
assert result.exit_code == 0
def test_main_no_violations_empty_dir(tmp_path: Path) -> None:
"""An empty directory passes."""
runner = CliRunner()
result = runner.invoke(main, ["--path", str(tmp_path)])
assert result.exit_code == 0
def test_main_catches_reversed_strftime(tmp_path: Path) -> None:
"""A file with reversed strftime args is flagged."""
test_file = tmp_path / "test.yml"
test_file.write_text("value: \"{{ (now().timestamp() | int + 3600) | strftime('%Y-%m-%dT%H:%M:%S+00:00') }}\"\n")
runner = CliRunner()
result = runner.invoke(main, ["--path", str(test_file)])
assert result.exit_code == 1
assert "reversed" in result.output.lower()
def test_render_skips_undefined_var():
"""Undefined variables are skipped (MockDict returns mock for missing keys)."""
ok, _ = _render_expression("nonexistent_var_in_mock | upper")
assert ok
def test_render_skips_other_errors():
"""Non-filter errors from missing mocks are skipped."""
ok, _ = _render_expression("some_undefined.attr.method()")
assert ok
def test_extract_skips_backtick():
"""Backtick fragments are skipped (caught by single-char check)."""
content = 'value: "{{ ` }}"'
expressions = _extract_expressions(content)
assert len(expressions) == 0
def test_extract_skips_network_settings():
"""Expressions with .NetworkSettings. patterns are skipped."""
content = 'value: "{{ foo.NetworkSettings.IPAddress }}"'
expressions = _extract_expressions(content)
assert len(expressions) == 0
def test_extract_skips_unbalanced_parens():
"""Expressions with unbalanced parens are skipped."""
content = 'value: "{{ foo(bar }}"'
expressions = _extract_expressions(content)
assert len(expressions) == 0
def test_extract_skips_unbalanced_braces():
"""Expressions with unbalanced braces are skipped."""
content = 'value: "{{ foo{bar }}"'
expressions = _extract_expressions(content)
assert len(expressions) == 0
def test_extract_skips_unbalanced_brackets():
"""Expressions with unbalanced brackets are skipped."""
content = 'value: "{{ foo[0 }}"'
expressions = _extract_expressions(content)
assert len(expressions) == 0
def test_extract_skips_control_flow():
"""Control flow fragments starting with % are skipped."""
content = 'value: "{{ % if x }}"'
expressions = _extract_expressions(content)
assert len(expressions) == 0
def test_check_file_outside_repo(tmp_path: Path) -> None:
"""Files outside REPO_ROOT are handled (no relative_to error)."""
test_file = tmp_path / "test.yml"
test_file.write_text("value: \"{{ (now().timestamp() | int + 3600) | strftime('%Y-%m-%dT%H:%M:%S+00:00') }}\"\n")
violations = _check_file(test_file, Path("/other/repo"))
assert len(violations) == 1
assert "reversed" in violations[0].lower()
def test_render_mock_dict_missing_key():
"""MockDict returns a mock for missing keys (no UndefinedError)."""
ok, _ = _render_expression("undefined_var.some_attr | upper")
assert ok
def test_render_syntax_error():
"""Syntax errors are reported as failures."""
ok, msg = _render_expression("{{ invalid syntax +")
assert not ok
assert "Syntax error" in msg
def test_render_unknown_filter_error():
"""Unknown filters are reported as failures (not skipped)."""
ok, msg = _render_expression("'test' | nonexistent_filter")
assert not ok
assert "filter" in msg.lower()
def test_render_generic_exception_skipped():
"""Non-filter exceptions from missing mocks are skipped."""
# replace() with no args triggers TypeError (missing required args)
# which is not a filter-not-found or strftime error — should be skipped.
ok, msg = _render_expression("my_var | replace")
assert ok
assert "Skipped" in msg
def test_default_ansible_dirs():
"""_default_ansible_dirs returns playbooks and roles paths."""
dirs = _default_ansible_dirs()
assert Path.cwd() / "ansible" / "playbooks" in dirs
assert Path.cwd() / "ansible" / "roles" in dirs
def test_main_default_dirs(tmp_path: Path) -> None:
"""Running with no --path scans default dirs (uses small temp fixture)."""
(tmp_path / "playbooks").mkdir()
(tmp_path / "roles").mkdir()
(tmp_path / "playbooks" / "test.yml").write_text("value: \"{{ my_var | default('x') }}\"\n")
with patch(
"devx.tools.check_jinja_expr._default_ansible_dirs",
return_value=[tmp_path / "playbooks", tmp_path / "roles"],
):
runner = CliRunner()
result = runner.invoke(main, [])
assert result.exit_code == 0
def test_main_custom_ansible_dirs(tmp_path: Path) -> None:
"""--ansible-dir option works."""
(tmp_path / "test.yml").write_text("value: \"{{ my_var | default('x') }}\"\n")
runner = CliRunner()
result = runner.invoke(main, ["--ansible-dir", str(tmp_path)])
assert result.exit_code == 0
def test_extract_skips_println():
"""Expressions with 'println' (Go template) are skipped."""
content = 'value: "{{ println something }}"'
expressions = _extract_expressions(content)
assert len(expressions) == 0
def test_extract_skips_state_dot():
"""Expressions with .State. patterns are skipped."""
content = 'value: "{{ foo.State.Running }}"'
expressions = _extract_expressions(content)
assert len(expressions) == 0
def test_render_now_with_format():
"""now() with a format argument works."""
ok, _ = _render_expression("now('%Y-%m-%d')")
assert ok
def test_find_yaml_files_skips_molecule(tmp_path: Path) -> None:
"""Molecule directories are excluded from file search."""
from devx.tools.check_jinja_expr import _find_yaml_files
(tmp_path / "tasks.yml").write_text("value: test\n")
(tmp_path / "molecule").mkdir()
(tmp_path / "molecule" / "test.yml").write_text("value: test\n")
files = _find_yaml_files(tmp_path)
assert all("molecule" not in f.parts for f in files)