Updating tower_org to use the new tower_api format

Pass sanity and unit tests, update tests

Remove placeholder test function, convert tower_host module, fix misc typos
This commit is contained in:
John Westcott IV
2020-01-29 13:33:04 -05:00
committed by beeankha
parent c23d605a7a
commit f89061da41
14 changed files with 120 additions and 167 deletions

View File

@@ -42,77 +42,6 @@ def sanitize_dict(din):
@pytest.fixture
def run_module(request):
def rf(module_name, module_params, request_user):
def new_request(self, method, url, **kwargs):
kwargs_copy = kwargs.copy()
if 'data' in kwargs:
kwargs_copy['data'] = json.loads(kwargs['data'])
if 'params' in kwargs and method == 'GET':
# query params for GET are handled a bit differently by
# tower-cli and python requests as opposed to REST framework APIRequestFactory
kwargs_copy.setdefault('data', {})
if isinstance(kwargs['params'], dict):
kwargs_copy['data'].update(kwargs['params'])
elif isinstance(kwargs['params'], list):
for k, v in kwargs['params']:
kwargs_copy['data'][k] = v
# make request
rf = _request(method.lower())
django_response = rf(url, user=request_user, expect=None, **kwargs_copy)
# requests library response object is different from the Django response, but they are the same concept
# this converts the Django response object into a requests response object for consumption
resp = Response()
py_data = django_response.data
sanitize_dict(py_data)
resp._content = bytes(json.dumps(django_response.data), encoding='utf8')
resp.status_code = django_response.status_code
if request.config.getoption('verbose') > 0:
logger.info(
'%s %s by %s, code:%s',
method, '/api/' + url.split('/api/')[1],
request_user.username, resp.status_code
)
return resp
stdout_buffer = io.StringIO()
# Requies specific PYTHONPATH, see docs
# Note that a proper Ansiballz explosion of the modules will have an import path like:
# ansible_collections.awx.awx.plugins.modules.{}
# We should consider supporting that in the future
resource_module = importlib.import_module('plugins.modules.{0}'.format(module_name))
if not isinstance(module_params, dict):
raise RuntimeError('Module params must be dict, got {0}'.format(type(module_params)))
# Ansible params can be passed as an invocation argument or over stdin
# this short circuits within the AnsibleModule interface
def mock_load_params(self):
self.params = module_params
with mock.patch.object(resource_module.TowerModule, '_load_params', new=mock_load_params):
# Call the test utility (like a mock server) instead of issuing HTTP requests
with mock.patch('tower_cli.api.Session.request', new=new_request):
# Ansible modules return data to the mothership over stdout
with redirect_stdout(stdout_buffer):
try:
resource_module.main()
except SystemExit:
pass # A system exit indicates successful execution
module_stdout = stdout_buffer.getvalue().strip()
result = json.loads(module_stdout)
return result
return rf
@pytest.fixture
def run_converted_module(request):
# A placeholder to use while modules get converted
def rf(module_name, module_params, request_user):

View File

