Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
Set the `gathered_facts` parameter to `False` to simulate a run with no
facts gathered.
Set the `ara_record` parameter to `True` to simulate a run with an
ara_record task.
'''
playbook = m.Playbook(path='testing.yml')
playbook_file = m.File(path=playbook.path,
playbook=playbook,
is_playbook=True)
play = m.Play(playbook=playbook, name='test play')
host = m.Host(name='host-%04d' % random.randint(0, 9999),
playbook=playbook)
if ara_record:
task = m.Task(play=play, playbook=playbook, action='ara_record')
msg = 'Data recorded in ARA for this playbook.'
else:
task = m.Task(play=play, playbook=playbook, action='test-action')
msg = 'This is a test'
result = m.TaskResult(task=task, status='ok', host=host, result=msg)
ctx = dict(
playbook=playbook,
play=play,
file=playbook_file,
task=task,
host=host,
result=result)
if gather_facts:
def test_task(self):
task = m.Task.query.get(self.task.id)
assert task in self.playbook.tasks
assert task in self.play.tasks
def show_task(task):
task = models.Task.query.get(task)
if task is None:
abort(404)
task_results = (models.TaskResult.query
.join(models.Task)
.join(models.Host)
.join(models.Playbook)
.filter(models.Task.id == task.id)
.order_by(models.TaskResult.time_start))
if request.args.get('host'):
hosts = [str(host) for host in request.args.get('host').split(',')]
task_results = (task_results
.filter(models.Host.name.in_(hosts)))
if request.args.get('status'):
def show_task(task):
task = models.Task.query.get(task)
if task is None:
abort(404)
task_results = (models.TaskResult.query
.join(models.Task)
.join(models.Host)
.join(models.Playbook)
.filter(models.Task.id == task.id)
.order_by(models.TaskResult.time_start))
if request.args.get('host'):
hosts = [str(host) for host in request.args.get('host').split(',')]
task_results = (task_results
.filter(models.Host.name.in_(hosts)))
if request.args.get('status'):
status = request.args.get('status').split(',')
task_results = (res for res in task_results
if res.derived_status in status)
return render_template('task.html',
def index():
"""
This is not served anywhere in the web application.
It is used explicitly in the context of generating static files since
flask-frozen requires url_for's to crawl content.
url_for's are not used with result.show_result directly and are instead
dynamically generated through javascript for performance purposes.
"""
if current_app.config['ARA_PLAYBOOK_OVERRIDE'] is not None:
override = current_app.config['ARA_PLAYBOOK_OVERRIDE']
results = (models.TaskResult.query
.join(models.Task)
.filter(models.Task.playbook_id.in_(override)))
else:
results = models.TaskResult.query.all()
return render_template('task_result_index.html', results=results)
if args.output_file == '-':
output_stream = sys.stdout
else:
output_stream = open(args.output_file, 'wb')
# Create the output stream
output = StreamResultToBytes(output_stream)
# Create the test run
output.startTestRun()
if args.playbook is not None:
playbooks = args.playbook
results = (models.TaskResult().query
.join(models.Task)
.filter(models.TaskResult.task_id == models.Task.id)
.filter(models.Task.playbook_id.in_(playbooks)))
else:
results = models.TaskResult().query.all()
for result in results:
# Generate a fixed length identifier for the task
test_id = utils.generate_identifier(result)
# Assign the test_status value
if result.status in ('failed', 'unreachable'):
if result.ignore_errors is False:
test_status = 'xfail'
else:
test_status = 'fail'
elif result.status == 'skipped':
test_status = 'skip'
def ajax_results(playbook):
tasks_in_playbook = models.Task.query.filter(
models.Task.playbook_id == playbook)
if not utils.fast_count(tasks_in_playbook):
abort(404)
jinja = current_app.jinja_env
time = jinja.from_string('{{ time | timefmt }}')
action_link = jinja.get_template('ajax/action.html')
name_cell = jinja.get_template('ajax/task_name.html')
task_status_link = jinja.get_template('ajax/task_status.html')
results = dict()
results['data'] = list()
log.debug('Loading results')
for task in tasks_in_playbook:
task_results = task.task_results
for result in task_results:
def show_play(play):
play = models.Play.query.get(play)
if play is None:
abort(404)
tasks = (models.Task.query
.filter(models.Task.play_id == play.id)
.order_by(models.Task.sortkey))
return render_template('play.html',
play=play,
tasks=tasks)
def task(task):
task = models.Task.query.get(task)
return render_template('task.html', task=task)
def index():
"""
This is not served anywhere in the web application.
It is used explicitly in the context of generating static files since
flask-frozen requires url_for's to crawl content.
url_for's are not used with result.show_result directly and are instead
dynamically generated through javascript for performance purposes.
"""
if current_app.config['ARA_PLAYBOOK_OVERRIDE'] is not None:
override = current_app.config['ARA_PLAYBOOK_OVERRIDE']
results = (models.TaskResult.query
.join(models.Task)
.filter(models.Task.playbook_id.in_(override)))
else:
results = models.TaskResult.query.all()
return render_template('task_result_index.html', results=results)