remove conditional inventory sources POST

* Move logic from validator down to model
* Allow only 1 inventory source of type scm with
update_on_project_update set to True; for each inventory

awx/api/serializers.py

@@ -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):
 

awx/main/access.py

@@ -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 (
+        return self.check_related('inventory', Inventory, data)
-            self.check_related('inventory', Inventory, data) and
-            not InventorySource.objects.filter(
-                inventory=data.get('inventory'),
-                update_on_project_update=True, source='scm').exists())
 
     def can_delete(self, obj):
         if not self.user.is_superuser and \

awx/main/models/inventory.py

@@ -1389,6 +1389,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):
     '''

awx/main/tests/functional/api/test_inventory.py

@@ -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)
@@ -361,21 +373,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(
+        post(reverse('api:inventory_inventory_sources_list', kwargs={'pk': inventory.id}),
-            reverse('api:inventory_inventory_sources_list', kwargs={'pk': inventory.id}),
+             {'name': 'new inv src', 'source_project': project.pk, 'source': 'scm', 'overwrite_vars': True},
-            {'name': 'new inv src', 'source_project': project.pk, 'source': 'scm', 'overwrite_vars': True},
+             rando, expect=403)
-            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']
 
 
 @pytest.mark.django_db

awx/main/tests/functional/models/test_inventory.py

@@ -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)

awx/main/tests/unit/models/test_inventory.py

@@ -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()
+