@@ -63,9 +63,9 @@ def test_create_vault_credential(run_module, admin_user):
@pytest.mark.django_db
def test_create_custom_credential_type(run_converted_module, admin_user):
def test_create_custom_credential_type(run_module, admin_user):
# Example from docs
result = run_converted_module('tower_credential_type', dict(
result = run_module('tower_credential_type', dict(
name='Nexus',
description='Credentials type for Nexus',
kind='cloud',
@@ -78,12 +78,14 @@ def test_create_custom_credential_type(run_converted_module, admin_user):
ct = CredentialType.objects.get(name='Nexus')
result.pop('invocation')
result.pop('existing_credential_type')
result.pop('name')
assert result == {
"name": "Nexus",
"credential_type": "Nexus",
"state": "present",
"id": ct.pk,
"changed": True
"changed": True,
}
assert ct.inputs == {"fields": [{"id": "server", "type": "string", "default": "", "label": ""}], "required": []}
assert ct.injectors == {'extra_vars': {'nexus_credential': 'test'}}

View File

@@ -7,7 +7,7 @@ from awx.main.models import Organization
@pytest.mark.django_db
def test_create_organization(run_converted_module, admin_user):
def test_create_organization(run_module, admin_user):
module_args = {
'name': 'foo',
@@ -23,14 +23,16 @@ def test_create_organization(run_converted_module, admin_user):
'custom_virtualenv': None
}
result = run_converted_module('tower_organization', module_args, admin_user)
result = run_module('tower_organization', module_args, admin_user)
assert result.get('changed'), result
org = Organization.objects.get(name='foo')
result.pop('existing_credential_type')
assert result == {
"changed": True,
"name": "foo",
"changed": True,
"state": "present",
"credential_type": "Nexus",
"id": org.id,
"invocation": {
"module_args": module_args
@@ -41,10 +43,10 @@ def test_create_organization(run_converted_module, admin_user):
@pytest.mark.django_db
def test_create_organization_with_venv(run_converted_module, admin_user, mocker):
def test_create_organization_with_venv(run_module, admin_user, mocker):
path = '/var/lib/awx/venv/custom-venv/foobar13489435/'
with mocker.patch('awx.main.models.mixins.get_custom_venv_choices', return_value=[path]):
result = run_converted_module('tower_organization', {
result = run_module('tower_organization', {
'name': 'foo',
'custom_virtualenv': path,
'state': 'present'
@@ -53,8 +55,10 @@ def test_create_organization_with_venv(run_converted_module, admin_user, mocker)
org = Organization.objects.get(name='foo')
result.pop('invocation')
result.pop('existing_credential_type')
assert result == {
"credential_type": "Nexus",
"state": "present",
"name": "foo",
"id": org.id
}

View File

@@ -7,8 +7,8 @@ from awx.main.models import Project
@pytest.mark.django_db
def test_create_project(run_converted_module, admin_user, organization):
result = run_converted_module('tower_project', dict(
def test_create_project(run_module, admin_user, organization):
result = run_module('tower_project', dict(
name='foo',
organization=organization.name,
scm_type='git',
@@ -23,7 +23,10 @@ def test_create_project(run_converted_module, admin_user, organization):
assert proj.organization == organization
result.pop('invocation')
result.pop('existing_credential_type')
assert result == {
'credential_type': 'Nexus',
'state': 'present',
'name': 'foo',
'id': proj.id,
'warnings': warning

View File

@@ -7,10 +7,10 @@ from awx.main.models import Organization, Team
@pytest.mark.django_db
def test_create_team(run_converted_module, admin_user):
def test_create_team(run_module, admin_user):
org = Organization.objects.create(name='foo')
result = run_converted_module('tower_team', {
result = run_module('tower_team', {
'name': 'foo_team',
'description': 'fooin around',
'state': 'present',
@@ -20,9 +20,12 @@ def test_create_team(run_converted_module, admin_user):
team = Team.objects.filter(name='foo_team').first()
result.pop('invocation')
result.pop('existing_credential_type')
assert result == {
"changed": True,
"name": "foo_team",
"credential_type": "Nexus",
"state": "present",
"id": team.id if team else None,
}
team = Team.objects.get(name='foo_team')
@@ -31,7 +34,7 @@ def test_create_team(run_converted_module, admin_user):
@pytest.mark.django_db
def test_modify_team(run_converted_module, admin_user):
def test_modify_team(run_module, admin_user):
org = Organization.objects.create(name='foo')
team = Team.objects.create(
name='foo_team',
@@ -40,27 +43,35 @@ def test_modify_team(run_converted_module, admin_user):
)
assert team.description == 'flat foo'
result = run_converted_module('tower_team', {
result = run_module('tower_team', {
'name': 'foo_team',
'description': 'fooin around',
'organization': 'foo'
}, admin_user)
team.refresh_from_db()
result.pop('invocation')
result.pop('existing_credential_type')
assert result == {
"state": "present",
"changed": True,
"name": "foo_team",
"credential_type": "Nexus",
"id": team.id,
"changed": True
}
assert team.description == 'fooin around'
# 2nd modification, should cause no change
result = run_converted_module('tower_team', {
result = run_module('tower_team', {
'name': 'foo_team',
'description': 'fooin around',
'organization': 'foo'
}, admin_user)
result.pop('invocation')
result.pop('existing_credential_type')
assert result == {
"credential_type": "Nexus",
"name": "foo_team",
"id": team.id,
"state": "present",
"changed": False
}