Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add support for acceptable command options #1457

Closed
wants to merge 16 commits into from
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
129 changes: 128 additions & 1 deletion src/ansiblelint/rules/CommandsInsteadOfModulesRule.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,11 @@
# THE SOFTWARE.

import os
import sys
from typing import TYPE_CHECKING, Any, Dict, Union

from ansiblelint.rules import AnsibleLintRule
from ansiblelint.utils import convert_to_boolean, get_first_cmd_arg
from ansiblelint.utils import convert_to_boolean, get_first_cmd_arg, get_second_cmd_arg

if TYPE_CHECKING:
from typing import Optional
Expand Down Expand Up @@ -64,20 +65,146 @@ class CommandsInsteadOfModulesRule(AnsibleLintRule):
'yum': 'yum',
}


_executable_options = {
'git': ['branch', 'log'],
'systemctl': ['set-default', 'show-environment', 'status'],
}

def matchtask(
self, task: Dict[str, Any], file: 'Optional[Lintable]' = None
) -> Union[bool, str]:

if task['action']['__ansible_module__'] not in self._commands:
return False

first_cmd_arg = get_first_cmd_arg(task)
second_cmd_arg = get_second_cmd_arg(task)

if not first_cmd_arg:
return False

executable = os.path.basename(first_cmd_arg)

if (
second_cmd_arg
and executable in self._executable_options
and second_cmd_arg in self._executable_options[executable]
):
return False

if executable in self._modules and convert_to_boolean(
task['action'].get('warn', True)
):
message = '{0} used in place of {1} module'
return message.format(executable, self._modules[executable])
return False


if "pytest" in sys.modules:
import pytest

APT_GET = '''
- hosts: all
tasks:
- name: run apt-get update
command: apt-get update
'''

GIT_BRANCH = '''
- hosts: all
tasks:
- name: print current git branch
command: git branch
'''

GIT_LOG = '''
- hosts: all
tasks:
- name: print git log
command: git log
'''

RESTART_SSHD = '''
- hosts: all
tasks:
- name: restart sshd
command: systemctl restart sshd
'''

SYSTEMCTL_STATUS = '''
- hosts: all
tasks:
- name: show systemctl service status
command: systemctl status systemd-timesyncd
'''

SYSTEMD_ENVIRONMENT = '''
- hosts: all
tasks:
- name: show systemd environment
command: systemctl show-environment
'''

SYSTEMD_RUNLEVEL = '''
- hosts: all
tasks:
- name: set systemd runlevel
command: systemctl set-default multi-user.target
'''

@pytest.mark.parametrize(
'rule_runner', (CommandsInsteadOfModulesRule,), indirect=['rule_runner']
)
def test_apt_get(rule_runner: Any) -> None:
"""The apt module supports update."""
results = rule_runner.run_playbook(APT_GET)
assert len(results) == 1

@pytest.mark.parametrize(
'rule_runner', (CommandsInsteadOfModulesRule,), indirect=['rule_runner']
)
def test_restart_sshd(rule_runner: Any) -> None:
"""Restarting services is supported by the systemd module."""
results = rule_runner.run_playbook(RESTART_SSHD)
assert len(results) == 1

@pytest.mark.parametrize(
'rule_runner', (CommandsInsteadOfModulesRule,), indirect=['rule_runner']
)
def test_git_log(rule_runner: Any) -> None:
"""The git log command is not supported by the git module."""
results = rule_runner.run_playbook(GIT_LOG)
assert len(results) == 0

@pytest.mark.parametrize(
'rule_runner', (CommandsInsteadOfModulesRule,), indirect=['rule_runner']
)
def test_git_branch(rule_runner: Any) -> None:
"""The git branch command is not supported by the git module."""
results = rule_runner.run_playbook(GIT_BRANCH)
assert len(results) == 0

@pytest.mark.parametrize(
'rule_runner', (CommandsInsteadOfModulesRule,), indirect=['rule_runner']
)
def test_systemd_status(rule_runner: Any) -> None:
"""Set-default is not supported by the systemd module."""
results = rule_runner.run_playbook(SYSTEMCTL_STATUS)
assert len(results) == 0

@pytest.mark.parametrize(
'rule_runner', (CommandsInsteadOfModulesRule,), indirect=['rule_runner']
)
def test_systemd_environment(rule_runner: Any) -> None:
"""Showing the environment is not supported by the systemd module."""
results = rule_runner.run_playbook(SYSTEMD_ENVIRONMENT)
assert len(results) == 0

@pytest.mark.parametrize(
'rule_runner', (CommandsInsteadOfModulesRule,), indirect=['rule_runner']
)
def test_systemd_runlevel(rule_runner: Any) -> None:
"""Set-default is not supported by the systemd module."""
results = rule_runner.run_playbook(SYSTEMD_RUNLEVEL)
assert len(results) == 0
11 changes: 11 additions & 0 deletions src/ansiblelint/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -730,6 +730,17 @@ def get_first_cmd_arg(task: Dict[str, Any]) -> Any:
return first_cmd_arg


def get_second_cmd_arg(task: Dict[str, Any]) -> Any:
try:
if 'cmd' in task['action']:
second_cmd_arg = task['action']['cmd'].split()[1]
else:
second_cmd_arg = task['action']['__ansible_arguments__'][1]
except IndexError:
return None
return second_cmd_arg


def is_playbook(filename: str) -> bool:
"""
Check if the file is a playbook.
Expand Down