Merge pull request #1389 from anoek/credential-api

Added and updated several credential creation and listing API endpoints

awx/main/access.py

@@ -542,20 +542,12 @@ class CredentialAccess(BaseAccess):
         qs = self.model.accessible_objects(self.user, {'read':True})
         return qs.select_related('created_by', 'modified_by').all()
 
+    def can_read(self, obj):
+        return obj.accessible_by(self.user, {'read': True})
+
     def can_add(self, data):
-        if self.user.is_superuser:
-            return True
-
-        if 'user' in data:
-            pk = get_pk_from_dict(data, 'user')
-            user = get_object_or_400(User, pk=pk)
-            return user.accessible_by(self.user, {'write': True})
-        elif 'organization' in data:
-            pk = get_pk_from_dict(data, 'organization')
-            org = get_object_or_400(Organization, pk=pk)
-            return org.accessible_by(self.user, {'write': True})
-
-        return False
+        # Access enforced in our view where we have context enough to make a decision
+        return True
 
     def can_change(self, obj, data):
         if self.user.is_superuser:

awx/main/models/mixins.py

@@ -43,6 +43,10 @@ class ResourceMixin(models.Model):
             qs = cls.objects.filter(
                 role_permissions__role__ancestors__members=accessor
             )
+        elif type(accessor) == Role:
+            qs = cls.objects.filter(
+                role_permissions__role__ancestors=accessor
+            )
         else:
             accessor_type = ContentType.objects.get_for_model(accessor)
             roles = Role.objects.filter(content_type__pk=accessor_type.id,

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

@@ -0,0 +1,223 @@
+import mock # noqa
+import pytest
+
+from django.core.urlresolvers import reverse
+
+
+#
+# user credential creation
+#
+
+@pytest.mark.django_db
+def test_create_user_credential_via_credentials_list(post, get, alice):
+    response = post(reverse('api:credential_list'), {
+        'user': alice.id,
+        'name': 'Some name',
+        'username': 'someusername'
+    }, alice)
+    assert response.status_code == 201
+
+    response = get(reverse('api:credential_list'), alice)
+    assert response.status_code == 200
+    assert response.data['count'] == 1
+
+@pytest.mark.django_db
+def test_create_user_credential_via_user_credentials_list(post, get, alice):
+    response = post(reverse('api:user_credentials_list', args=(alice.pk,)), {
+        'name': 'Some name',
+        'username': 'someusername',
+    }, alice)
+    assert response.status_code == 201
+
+    response = get(reverse('api:user_credentials_list', args=(alice.pk,)), alice)
+    assert response.status_code == 200
+    assert response.data['count'] == 1
+
+@pytest.mark.django_db
+def test_create_user_credential_via_credentials_list_xfail(post, alice, bob):
+    response = post(reverse('api:credential_list'), {
+        'user': bob.id,
+        'name': 'Some name',
+        'username': 'someusername'
+    }, alice)
+    assert response.status_code == 403
+
+@pytest.mark.django_db
+def test_create_user_credential_via_user_credentials_list_xfail(post, alice, bob):
+    response = post(reverse('api:user_credentials_list', args=(bob.pk,)), {
+        'name': 'Some name',
+        'username': 'someusername'
+    }, alice)
+    assert response.status_code == 403
+
+
+#
+# team credential creation
+#
+
+@pytest.mark.django_db
+def test_create_team_credential(post, get, team, org_admin, team_member):
+    response = post(reverse('api:credential_list'), {
+        'team': team.id,
+        'name': 'Some name',
+        'username': 'someusername'
+    }, org_admin)
+    assert response.status_code == 201
+
+    response = get(reverse('api:team_credentials_list', args=(team.pk,)), team_member)
+    assert response.status_code == 200
+    assert response.data['count'] == 1
+
+@pytest.mark.django_db
+def test_create_team_credential_via_team_credentials_list(post, get, team, org_admin, team_member):
+    response = post(reverse('api:team_credentials_list', args=(team.pk,)), {
+        'name': 'Some name',
+        'username': 'someusername',
+    }, org_admin)
+    assert response.status_code == 201
+
+    response = get(reverse('api:team_credentials_list', args=(team.pk,)), team_member)
+    assert response.status_code == 200
+    assert response.data['count'] == 1
+
+@pytest.mark.django_db
+def test_create_team_credential_by_urelated_user_xfail(post, team, alice, team_member):
+    response = post(reverse('api:credential_list'), {
+        'team': team.id,
+        'name': 'Some name',
+        'username': 'someusername'
+    }, alice)
+    assert response.status_code == 403
+
+@pytest.mark.django_db
+def test_create_team_credential_by_team_member_xfail(post, team, alice, team_member):
+    # Members can't add credentials, only org admins.. for now?
+    response = post(reverse('api:credential_list'), {
+        'team': team.id,
+        'name': 'Some name',
+        'username': 'someusername'
+    }, team_member)
+    assert response.status_code == 403
+
+
+
+#
+# organization credentials
+#
+
+@pytest.mark.django_db
+def test_create_org_credential_as_not_admin(post, organization, org_member):
+    response = post(reverse('api:credential_list'), {
+        'name': 'Some name',
+        'username': 'someusername',
+        'organization': organization.id,
+    }, org_member)
+    assert response.status_code == 403
+
+@pytest.mark.django_db
+def test_create_org_credential_as_admin(post, organization, org_admin):
+    response = post(reverse('api:credential_list'), {
+        'name': 'Some name',
+        'username': 'someusername',
+        'organization': organization.id,
+    }, org_admin)
+    assert response.status_code == 201
+
+@pytest.mark.django_db
+def test_list_created_org_credentials(post, get, organization, org_admin, org_member):
+    response = post(reverse('api:credential_list'), {
+        'name': 'Some name',
+        'username': 'someusername',
+        'organization': organization.id,
+    }, org_admin)
+    assert response.status_code == 201
+
+    response = get(reverse('api:credential_list'), org_admin)
+    assert response.status_code == 200
+    assert response.data['count'] == 1
+
+    response = get(reverse('api:credential_list'), org_member)
+    assert response.status_code == 200
+    assert response.data['count'] == 0
+
+    response = get(reverse('api:organization_credential_list', args=(organization.pk,)), org_admin)
+    assert response.status_code == 200
+    assert response.data['count'] == 1
+
+    response = get(reverse('api:organization_credential_list', args=(organization.pk,)), org_member)
+    assert response.status_code == 200
+    assert response.data['count'] == 0
+
+
+
+#
+# Openstack Credentials
+#
+
+@pytest.mark.django_db
+def test_openstack_create_ok(post, organization, admin):
+    data = {
+        'kind': 'openstack',
+        'name': 'Best credential ever',
+        'username': 'some_user',
+        'password': 'some_password',
+        'project': 'some_project',
+        'host': 'some_host',
+        'organization': organization.id,
+    }
+    response = post(reverse('api:credential_list'), data, admin)
+    assert response.status_code == 201
+
+@pytest.mark.django_db
+def test_openstack_create_fail_required_fields(post, organization, admin):
+    data = {
+        'kind': 'openstack',
+        'name': 'Best credential ever',
+        'organization': organization.id,
+    }
+    response = post(reverse('api:credential_list'), data, admin)
+    assert response.status_code == 400
+    assert 'username' in response.data
+    assert 'password' in response.data
+    assert 'host' in response.data
+    assert 'project' in response.data
+
+
+#
+# misc xfail conditions
+#
+
+@pytest.mark.django_db
+def test_create_credential_xfails(post, organization, team, admin):
+    # Must specify one of user, team, or organization
+    response = post(reverse('api:credential_list'), {
+        'name': 'Some name',
+        'username': 'someusername',
+    }, admin)
+    assert response.status_code == 400
+    # Can only specify one of user, team, or organization
+    response = post(reverse('api:credential_list'), {
+        'name': 'Some name',
+        'username': 'someusername',
+        'user': admin.id,
+        'organization': organization.id,
+    }, admin)
+    assert response.status_code == 400
+    response = post(reverse('api:credential_list'), {
+        'name': 'Some name',
+        'username': 'someusername',
+        'organization': organization.id,
+        'team': team.id,
+    }, admin)
+    assert response.status_code == 400
+    response = post(reverse('api:credential_list'), {
+        'name': 'Some name',
+        'username': 'someusername',
+        'user': admin.id,
+        'team': team.id,
+    }, admin)
+    assert response.status_code == 400
+
+
+
+

awx/main/tests/functional/conftest.py

@@ -92,6 +92,13 @@ def deploy_jobtemplate(project, inventory, credential):
 def team(organization):
     return organization.teams.create(name='test-team')
 
+@pytest.fixture
+def team_member(user, team):
+    ret = user('team-member', False)
+    team.member_role.members.add(ret)
+    return ret
+
+
 @pytest.fixture
 @mock.patch.object(Project, "update", lambda self, **kwargs: None)
 def project(instance, organization):

awx/main/tests/old/inventory.py

@@ -2003,34 +2003,3 @@ class InventoryUpdatesTest(BaseTransactionTest):
         self.check_inventory_source(inventory_source)
         self.assertFalse(self.group.all_hosts.filter(instance_id='').exists())
 
-
-class InventoryCredentialTest(BaseTest):
-    def setUp(self):
-        super(InventoryCredentialTest, self).setUp()
-        #self.start_redis()
-        self.setup_instances()
-        self.setup_users()
-
-        self.url = reverse('api:credential_list')
-
-    def test_openstack_create_ok(self):
-        data = {
-            'kind': 'openstack',
-            'name': 'Best credential ever',
-            'username': 'some_user',
-            'password': 'some_password',
-            'project': 'some_project',
-            'host': 'some_host',
-        }
-        self.post(self.url, data=data, expect=201, auth=self.get_super_credentials())
-
-    def test_openstack_create_fail_required_fields(self):
-        data = {
-            'kind': 'openstack',
-            'name': 'Best credential ever',
-        }
-        response = self.post(self.url, data=data, expect=400, auth=self.get_super_credentials())
-        self.assertIn('username', response)
-        self.assertIn('password', response)
-        self.assertIn('host', response)
-        self.assertIn('project', response)