mirror of
https://github.com/ansible/awx.git
synced 2026-02-17 11:10:03 -03:30
6ebe45b1bd
* Add separate Django app for configuration: awx.conf. * Migrate from existing main.TowerSettings model to conf.Setting. * Add settings wrapper to allow get/set/del via django.conf.settings. * Update existing references to tower_settings to use django.conf.settings. * Add a settings registry to allow for each Django app to register configurable settings. * Support setting validation and conversion using Django REST Framework fields. * Add /api/v1/settings/ to display a list of setting categories. * Add /api/v1/settings/<slug>/ to display all settings in a category as a single object. * Allow PUT/PATCH to update setting singleton, DELETE to reset to defaults. * Add "all" category to display all settings across categories. * Add "changed" category to display only settings configured in the database. * Support per-user settings via "user" category (/api/v1/settings/user/). * Support defaults for user settings via "user-defaults" category (/api/v1/settings/user-defaults/). * Update serializer metadata to support category, category_slug and placeholder on OPTIONS responses. * Update serializer metadata to handle child fields of a list/dict. * Hide raw data form in browsable API for OPTIONS and DELETE. * Combine existing licensing code into single "TaskEnhancer" class. * Move license helper functions from awx.api.license into awx.conf.license. * Update /api/v1/config/ to read/verify/update license using TaskEnhancer and settings wrapper. * Add support for caching settings accessed via settings wrapper. * Invalidate cached settings when Setting model changes or is deleted. * Preload all database settings into cache on first access via settings wrapper. * Add support for read-only settings than can update their value depending on other settings. * Use setting_changed signal whenever a setting changes. * Register configurable authentication, jobs, system and ui settings. * Register configurable LDAP, RADIUS and social auth settings. * Add custom fields and validators for URL, LDAP, RADIUS and social auth settings. * Rewrite existing validator for Credential ssh_private_key to support validating private keys, certs or combinations of both. * Get all unit/functional tests working with above changes. * Add "migrate_to_database_settings" command to determine settings to be migrated into the database and comment them out when set in Python settings files. * Add support for migrating license key from file to database. * Remove database-configuable settings from local_settings.py example files. * Update setup role to no longer install files for database-configurable settings. f 94ff6ee More settings work. f af4c4e0 Even more db settings stuff. f 96ea9c0 More settings, attempt at singleton serializer for settings. f 937c760 More work on singleton/category views in API, add code to comment out settings in Python files, work on command to migrate settings to database. f 425b0d3 Minor fixes for sprint demo. f ea402a4 Add support for read-only settings, cleanup license engine, get license support working with DB settings. f ec289e4 Rename migration, minor fixmes, update setup role. f 603640b Rewrite key/cert validator, finish adding social auth fields, hook up signals for setting_changed, use None to imply a setting is not set. f 67d1b5a Get functional/unit tests passing. f 2919b62 Flake8 fixes. f e62f421 Add redbaron to requirements, get file to database migration working (except for license). f c564508 Add support for migrating license file. f 982f767 Add support for regex in social map fields.
259 lines
9.4 KiB
Python
259 lines
9.4 KiB
Python
# Python
|
|
import mock
|
|
import pytest
|
|
from datetime import timedelta
|
|
import urlparse
|
|
import urllib
|
|
|
|
# AWX
|
|
from awx.main.models.fact import Fact
|
|
from awx.main.utils import timestamp_apiformat
|
|
|
|
# Django
|
|
from django.core.urlresolvers import reverse
|
|
from django.utils import timezone
|
|
|
|
def mock_feature_enabled(feature):
|
|
return True
|
|
|
|
def mock_feature_disabled(feature):
|
|
return False
|
|
|
|
def setup_common(hosts, fact_scans, get, user, epoch=timezone.now(), get_params={}, host_count=1):
|
|
hosts = hosts(host_count=host_count)
|
|
fact_scans(fact_scans=3, timestamp_epoch=epoch)
|
|
|
|
url = reverse('api:host_fact_versions_list', args=(hosts[0].pk,))
|
|
response = get(url, user('admin', True), data=get_params)
|
|
|
|
return (hosts[0], response)
|
|
|
|
def check_url(url1_full, fact_known, module):
|
|
url1_split = urlparse.urlsplit(url1_full)
|
|
url1 = url1_split.path
|
|
url1_params = urlparse.parse_qsl(url1_split.query)
|
|
|
|
url2 = reverse('api:host_fact_compare_view', args=(fact_known.host.pk,))
|
|
url2_params = [('module', module), ('datetime', timestamp_apiformat(fact_known.timestamp))]
|
|
|
|
assert url1 == url2
|
|
# Sort before comparing because urlencode can't be trusted
|
|
url1_params_sorted = sorted(url1_params, key=lambda val: val[0])
|
|
url2_params_sorted = sorted(url2_params, key=lambda val: val[0])
|
|
assert urllib.urlencode(url1_params_sorted) == urllib.urlencode(url2_params_sorted)
|
|
|
|
def check_response_facts(facts_known, response):
|
|
for i, fact_known in enumerate(facts_known):
|
|
assert fact_known.module == response.data['results'][i]['module']
|
|
assert timestamp_apiformat(fact_known.timestamp) == response.data['results'][i]['timestamp']
|
|
check_url(response.data['results'][i]['related']['fact_view'], fact_known, fact_known.module)
|
|
|
|
def check_system_tracking_feature_forbidden(response):
|
|
assert 402 == response.status_code
|
|
assert 'Your license does not permit use of system tracking.' == response.data['detail']
|
|
|
|
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_disabled)
|
|
@pytest.mark.django_db
|
|
@pytest.mark.license_feature
|
|
def test_system_tracking_license_get(hosts, get, user):
|
|
hosts = hosts(host_count=1)
|
|
url = reverse('api:host_fact_versions_list', args=(hosts[0].pk,))
|
|
response = get(url, user('admin', True))
|
|
|
|
check_system_tracking_feature_forbidden(response)
|
|
|
|
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_disabled)
|
|
@pytest.mark.django_db
|
|
@pytest.mark.license_feature
|
|
def test_system_tracking_license_options(hosts, options, user):
|
|
hosts = hosts(host_count=1)
|
|
url = reverse('api:host_fact_versions_list', args=(hosts[0].pk,))
|
|
response = options(url, None, user('admin', True))
|
|
|
|
check_system_tracking_feature_forbidden(response)
|
|
|
|
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
|
@pytest.mark.django_db
|
|
@pytest.mark.license_feature
|
|
def test_no_facts_db(hosts, get, user):
|
|
hosts = hosts(host_count=1)
|
|
url = reverse('api:host_fact_versions_list', args=(hosts[0].pk,))
|
|
response = get(url, user('admin', True))
|
|
|
|
response_expected = {
|
|
'results': []
|
|
}
|
|
assert response_expected == response.data
|
|
|
|
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
|
@pytest.mark.django_db
|
|
def test_basic_fields(hosts, fact_scans, get, user):
|
|
epoch = timezone.now()
|
|
search = {
|
|
'from': epoch,
|
|
'to': epoch,
|
|
}
|
|
|
|
(host, response) = setup_common(hosts, fact_scans, get, user, epoch=epoch, get_params=search)
|
|
|
|
results = response.data['results']
|
|
assert 'related' in results[0]
|
|
assert 'timestamp' in results[0]
|
|
assert 'module' in results[0]
|
|
|
|
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
|
@pytest.mark.django_db
|
|
@pytest.mark.license_feature
|
|
def test_basic_options_fields(hosts, fact_scans, options, user):
|
|
hosts = hosts(host_count=1)
|
|
fact_scans(fact_scans=1)
|
|
|
|
url = reverse('api:host_fact_versions_list', args=(hosts[0].pk,))
|
|
response = options(url, None, user('admin', True), pk=hosts[0].id)
|
|
|
|
assert 'related' in response.data['actions']['GET']
|
|
assert 'module' in response.data['actions']['GET']
|
|
assert ("ansible", "Ansible") in response.data['actions']['GET']['module']['choices']
|
|
assert ("services", "Services") in response.data['actions']['GET']['module']['choices']
|
|
assert ("packages", "Packages") in response.data['actions']['GET']['module']['choices']
|
|
|
|
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
|
@pytest.mark.django_db
|
|
def test_related_fact_view(hosts, fact_scans, get, user):
|
|
epoch = timezone.now()
|
|
|
|
(host, response) = setup_common(hosts, fact_scans, get, user, epoch=epoch)
|
|
facts_known = Fact.get_timeline(host.id)
|
|
assert 9 == len(facts_known)
|
|
assert 9 == len(response.data['results'])
|
|
|
|
for i, fact_known in enumerate(facts_known):
|
|
check_url(response.data['results'][i]['related']['fact_view'], fact_known, fact_known.module)
|
|
|
|
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
|
@pytest.mark.django_db
|
|
def test_multiple_hosts(hosts, fact_scans, get, user):
|
|
epoch = timezone.now()
|
|
|
|
(host, response) = setup_common(hosts, fact_scans, get, user, epoch=epoch, host_count=3)
|
|
facts_known = Fact.get_timeline(host.id)
|
|
assert 9 == len(facts_known)
|
|
assert 9 == len(response.data['results'])
|
|
|
|
for i, fact_known in enumerate(facts_known):
|
|
check_url(response.data['results'][i]['related']['fact_view'], fact_known, fact_known.module)
|
|
|
|
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
|
@pytest.mark.django_db
|
|
def test_param_to_from(hosts, fact_scans, get, user):
|
|
epoch = timezone.now()
|
|
search = {
|
|
'from': epoch - timedelta(days=10),
|
|
'to': epoch + timedelta(days=10),
|
|
}
|
|
|
|
(host, response) = setup_common(hosts, fact_scans, get, user, epoch=epoch, get_params=search)
|
|
facts_known = Fact.get_timeline(host.id, ts_from=search['from'], ts_to=search['to'])
|
|
assert 9 == len(facts_known)
|
|
assert 9 == len(response.data['results'])
|
|
|
|
check_response_facts(facts_known, response)
|
|
|
|
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
|
@pytest.mark.django_db
|
|
def test_param_module(hosts, fact_scans, get, user):
|
|
epoch = timezone.now()
|
|
search = {
|
|
'module': 'packages',
|
|
}
|
|
|
|
(host, response) = setup_common(hosts, fact_scans, get, user, epoch=epoch, get_params=search)
|
|
facts_known = Fact.get_timeline(host.id, module=search['module'])
|
|
assert 3 == len(facts_known)
|
|
assert 3 == len(response.data['results'])
|
|
|
|
check_response_facts(facts_known, response)
|
|
|
|
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
|
@pytest.mark.django_db
|
|
def test_param_from(hosts, fact_scans, get, user):
|
|
epoch = timezone.now()
|
|
search = {
|
|
'from': epoch + timedelta(days=1),
|
|
}
|
|
|
|
(host, response) = setup_common(hosts, fact_scans, get, user, epoch=epoch, get_params=search)
|
|
facts_known = Fact.get_timeline(host.id, ts_from=search['from'])
|
|
assert 3 == len(facts_known)
|
|
assert 3 == len(response.data['results'])
|
|
|
|
check_response_facts(facts_known, response)
|
|
|
|
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
|
@pytest.mark.django_db
|
|
def test_param_to(hosts, fact_scans, get, user):
|
|
epoch = timezone.now()
|
|
search = {
|
|
'to': epoch + timedelta(days=1),
|
|
}
|
|
|
|
(host, response) = setup_common(hosts, fact_scans, get, user, epoch=epoch, get_params=search)
|
|
facts_known = Fact.get_timeline(host.id, ts_to=search['to'])
|
|
assert 6 == len(facts_known)
|
|
assert 6 == len(response.data['results'])
|
|
|
|
check_response_facts(facts_known, response)
|
|
|
|
def _test_user_access_control(hosts, fact_scans, get, user_obj, team_obj):
|
|
hosts = hosts(host_count=1)
|
|
fact_scans(fact_scans=1)
|
|
|
|
team_obj.member_role.members.add(user_obj)
|
|
|
|
url = reverse('api:host_fact_versions_list', args=(hosts[0].pk,))
|
|
response = get(url, user_obj)
|
|
return response
|
|
|
|
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
|
@pytest.mark.ac
|
|
@pytest.mark.django_db
|
|
def test_normal_user_403(hosts, fact_scans, get, user, team):
|
|
user_bob = user('bob', False)
|
|
response = _test_user_access_control(hosts, fact_scans, get, user_bob, team)
|
|
|
|
assert 403 == response.status_code
|
|
assert "You do not have permission to perform this action." == response.data['detail']
|
|
|
|
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
|
@pytest.mark.ac
|
|
@pytest.mark.django_db
|
|
def test_super_user_ok(hosts, fact_scans, get, user, team):
|
|
user_super = user('bob', True)
|
|
response = _test_user_access_control(hosts, fact_scans, get, user_super, team)
|
|
|
|
assert 200 == response.status_code
|
|
|
|
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
|
@pytest.mark.ac
|
|
@pytest.mark.django_db
|
|
def test_user_admin_ok(organization, hosts, fact_scans, get, user, team):
|
|
user_admin = user('johnson', False)
|
|
organization.admin_role.members.add(user_admin)
|
|
|
|
response = _test_user_access_control(hosts, fact_scans, get, user_admin, team)
|
|
|
|
assert 200 == response.status_code
|
|
|
|
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
|
@pytest.mark.ac
|
|
@pytest.mark.django_db
|
|
def test_user_admin_403(organization, organizations, hosts, fact_scans, get, user, team):
|
|
user_admin = user('johnson', False)
|
|
org2 = organizations(1)
|
|
org2[0].admin_role.members.add(user_admin)
|
|
|
|
response = _test_user_access_control(hosts, fact_scans, get, user_admin, team)
|
|
|
|
assert 403 == response.status_code
|
|
|