SCM inventory acceptance and bug updates

* fixed a problem where the SCM last revision was not getting reset
* change the 400 editing error to just reset the last revision
* handle `source_vars` in the same way as custom scripts
* block unallowed `source_vars` in the validator
* hide POST in OPTIONS for inventory source sublist if not allowed
This commit is contained in:
AlanCoding
2017-05-11 10:07:14 -04:00
parent 55c2f5a2d6
commit 97ca6bc897
10 changed files with 136 additions and 45 deletions

View File

@@ -1,7 +1,10 @@
import pytest
import mock
from awx.api.versioning import reverse
from awx.main.models import InventorySource
@pytest.mark.django_db
def test_inventory_source_notification_on_cloud_only(get, post, inventory_source_factory, user, notification_template):
@@ -199,11 +202,23 @@ def test_inventory_source_update(post, inventory_source, alice, role_field, expe
post(reverse('api:inventory_source_update_view', kwargs={'pk': inventory_source.id}), {}, alice, expect=expected_status_code)
@pytest.mark.django_db
def test_inventory_source_vars_prohibition(post, inventory, admin_user):
with mock.patch('awx.api.serializers.settings') as mock_settings:
mock_settings.INV_ENV_VARIABLE_BLACKLIST = ('FOOBAR',)
r = post(reverse('api:inventory_source_list'),
{'name': 'new inv src', 'source_vars': '{\"FOOBAR\": \"val\"}', 'inventory': inventory.pk},
admin_user, expect=400)
assert 'prohibited environment variable' in r.data['source_vars'][0]
assert 'FOOBAR' in r.data['source_vars'][0]
@pytest.fixture
def scm_inventory(inventory, project):
inventory.inventory_sources.create(
name='foobar', update_on_project_update=True, source='scm',
source_project=project)
with mock.patch.object(project, 'update'):
inventory.inventory_sources.create(
name='foobar', update_on_project_update=True, source='scm',
source_project=project, scm_last_revision=project.scm_revision)
return inventory
@@ -216,37 +231,43 @@ class TestControlledBySCM:
def test_safe_method_works(self, get, options, scm_inventory, admin_user):
get(scm_inventory.get_absolute_url(), admin_user, expect=200)
options(scm_inventory.get_absolute_url(), admin_user, expect=200)
assert InventorySource.objects.get(inventory=scm_inventory.pk).scm_last_revision != ''
def test_vars_edit_prohibited(self, patch, scm_inventory, admin_user):
def test_vars_edit_reset(self, patch, scm_inventory, admin_user):
patch(scm_inventory.get_absolute_url(), {'variables': 'hello: world'},
admin_user, expect=403)
admin_user, expect=200)
assert InventorySource.objects.get(inventory=scm_inventory.pk).scm_last_revision == ''
def test_name_edit_allowed(self, patch, scm_inventory, admin_user):
patch(scm_inventory.get_absolute_url(), {'variables': '---', 'name': 'newname'},
admin_user, expect=200)
assert InventorySource.objects.get(inventory=scm_inventory.pk).scm_last_revision != ''
def test_host_associations_prohibited(self, post, scm_inventory, admin_user):
def test_host_associations_reset(self, post, scm_inventory, admin_user):
inv_src = scm_inventory.inventory_sources.first()
h = inv_src.hosts.create(name='barfoo', inventory=scm_inventory)
g = inv_src.groups.create(name='fooland', inventory=scm_inventory)
post(reverse('api:host_groups_list', kwargs={'pk': h.id}), {'id': g.id},
admin_user, expect=403)
admin_user, expect=204)
post(reverse('api:group_hosts_list', kwargs={'pk': g.id}), {'id': h.id},
admin_user, expect=403)
admin_user, expect=204)
assert InventorySource.objects.get(inventory=scm_inventory.pk).scm_last_revision == ''
def test_group_group_associations_prohibited(self, post, scm_inventory, admin_user):
def test_group_group_associations_reset(self, post, scm_inventory, admin_user):
inv_src = scm_inventory.inventory_sources.first()
g1 = inv_src.groups.create(name='barland', inventory=scm_inventory)
g2 = inv_src.groups.create(name='fooland', inventory=scm_inventory)
post(reverse('api:group_children_list', kwargs={'pk': g1.id}), {'id': g2.id},
admin_user, expect=403)
admin_user, expect=204)
assert InventorySource.objects.get(inventory=scm_inventory.pk).scm_last_revision == ''
def test_host_group_delete_prohibited(self, delete, scm_inventory, admin_user):
def test_host_group_delete_reset(self, delete, scm_inventory, admin_user):
inv_src = scm_inventory.inventory_sources.first()
h = inv_src.hosts.create(name='barfoo', inventory=scm_inventory)
g = inv_src.groups.create(name='fooland', inventory=scm_inventory)
delete(h.get_absolute_url(), admin_user, expect=403)
delete(g.get_absolute_url(), admin_user, expect=403)
delete(h.get_absolute_url(), admin_user, expect=204)
delete(g.get_absolute_url(), admin_user, expect=204)
assert InventorySource.objects.get(inventory=scm_inventory.pk).scm_last_revision == ''
def test_remove_scm_inv_src(self, delete, scm_inventory, admin_user):
inv_src = scm_inventory.inventory_sources.first()
@@ -255,4 +276,14 @@ class TestControlledBySCM:
def test_adding_inv_src_prohibited(self, post, scm_inventory, admin_user):
post(reverse('api:inventory_inventory_sources_list', kwargs={'pk': scm_inventory.id}),
{'name': 'new inv src'}, admin_user, expect=400)
{'name': 'new inv src'}, admin_user, expect=403)
def test_adding_inv_src_without_proj_access_prohibited(self, post, project, inventory, rando):
inventory.admin_role.members.add(rando)
post(reverse('api:inventory_inventory_sources_list', kwargs={'pk': inventory.id}),
{'name': 'new inv src', 'source_project': project.pk}, rando, expect=403)
def test_no_post_in_options(self, options, scm_inventory, admin_user):
r = options(reverse('api:inventory_inventory_sources_list', kwargs={'pk': scm_inventory.id}),
admin_user, expect=200)
assert 'POST' not in r.data['actions']

