add api for testing credential plugins

This commit is contained in:
Jake McDermott
2019-02-25 09:33:43 -05:00
parent 7a43f00a5d
commit 0ee223f799
14 changed files with 224 additions and 35 deletions

View File

@@ -13,6 +13,7 @@ from awx.api.views import (
CredentialOwnerTeamsList,
CredentialCopy,
CredentialInputSourceSubList,
CredentialExternalTest,
)
@@ -26,6 +27,7 @@ urls = [
url(r'^(?P<pk>[0-9]+)/owner_teams/$', CredentialOwnerTeamsList.as_view(), name='credential_owner_teams_list'),
url(r'^(?P<pk>[0-9]+)/copy/$', CredentialCopy.as_view(), name='credential_copy'),
url(r'^(?P<pk>[0-9]+)/input_sources/$', CredentialInputSourceSubList.as_view(), name='credential_input_source_sublist'),
url(r'^(?P<pk>[0-9]+)/test/$', CredentialExternalTest.as_view(), name='credential_external_test'),
]
__all__ = ['urls']

View File

@@ -8,6 +8,7 @@ from awx.api.views import (
CredentialTypeDetail,
CredentialTypeCredentialList,
CredentialTypeActivityStreamList,
CredentialTypeExternalTest,
)
@@ -16,6 +17,7 @@ urls = [
url(r'^(?P<pk>[0-9]+)/$', CredentialTypeDetail.as_view(), name='credential_type_detail'),
url(r'^(?P<pk>[0-9]+)/credentials/$', CredentialTypeCredentialList.as_view(), name='credential_type_credential_list'),
url(r'^(?P<pk>[0-9]+)/activity_stream/$', CredentialTypeActivityStreamList.as_view(), name='credential_type_activity_stream_list'),
url(r'^(?P<pk>[0-9]+)/test/$', CredentialTypeExternalTest.as_view(), name='credential_type_external_test'),
]
__all__ = ['urls']

View File

@@ -1419,6 +1419,32 @@ class CredentialCopy(CopyAPIView):
copy_return_serializer_class = serializers.CredentialSerializer
class CredentialExternalTest(SubDetailAPIView):
"""
Test updates to the input values of an external credential before
saving them.
"""
view_name = _('External Credential Test')
model = models.Credential
serializer_class = serializers.EmptySerializer
def post(self, request, *args, **kwargs):
obj = self.get_object()
test_inputs = {}
for field_name, value in request.data.get('inputs', {}).items():
if value == '$encrypted$':
test_inputs[field_name] = obj.get_input(field_name)
else:
test_inputs[field_name] = value
try:
obj.credential_type.plugin.backend(None, **test_inputs)
return Response({}, status=status.HTTP_202_ACCEPTED)
except Exception as exc:
return Response({'inputs': str(exc)}, status=status.HTTP_400_BAD_REQUEST)
class CredentialInputSourceDetail(RetrieveUpdateDestroyAPIView):
view_name = _("Credential Input Source Detail")
@@ -1446,6 +1472,27 @@ class CredentialInputSourceSubList(SubListCreateAttachDetachAPIView):
parent_key = 'target_credential'
class CredentialTypeExternalTest(SubDetailAPIView):
"""
Test a complete set of input values for an external credential before
saving it.
"""
view_name = _('External Credential Type Test')
model = models.CredentialType
serializer_class = serializers.EmptySerializer
def post(self, request, *args, **kwargs):
obj = self.get_object()
test_inputs = request.data.get('inputs', {})
try:
obj.plugin.backend(None, **test_inputs)
return Response({}, status=status.HTTP_202_ACCEPTED)
except Exception as exc:
return Response({'inputs': str(exc)}, status=status.HTTP_400_BAD_REQUEST)
class HostRelatedSearchMixin(object):
@property