External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-01-11 18:09:57 -03:30
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #127 from chrismeyersfsu/fix-6559

Browse Source 
remove conditional inventory sources POST
This commit is contained in:
Chris Meyers 2017-08-02 12:25:50 -04:00 committed by GitHub
commit 22aa251a24
6 changed files with 112 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
awx/api/serializers.py
View File

@ -1675,30 +1675,6 @@ class InventorySourceSerializer(UnifiedJobTemplateSerializer, InventorySourceOpt
raise serializers.ValidationError({"detail": _("Cannot create Inventory Source for Smart Inventory")})
return value
def validate(self, attrs):
def get_field_from_model_or_attrs(fd):
return attrs.get(fd, self.instance and getattr(self.instance, fd) or None)
update_on_launch = attrs.get('update_on_launch', self.instance and self.instance.update_on_launch)
update_on_project_update = get_field_from_model_or_attrs('update_on_project_update')
source = get_field_from_model_or_attrs('source')
overwrite_vars = get_field_from_model_or_attrs('overwrite_vars')
if attrs.get('source_path', None) and source!='scm':
raise serializers.ValidationError({"detail": _("Cannot set source_path if not SCM type.")})
elif update_on_launch and source=='scm' and update_on_project_update:
raise serializers.ValidationError({"detail": _(
"Cannot update SCM-based inventory source on launch if set to update on project update. "
"Instead, configure the corresponding source project to update on launch.")})
elif not self.instance and attrs.get('inventory', None) and InventorySource.objects.filter(
inventory=attrs.get('inventory', None), update_on_project_update=True, source='scm').exists():
raise serializers.ValidationError({"detail": _("Inventory controlled by project-following SCM.")})
elif source=='scm' and not overwrite_vars:
raise serializers.ValidationError({"detail": _(
"SCM type sources must set `overwrite_vars` to `true`.")})
return super(InventorySourceSerializer, self).validate(attrs)
class InventorySourceUpdateSerializer(InventorySourceSerializer):

6
awx/main/access.py
View File

@ -788,11 +788,7 @@ class InventorySourceAccess(BaseAccess):
if not self.check_related('source_project', Project, data, role_field='use_role'):
return False
# Checks for admin or change permission on inventory.
return (
self.check_related('inventory', Inventory, data) and
not InventorySource.objects.filter(
inventory=data.get('inventory'),
update_on_project_update=True, source='scm').exists())
return self.check_related('inventory', Inventory, data)
def can_delete(self, obj):
if not self.user.is_superuser and \

20
awx/main/models/inventory.py
View File

@ -1390,6 +1390,26 @@ class InventorySource(UnifiedJobTemplate, InventorySourceOptions):
raise ValidationError(_('Unable to configure this item for cloud sync. It is already managed by %s.') % s)
return source
def clean_update_on_project_update(self):
if self.update_on_project_update is True and \
self.source == 'scm' and \
InventorySource.objects.filter(
inventory=self.inventory,
update_on_project_update=True, source='scm').exists():
raise ValidationError(_("Cannot update SCM-based inventory source on launch if set to update on project update. "
"Instead, configure the corresponding source project to update on launch."))
return self.update_on_project_update
def clean_overwrite_vars(self):
if self.source == 'scm' and not self.overwrite_vars:
raise ValidationError(_("SCM type sources must set `overwrite_vars` to `true`."))
return self.overwrite_vars
def clean_source_path(self):
if self.source != 'scm' and self.source_path:
raise ValidationError(_("Cannot set source_path if not SCM type."))
return self.source_path
class InventoryUpdate(UnifiedJob, InventorySourceOptions, JobNotificationMixin, TaskManagerInventoryUpdateMixin):
'''

44
awx/main/tests/functional/api/test_inventory.py
View File

@ -19,6 +19,18 @@ def scm_inventory(inventory, project):
return inventory
@pytest.fixture
def factory_scm_inventory(inventory, project):
def fn(**kwargs):
with mock.patch('awx.main.models.unified_jobs.UnifiedJobTemplate.update'):
return inventory.inventory_sources.create(source_project=project,
overwrite_vars=True,
source='scm',
scm_last_revision=project.scm_revision,
**kwargs)
return fn
@pytest.mark.django_db
def test_inventory_source_notification_on_cloud_only(get, post, inventory_source_factory, user, notification_template):
u = user('admin', True)
@ -375,21 +387,31 @@ class TestControlledBySCM:
delete(inv_src.get_absolute_url(), admin_user, expect=204)
assert scm_inventory.inventory_sources.count() == 0
def test_adding_inv_src_prohibited(self, post, scm_inventory, admin_user):
def test_adding_inv_src_ok(self, post, scm_inventory, admin_user):
post(reverse('api:inventory_inventory_sources_list', kwargs={'version': 'v2', 'pk': scm_inventory.id}),
{'name': 'new inv src', 'update_on_project_update': False, 'source': 'scm', 'overwrite_vars': True},
admin_user, expect=201)
def test_adding_inv_src_prohibited(self, post, scm_inventory, project, admin_user):
post(reverse('api:inventory_inventory_sources_list', kwargs={'pk': scm_inventory.id}),
{'name': 'new inv src'}, admin_user, expect=403)
{'name': 'new inv src', 'source_project': project.pk, 'update_on_project_update': True, 'source': 'scm', 'overwrite_vars': True},
admin_user, expect=400)
def test_two_update_on_project_update_inv_src_prohibited(self, patch, scm_inventory, factory_scm_inventory, project, admin_user):
scm_inventory2 = factory_scm_inventory(name="scm_inventory2")
res = patch(reverse('api:inventory_source_detail', kwargs={'version': 'v2', 'pk': scm_inventory2.id}),
{'update_on_project_update': True,},
admin_user, expect=400)
content = json.loads(res.content)
assert content['update_on_project_update'] == ["Cannot update SCM-based inventory source on launch if set to update on "
"project update. Instead, configure the corresponding source project to "
"update on launch."]
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, 'source': 'scm', 'overwrite_vars': True},
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']
post(reverse('api:inventory_inventory_sources_list', kwargs={'pk': inventory.id}),
{'name': 'new inv src', 'source_project': project.pk, 'source': 'scm', 'overwrite_vars': True},
rando, expect=403)
@pytest.mark.django_db

19
awx/main/tests/functional/models/test_inventory.py
View File

@ -1,6 +1,8 @@
import pytest
import mock
from django.core.exceptions import ValidationError
# AWX
from awx.main.models import (
Host,
@ -47,6 +49,23 @@ class TestSCMUpdateFeatures:
assert not mck_update.called
@pytest.mark.django_db
class TestSCMClean:
def test_clean_update_on_project_update_multiple(self, inventory):
inv_src1 = InventorySource(inventory=inventory,
update_on_project_update=True,
source='scm')
inv_src1.clean_update_on_project_update()
inv_src1.save()
inv_src2 = InventorySource(inventory=inventory,
update_on_project_update=True,
source='scm')
with pytest.raises(ValidationError):
inv_src2.clean_update_on_project_update()
@pytest.fixture
def setup_ec2_gce(organization):
ec2_inv = Inventory.objects.create(name='test_ec2', organization=organization)

39
awx/main/tests/unit/models/test_inventory.py
View File

@ -10,6 +10,7 @@ from awx.main.models import (
Inventory,
Credential,
CredentialType,
InventorySource,
)
@ -69,3 +70,41 @@ def test_invalid_kind_clean_insights_credential():
inv.clean_insights_credential()
assert json.dumps(str(e.value)) == json.dumps(str([u'Assignment not allowed for Smart Inventory']))
class TestControlledBySCM():
@pytest.mark.parametrize('source', [
'scm',
'ec2',
'manual',
])
def test_clean_overwrite_vars_valid(self, source):
inv_src = InventorySource(overwrite_vars=True,
source=source)
inv_src.clean_overwrite_vars()
def test_clean_overwrite_vars_invalid(self):
inv_src = InventorySource(overwrite_vars=False,
source='scm')
with pytest.raises(ValidationError):
inv_src.clean_overwrite_vars()
def test_clean_source_path_valid(self):
inv_src = InventorySource(source_path='/not_real/',
source='scm')
inv_src.clean_source_path()
@pytest.mark.parametrize('source', [
'ec2',
'manual',
])
def test_clean_source_path_invalid(self, source):
inv_src = InventorySource(source_path='/not_real/',
source=source)
with pytest.raises(ValidationError):
inv_src.clean_source_path()