mirror of
https://github.com/ansible/awx.git
synced 2026-02-16 18:50:04 -03:30
357 lines
13 KiB
Python
357 lines
13 KiB
Python
# -*- coding: utf-8 -*-
|
|
import tempfile
|
|
import json
|
|
import yaml
|
|
import pytest
|
|
from itertools import count
|
|
|
|
from awx.main.utils.encryption import encrypt_value
|
|
from awx.main.tasks import RunJob
|
|
from awx.main.models import (
|
|
Job,
|
|
JobTemplate,
|
|
JobLaunchConfig,
|
|
WorkflowJobTemplate,
|
|
Project,
|
|
Inventory
|
|
)
|
|
from awx.main.utils.safe_yaml import SafeLoader
|
|
|
|
ENCRYPTED_SECRET = encrypt_value('secret')
|
|
|
|
|
|
class DistinctParametrize(object):
|
|
|
|
def __init__(self):
|
|
self._gen = count(0)
|
|
|
|
def __call__(self, value):
|
|
return str(next(self._gen))
|
|
|
|
|
|
@pytest.mark.survey
|
|
class SurveyVariableValidation:
|
|
|
|
def test_survey_answers_as_string(self, job_template_factory):
|
|
objects = job_template_factory(
|
|
'job-template-with-survey',
|
|
survey=[{'variable': 'var1', 'type': 'text'}],
|
|
persisted=False)
|
|
jt = objects.job_template
|
|
user_extra_vars = json.dumps({'var1': 'asdf'})
|
|
accepted, ignored, errors = jt._accept_or_ignore_job_kwargs(extra_vars=user_extra_vars)
|
|
assert ignored.get('extra_vars', {}) == {}, [str(element) for element in errors]
|
|
assert 'var1' in accepted['extra_vars']
|
|
|
|
def test_job_template_survey_variable_validation(self, job_template_factory):
|
|
objects = job_template_factory(
|
|
'survey_variable_validation',
|
|
organization='org1',
|
|
inventory='inventory1',
|
|
credential='cred1',
|
|
persisted=False,
|
|
)
|
|
obj = objects.job_template
|
|
obj.survey_spec = {
|
|
"description": "",
|
|
"spec": [
|
|
{
|
|
"required": True,
|
|
"min": 0,
|
|
"default": "5",
|
|
"max": 1024,
|
|
"question_description": "",
|
|
"choices": "",
|
|
"variable": "a",
|
|
"question_name": "Whosyourdaddy",
|
|
"type": "text"
|
|
}
|
|
],
|
|
"name": ""
|
|
}
|
|
obj.survey_enabled = True
|
|
accepted, rejected, errors = obj.accept_or_ignore_variables({"a": 5})
|
|
assert rejected == {"a": 5}
|
|
assert accepted == {}
|
|
assert str(errors[0]) == "Value 5 for 'a' expected to be a string."
|
|
|
|
|
|
@pytest.fixture
|
|
def job(mocker):
|
|
ret = mocker.MagicMock(**{
|
|
'decrypted_extra_vars.return_value': '{\"secret_key\": \"my_password\"}',
|
|
'display_extra_vars.return_value': '{\"secret_key\": \"$encrypted$\"}',
|
|
'extra_vars_dict': {"secret_key": "my_password"},
|
|
'pk': 1, 'job_template.pk': 1, 'job_template.name': '',
|
|
'created_by.pk': 1, 'created_by.username': 'admin',
|
|
'launch_type': 'manual',
|
|
'awx_meta_vars.return_value': {},
|
|
'inventory.get_script_data.return_value': {}})
|
|
ret.project = mocker.MagicMock(scm_revision='asdf1234')
|
|
return ret
|
|
|
|
|
|
@pytest.fixture
|
|
def job_with_survey():
|
|
return Job(
|
|
name="test-job-with-passwords",
|
|
extra_vars=json.dumps({
|
|
'submitter_email': 'foobar@redhat.com',
|
|
'secret_key': '6kQngg3h8lgiSTvIEb21',
|
|
'SSN': '123-45-6789'}),
|
|
survey_passwords={
|
|
'secret_key': '$encrypted$',
|
|
'SSN': '$encrypted$'})
|
|
|
|
|
|
@pytest.mark.survey
|
|
def test_job_survey_password_redaction(job_with_survey):
|
|
"""Tests the Job model's funciton to redact passwords from
|
|
extra_vars - used when displaying job information"""
|
|
assert json.loads(job_with_survey.display_extra_vars()) == {
|
|
'submitter_email': 'foobar@redhat.com',
|
|
'secret_key': '$encrypted$',
|
|
'SSN': '$encrypted$'}
|
|
|
|
|
|
@pytest.mark.survey
|
|
def test_survey_passwords_not_in_extra_vars():
|
|
"""Tests that survey passwords not included in extra_vars are
|
|
not included when displaying job information"""
|
|
job = Job(
|
|
name="test-survey-not-in",
|
|
extra_vars=json.dumps({
|
|
'submitter_email': 'foobar@redhat.com'}),
|
|
survey_passwords={
|
|
'secret_key': '$encrypted$',
|
|
'SSN': '$encrypted$'})
|
|
assert json.loads(job.display_extra_vars()) == {
|
|
'submitter_email': 'foobar@redhat.com',
|
|
}
|
|
|
|
|
|
def test_job_safe_args_redacted_passwords(job):
|
|
"""Verify that safe_args hides passwords in the job extra_vars"""
|
|
kwargs = {'ansible_version': '2.1', 'private_data_dir': tempfile.mkdtemp()}
|
|
run_job = RunJob()
|
|
safe_args = run_job.build_safe_args(job, **kwargs)
|
|
ev_index = safe_args.index('-e') + 1
|
|
extra_var_file = open(safe_args[ev_index][1:], 'r')
|
|
extra_vars = yaml.load(extra_var_file, SafeLoader)
|
|
extra_var_file.close()
|
|
assert extra_vars['secret_key'] == '$encrypted$'
|
|
|
|
|
|
def test_job_args_unredacted_passwords(job, tmpdir_factory):
|
|
kwargs = {'ansible_version': '2.1', 'private_data_dir': tempfile.mkdtemp()}
|
|
run_job = RunJob()
|
|
args = run_job.build_args(job, **kwargs)
|
|
ev_index = args.index('-e') + 1
|
|
extra_var_file = open(args[ev_index][1:], 'r')
|
|
extra_vars = yaml.load(extra_var_file, SafeLoader)
|
|
extra_var_file.close()
|
|
assert extra_vars['secret_key'] == 'my_password'
|
|
|
|
|
|
def test_launch_config_has_unprompted_vars(survey_spec_factory):
|
|
jt = JobTemplate(
|
|
survey_enabled = True,
|
|
survey_spec = survey_spec_factory(['question1', 'question2'])
|
|
)
|
|
unprompted_config = JobLaunchConfig(
|
|
extra_data = {
|
|
'question1': 'foobar',
|
|
'question4': 'foobar'
|
|
}
|
|
)
|
|
assert unprompted_config.has_unprompted(jt)
|
|
allowed_config = JobLaunchConfig(
|
|
extra_data = {
|
|
'question1': 'foobar',
|
|
'question2': 'foobar'
|
|
}
|
|
)
|
|
assert not allowed_config.has_unprompted(jt)
|
|
|
|
|
|
@pytest.mark.survey
|
|
def test_update_kwargs_survey_invalid_default(survey_spec_factory):
|
|
spec = survey_spec_factory('var2')
|
|
spec['spec'][0]['required'] = False
|
|
spec['spec'][0]['min'] = 3
|
|
spec['spec'][0]['default'] = 1
|
|
jt = JobTemplate(name="test-jt", survey_spec=spec, survey_enabled=True, extra_vars="var2: 2")
|
|
defaulted_extra_vars = jt._update_unified_job_kwargs({}, {})
|
|
assert 'extra_vars' in defaulted_extra_vars
|
|
# Make sure we did not set the invalid default of 1
|
|
assert json.loads(defaulted_extra_vars['extra_vars'])['var2'] == 2
|
|
|
|
|
|
@pytest.mark.survey
|
|
def test_display_survey_spec_encrypts_default(survey_spec_factory):
|
|
spec = survey_spec_factory('var2')
|
|
spec['spec'][0]['type'] = 'password'
|
|
spec['spec'][0]['default'] = 'some-default'
|
|
jt = JobTemplate(name="test-jt", survey_spec=spec, survey_enabled=True)
|
|
assert jt.display_survey_spec()['spec'][0]['default'] == '$encrypted$'
|
|
|
|
|
|
@pytest.mark.survey
|
|
@pytest.mark.parametrize("question_type,default,min,max,expect_use,expect_value", [
|
|
("text", "", 0, 0, True, ''), # default used
|
|
("text", "", 1, 0, False, 'N/A'), # value less than min length
|
|
("password", "", 1, 0, False, 'N/A'), # passwords behave the same as text
|
|
("multiplechoice", "", 0, 0, False, 'N/A'), # historical bug
|
|
("multiplechoice", "zeb", 0, 0, False, 'N/A'), # zeb not in choices
|
|
("multiplechoice", "coffee", 0, 0, True, 'coffee'),
|
|
("multiselect", None, 0, 0, False, 'N/A'), # NOTE: Behavior is arguable, value of [] may be prefered
|
|
("multiselect", "", 0, 0, False, 'N/A'),
|
|
("multiselect", ["zeb"], 0, 0, False, 'N/A'),
|
|
("multiselect", ["milk"], 0, 0, True, ["milk"]),
|
|
("multiselect", ["orange\nmilk"], 0, 0, False, 'N/A'), # historical bug
|
|
])
|
|
def test_optional_survey_question_defaults(
|
|
survey_spec_factory, question_type, default, min, max, expect_use, expect_value):
|
|
spec = survey_spec_factory([
|
|
{
|
|
"required": False,
|
|
"default": default,
|
|
"choices": "orange\nmilk\nchocolate\ncoffee",
|
|
"variable": "c",
|
|
"min": min,
|
|
"max": max,
|
|
"type": question_type
|
|
},
|
|
])
|
|
jt = JobTemplate(name="test-jt", survey_spec=spec, survey_enabled=True)
|
|
defaulted_extra_vars = jt._update_unified_job_kwargs({}, {})
|
|
element = spec['spec'][0]
|
|
if expect_use:
|
|
assert jt._survey_element_validation(element, {element['variable']: element['default']}) == []
|
|
else:
|
|
assert jt._survey_element_validation(element, {element['variable']: element['default']})
|
|
if expect_use:
|
|
assert json.loads(defaulted_extra_vars['extra_vars'])['c'] == expect_value
|
|
else:
|
|
assert 'c' not in defaulted_extra_vars['extra_vars']
|
|
|
|
|
|
@pytest.mark.survey
|
|
@pytest.mark.parametrize("question_type,default,maxlen,kwargs,expected", [
|
|
('text', None, 5, {}, {}),
|
|
('text', '', 5, {}, {'x': ''}),
|
|
('text', 'y', 5, {}, {'x': 'y'}),
|
|
('text', 'too-long', 5, {}, {}),
|
|
('password', None, 5, {}, {}),
|
|
('password', '', 5, {}, {'x': ''}),
|
|
('password', ENCRYPTED_SECRET, 5, {}, {}), # len(secret) == 6, invalid
|
|
('password', ENCRYPTED_SECRET, 10, {}, {'x': ENCRYPTED_SECRET}), # len(secret) < 10, valid
|
|
('password', None, 5, {'extra_vars': {'x': '$encrypted$'}}, {}),
|
|
('password', '', 5, {'extra_vars': {'x': '$encrypted$'}}, {'x': ''}),
|
|
('password', None, 5, {'extra_vars': {'x': 'y'}}, {'x': 'y'}),
|
|
('password', '', 5, {'extra_vars': {'x': 'y'}}, {'x': 'y'}),
|
|
('password', 'foo', 5, {'extra_vars': {'x': 'y'}}, {'x': 'y'}),
|
|
('password', None, 5, {'extra_vars': {'x': ''}}, {'x': ''}),
|
|
('password', '', 5, {'extra_vars': {'x': ''}}, {'x': ''}),
|
|
('password', 'foo', 5, {'extra_vars': {'x': ''}}, {'x': ''}),
|
|
('password', ENCRYPTED_SECRET, 5, {'extra_vars': {'x': '$encrypted$'}}, {}),
|
|
('password', ENCRYPTED_SECRET, 10, {'extra_vars': {'x': '$encrypted$'}}, {'x': ENCRYPTED_SECRET}),
|
|
], ids=DistinctParametrize())
|
|
def test_survey_encryption_defaults(survey_spec_factory, question_type, default, maxlen, kwargs, expected):
|
|
spec = survey_spec_factory([
|
|
{
|
|
"required": True,
|
|
"variable": "x",
|
|
"min": 0,
|
|
"max": maxlen,
|
|
"type": question_type
|
|
},
|
|
])
|
|
if default is not None:
|
|
spec['spec'][0]['default'] = default
|
|
else:
|
|
spec['spec'][0].pop('default', None)
|
|
jt = JobTemplate(name="test-jt", survey_spec=spec, survey_enabled=True)
|
|
extra_vars = json.loads(jt._update_unified_job_kwargs({}, kwargs).get('extra_vars'))
|
|
assert extra_vars == expected
|
|
|
|
|
|
@pytest.mark.survey
|
|
class TestWorkflowSurveys:
|
|
def test_update_kwargs_survey_defaults(self, survey_spec_factory):
|
|
"Assure that the survey default over-rides a JT variable"
|
|
spec = survey_spec_factory('var1')
|
|
spec['spec'][0]['default'] = 3
|
|
spec['spec'][0]['required'] = False
|
|
wfjt = WorkflowJobTemplate(
|
|
name="test-wfjt",
|
|
survey_spec=spec,
|
|
survey_enabled=True,
|
|
extra_vars="var1: 5"
|
|
)
|
|
updated_extra_vars = wfjt._update_unified_job_kwargs({}, {})
|
|
assert 'extra_vars' in updated_extra_vars
|
|
assert json.loads(updated_extra_vars['extra_vars'])['var1'] == 3
|
|
assert wfjt.can_start_without_user_input()
|
|
|
|
def test_variables_needed_to_start(self, survey_spec_factory):
|
|
"Assure that variables_needed_to_start output contains mandatory vars"
|
|
spec = survey_spec_factory(['question1', 'question2', 'question3'])
|
|
spec['spec'][0]['required'] = False
|
|
spec['spec'][1]['required'] = True
|
|
spec['spec'][2]['required'] = False
|
|
wfjt = WorkflowJobTemplate(
|
|
name="test-wfjt",
|
|
survey_spec=spec,
|
|
survey_enabled=True,
|
|
extra_vars="question2: hiworld"
|
|
)
|
|
assert wfjt.variables_needed_to_start == ['question2']
|
|
assert not wfjt.can_start_without_user_input()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.parametrize('provided_vars,valid', [
|
|
({'tmpl_var': 'bar'}, True), # same as template, not counted as prompts
|
|
({'tmpl_var': 'bar2'}, False), # different value from template, not okay
|
|
({'tmpl_var': 'bar', 'a': 2}, False), # extra key, not okay
|
|
({'tmpl_var': 'bar', False: 2}, False), # Falsy key
|
|
({'tmpl_var': 'bar', u'🐉': u'🐉'}, False), # dragons
|
|
])
|
|
class TestExtraVarsNoPrompt:
|
|
def process_vars_and_assert(self, tmpl, provided_vars, valid):
|
|
prompted_fields, ignored_fields, errors = tmpl._accept_or_ignore_job_kwargs(
|
|
extra_vars=provided_vars
|
|
)
|
|
if valid:
|
|
assert not ignored_fields
|
|
assert not errors
|
|
else:
|
|
assert ignored_fields
|
|
assert errors
|
|
|
|
def test_jt_extra_vars_counting(self, provided_vars, valid):
|
|
jt = JobTemplate(
|
|
name='foo',
|
|
extra_vars={'tmpl_var': 'bar'},
|
|
project=Project(),
|
|
project_id=42,
|
|
playbook='helloworld.yml',
|
|
inventory=Inventory(),
|
|
inventory_id=42
|
|
)
|
|
prompted_fields, ignored_fields, errors = jt._accept_or_ignore_job_kwargs(
|
|
extra_vars=provided_vars
|
|
)
|
|
self.process_vars_and_assert(jt, provided_vars, valid)
|
|
|
|
def test_wfjt_extra_vars_counting(self, provided_vars, valid):
|
|
wfjt = WorkflowJobTemplate(
|
|
name='foo',
|
|
extra_vars={'tmpl_var': 'bar'}
|
|
)
|
|
prompted_fields, ignored_fields, errors = wfjt._accept_or_ignore_job_kwargs(
|
|
extra_vars=provided_vars
|
|
)
|
|
self.process_vars_and_assert(wfjt, provided_vars, valid)
|