Coverage for lib/ansible/cli/playbook.py : 34%

# (c) 2012, Michael DeHaan <michael.dehaan@gmail.com> 

# 

# This file is part of Ansible 

# 

# Ansible is free software: you can redistribute it and/or modify 

# it under the terms of the GNU General Public License as published by 

# the Free Software Foundation, either version 3 of the License, or 

# (at your option) any later version. 

# 

# Ansible is distributed in the hope that it will be useful, 

# but WITHOUT ANY WARRANTY; without even the implied warranty of 

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

# GNU General Public License for more details. 

# 

# You should have received a copy of the GNU General Public License 

# along with Ansible. If not, see <http://www.gnu.org/licenses/>. 

######################################################## 

from __future__ import (absolute_import, division, print_function) 

__metaclass__ = type 

import os 

import stat 

from ansible.cli import CLI 

from ansible.errors import AnsibleError, AnsibleOptionsError 

from ansible.executor.playbook_executor import PlaybookExecutor 

from ansible.playbook.block import Block 

from ansible.playbook.play_context import PlayContext 

try: 

from __main__ import display 

except ImportError: 

from ansible.utils.display import Display 

display = Display() 

class PlaybookCLI(CLI): 

''' the tool to run *Ansible playbooks*, which are a configuration and multinode deployment system. 

See the project home page (https://docs.ansible.com) for more information. ''' 

def parse(self): 

# create parser for CLI options 

parser = CLI.base_parser( 

usage="%prog [options] playbook.yml [playbook2 ...]", 

connect_opts=True, 

meta_opts=True, 

runas_opts=True, 

subset_opts=True, 

check_opts=True, 

inventory_opts=True, 

runtask_opts=True, 

vault_opts=True, 

fork_opts=True, 

module_opts=True, 

desc="Runs Ansible playbooks, executing the defined tasks on the targeted hosts.", 

) 

# ansible playbook specific opts 

parser.add_option('--list-tasks', dest='listtasks', action='store_true', 

help="list all tasks that would be executed") 

parser.add_option('--list-tags', dest='listtags', action='store_true', 

help="list all available tags") 

parser.add_option('--step', dest='step', action='store_true', 

help="one-step-at-a-time: confirm each task before running") 

parser.add_option('--start-at-task', dest='start_at_task', 

help="start the playbook at the task matching this name") 

self.parser = parser 

super(PlaybookCLI, self).parse() 

74 ↛ 75line 74 didn't jump to line 75, because the condition on line 74 was never true if len(self.args) == 0: 

raise AnsibleOptionsError("You must specify a playbook file to run") 

display.verbosity = self.options.verbosity 

self.validate_conflicts(runas_opts=True, vault_opts=True, fork_opts=True) 

def run(self): 

super(PlaybookCLI, self).run() 

# Note: slightly wrong, this is written so that implicit localhost 

# Manage passwords 

sshpass = None 

becomepass = None 

passwords = {} 

# initial error check, to make sure all specified playbooks are accessible 

# before we start running anything through the playbook executor 

for playbook in self.args: 

93 ↛ 94line 93 didn't jump to line 94, because the condition on line 93 was never true if not os.path.exists(playbook): 

raise AnsibleError("the playbook: %s could not be found" % playbook) 

95 ↛ 96line 95 didn't jump to line 96, because the condition on line 95 was never true if not (os.path.isfile(playbook) or stat.S_ISFIFO(os.stat(playbook).st_mode)): 

raise AnsibleError("the playbook: %s does not appear to be a file" % playbook) 

# don't deal with privilege escalation or passwords when we don't need to 

99 ↛ 104line 99 didn't jump to line 104, because the condition on line 99 was never false if not self.options.listhosts and not self.options.listtasks and not self.options.listtags and not self.options.syntax: 

self.normalize_become_options() 

(sshpass, becomepass) = self.ask_passwords() 

passwords = {'conn_pass': sshpass, 'become_pass': becomepass} 

loader, inventory, variable_manager = self._play_prereqs(self.options) 

# (which is not returned in list_hosts()) is taken into account for 

# warning if inventory is empty. But it can't be taken into account for 

# checking if limit doesn't match any hosts. Instead we don't worry about 

# limit if only implicit localhost was in inventory to start with. 

# 

# Fix this when we rewrite inventory by making localhost a real host (and thus show up in list_hosts()) 

hosts = CLI.get_host_list(inventory, self.options.subset) 

# flush fact cache if requested 

115 ↛ 116line 115 didn't jump to line 116, because the condition on line 115 was never true if self.options.flush_cache: 

self._flush_cache(inventory, variable_manager) 

# create the playbook executor, which manages running the plays via a task queue manager 

pbex = PlaybookExecutor(playbooks=self.args, inventory=inventory, variable_manager=variable_manager, loader=loader, options=self.options, 

passwords=passwords) 

results = pbex.run() 

124 ↛ 125line 124 didn't jump to line 125, because the condition on line 124 was never true if isinstance(results, list): 

for p in results: 

display.display('\nplaybook: %s' % p['playbook']) 

for idx, play in enumerate(p['plays']): 

if play._included_path is not None: 

loader.set_basedir(play._included_path) 

else: 

pb_dir = os.path.realpath(os.path.dirname(p['playbook'])) 

loader.set_basedir(pb_dir) 

msg = "\n play #%d (%s): %s" % (idx + 1, ','.join(play.hosts), play.name) 

mytags = set(play.tags) 

msg += '\tTAGS: [%s]' % (','.join(mytags)) 

if self.options.listhosts: 

playhosts = set(inventory.get_hosts(play.hosts)) 

msg += "\n pattern: %s\n hosts (%d):" % (play.hosts, len(playhosts)) 

for host in playhosts: 

msg += "\n %s" % host 

display.display(msg) 

all_tags = set() 

if self.options.listtags or self.options.listtasks: 

taskmsg = '' 

if self.options.listtasks: 

taskmsg = ' tasks:\n' 

def _process_block(b): 

taskmsg = '' 

for task in b.block: 

if isinstance(task, Block): 

taskmsg += _process_block(task) 

else: 

if task.action == 'meta': 

continue 

all_tags.update(task.tags) 

if self.options.listtasks: 

cur_tags = list(mytags.union(set(task.tags))) 

cur_tags.sort() 

if task.name: 

taskmsg += " %s" % task.get_name() 

else: 

taskmsg += " %s" % task.action 

taskmsg += "\tTAGS: [%s]\n" % ', '.join(cur_tags) 

return taskmsg 

all_vars = variable_manager.get_vars(play=play) 

play_context = PlayContext(play=play, options=self.options) 

for block in play.compile(): 

block = block.filter_tagged_tasks(play_context, all_vars) 

if not block.has_tasks(): 

continue 

taskmsg += _process_block(block) 

if self.options.listtags: 

cur_tags = list(mytags.union(all_tags)) 

cur_tags.sort() 

taskmsg += " TASK TAGS: [%s]\n" % ', '.join(cur_tags) 

display.display(taskmsg) 

return 0 

else: 

return results 

def _flush_cache(self, inventory, variable_manager): 

for host in inventory.list_hosts(): 

hostname = host.get_name() 

variable_manager.clear_facts(hostname)