Merge pull request #181 from mantiumai/AlexN/api-endpoint-asset-search

Implement APIEndpointAsset's fetch_api_data functionality

chirps/asset/providers/api_endpoint.py

@@ -1,9 +1,12 @@
 """Logic for interfacing with an API Endpoint Asset."""
+import json
 from logging import getLogger
 
-from asset.models import BaseAsset, PingResult, SearchResult
+import requests
+from asset.models import BaseAsset, PingResult
 from django.db import models
 from fernet_fields import EncryptedCharField
+from requests import RequestException, Timeout
 
 logger = getLogger(__name__)
 
@@ -38,9 +41,34 @@ def decrypted_api_key(self):
                 return 'Error: Decryption failed'
         return None
 
-    def search(self, query: str, max_results: int) -> list[SearchResult]:
-        """Search the API Endpoint asset with the specified query."""
-        raise NotImplementedError('The search method is not implemented for this asset.')
+    def fetch_api_data(self, query: str) -> dict:
+        """Fetch data from the API using the provided query."""
+        # Convert headers JSON string into a dictionary
+        headers_dict = json.loads(self.headers) if self.headers else {}
+
+        # Build the request headers
+        headers = headers_dict.copy()
+        if self.authentication_method == 'Bearer':
+            headers['Authorization'] = f'Bearer {self.api_key}'
+        elif self.authentication_method == 'Basic':
+            headers['Authorization'] = f'Basic {self.api_key}'
+
+        # Replace the %query% placeholder in the body
+        body = json.loads(json.dumps(self.body).replace('%query%', query))
+
+        # Send the request
+        try:
+            response = requests.post(self.url, headers=headers, json=body, timeout=15)
+        except Timeout as exc:
+            raise RequestException('Error: API request timed out') from exc
+
+        # Check if the request was successful
+        if response.status_code != 200:
+            raise RequestException(f'Error: API request failed with status code {response.status_code}')
+
+        # Parse the response and return the search results
+        response_data = response.json()
+        return response_data
 
     def test_connection(self) -> PingResult:
         """Ensure that the API Endpoint asset can be connected to."""

chirps/asset/tests.py

@@ -233,3 +233,64 @@ def test_ping__failure(self):
             asset = RedisAsset(host='localhost', port=12000)
             result = asset.test_connection()
             assert result.success is False
+
+
+class APIEndpointAssetTests(TestCase):
+    """Test the APIEndpointAsset."""
+
+    def setUp(self):
+        """Set up the APIEndpointAsset tests."""
+        # Login the user before performing any tests
+        self.client.post(reverse('login'), {'username': 'admin', 'password': 'admin'})
+
+        # Create an APIEndpointAsset instance for testing
+        self.api_endpoint_asset = APIEndpointAsset.objects.create(
+            name='Test API Endpoint Asset',
+            url='https://api.example.com/endpoint',
+            authentication_method='Bearer',
+            api_key='example-api-key',
+            headers='{"Content-Type": "application/json"}',
+            body='{"data": "%query%"}',
+        )
+
+    def test_fetch_api_data(self):
+        """Test that the search method sends the request and processes the response."""
+        # Define the mocked response data
+        mock_response_data = {
+            'chat': {
+                'instance': '46045911',
+                'application': '1077587932992295905',
+                'conversation': '4022404441860560655',
+                'speak': 'true',
+                'avatarFormat': 'webm',
+                'secure': 'true',
+                'message': 'hello',
+            }
+        }
+
+        # Mock the requests.post method to return the mocked response data
+        with mock.patch('requests.post') as mock_post:
+            mock_post.return_value.status_code = 200
+            mock_post.return_value.json.return_value = mock_response_data
+
+            # Call the search method and assert the results
+            search_results = self.api_endpoint_asset.fetch_api_data('test query')
+            self.assertEqual(len(search_results), 1)
+
+            # Assert that requests.post was called with the expected arguments
+            expected_url = self.api_endpoint_asset.url
+            expected_headers = json.loads(self.api_endpoint_asset.headers)
+            expected_headers['Authorization'] = f'Bearer {self.api_endpoint_asset.api_key}'
+            expected_body = self.api_endpoint_asset.body.replace('%query%', 'test query')
+
+            mock_post.assert_called_once_with(expected_url, headers=expected_headers, json=expected_body, timeout=15)
+
+            # Assert the search result attributes
+            result = search_results['chat']
+            self.assertEqual(result['instance'], '46045911')
+            self.assertEqual(result['application'], '1077587932992295905')
+            self.assertEqual(result['conversation'], '4022404441860560655')
+            self.assertEqual(result['speak'], 'true')
+            self.assertEqual(result['avatarFormat'], 'webm')
+            self.assertEqual(result['secure'], 'true')
+            self.assertEqual(result['message'], 'hello')