View File

@@ -241,13 +241,17 @@ def inventory(organization):
@pytest.fixture
def scm_inventory_source(inventory, project):
return InventorySource.objects.create(
inv_src = InventorySource(
name="test-scm-inv",
source_project=project,
source='scm',
source_path='inventory_file',
update_on_project_update=True,
inventory=inventory)
inventory=inventory,
scm_last_revision=project.scm_revision)
with mock.patch.object(inv_src.source_project, 'update'):
inv_src.save()
return inv_src
@pytest.fixture

View File

@@ -23,6 +23,13 @@ class TestSCMUpdateFeatures:
inv_src.save()
mck_update.assert_called_once_with()
def test_reset_scm_revision(self, scm_inventory_source):
starting_rev = scm_inventory_source.scm_last_revision
assert starting_rev != ''
scm_inventory_source.source_path = '/newfolder/newfile.ini'
scm_inventory_source.save()
assert scm_inventory_source.scm_last_revision == ''
def test_source_location(self, scm_inventory_source):
# Combines project directory with the inventory file specified
inventory_update = InventoryUpdate(

View File

@@ -36,6 +36,7 @@ class TestDependentInventoryUpdate:
def test_dependent_inventory_updates(self, scm_inventory_source):
task = RunProjectUpdate()
scm_inventory_source.scm_last_revision = ''
proj_update = ProjectUpdate.objects.create(project=scm_inventory_source.source_project)
with mock.patch.object(RunInventoryUpdate, 'run') as iu_run_mock:
task._update_dependent_inventories(proj_update, [scm_inventory_source])