api caller should work

pygeoweaver/api_call/pgw_base_api_caller.py

@@ -1,12 +1,30 @@
+import json
+from typing import Any, Dict, Optional
 import requests
 
+GEOWEAVER_PREFIX="/Geoweaver/web"
 
 class BaseAPI:
     def __init__(self, base_url):
         self.base_url = base_url
 
-    def _call_api(self, endpoint, method='GET', data=None):
-        url = self.base_url + endpoint
+    def _call_api(self, endpoint: str, method: str = 'GET', 
+                  data: Optional[Dict[str, Any]] = None) -> Optional[Dict[str, Any]]:
+        """
+        Sends an HTTP request to the specified API endpoint.
+
+        Parameters:
+        - endpoint (str): The API endpoint to send the request to.
+        - method (str): The HTTP method to use ('GET', 'POST', 'PUT', 'DELETE'). Defaults to 'GET'.
+        - data (Optional[Dict[str, Any]]): The data to send with the request. For 'POST' and 'PUT', it should be a dictionary.
+
+        Returns:
+        - Optional[Dict[str, Any]]: The JSON response from the server, parsed into a dictionary. Returns None if an error occurs.
+
+        Raises:
+        - ValueError: If an unsupported HTTP method is specified.
+        """
+        url = self.base_url + GEOWEAVER_PREFIX + endpoint
         headers = {
             'Content-Type': 'application/json'
         }
@@ -24,8 +42,11 @@ def _call_api(self, endpoint, method='GET', data=None):
                 raise ValueError(f"Unsupported HTTP method: {method}")
 
             response.raise_for_status()
+            print("return is ", response)
             return response.json()
         except requests.exceptions.HTTPError as http_err:
             print(f"HTTP error occurred: {http_err}")
+            return None
         except Exception as err:
             print(f"An error occurred: {err}")
+            return None

pygeoweaver/api_call/pyw_workflow_api_caller.py

@@ -1,17 +1,53 @@
 
+from typing import Any, Dict
 from pygeoweaver.api_call.pgw_base_api_caller import BaseAPI
 
 
 class GeoweaverWorkflowAPI(BaseAPI):
-    def add_workflow(self, workflow_data):
+    def add_workflow(self, workflow_data: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Adds a new workflow by sending a POST request to the API.
+
+        Parameters:
+        - workflow_data (Dict[str, Any]): The data of the workflow to add. This should be a dictionary representing the workflow.
+
+        Returns:
+        - Dict[str, Any]: The JSON response from the server, parsed into a dictionary.
+        """
         return self._call_api('/add/workflow', method='POST', data=workflow_data)
 
-    def edit_workflow(self, workflow_data):
+    def edit_workflow(self, workflow_data: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Edits an existing workflow by sending a POST request to the API.
+
+        Parameters:
+        - workflow_data (Dict[str, Any]): The data of the workflow to edit. This should be a dictionary representing the workflow with updates.
+
+        Returns:
+        - Dict[str, Any]: The JSON response from the server, parsed into a dictionary.
+        """
         return self._call_api('/edit/workflow', method='POST', data=workflow_data)
 
-    def get_workflow(self, workflow_id):
+    def get_workflow(self, workflow_id: str) -> Dict[str, Any]:
+        """
+        Retrieves a workflow by its ID by sending a GET request to the API.
+
+        Parameters:
+        - workflow_id (str): The ID of the workflow to retrieve.
+
+        Returns:
+        - Dict[str, Any]: The JSON response from the server, parsed into a dictionary.
+        """
         return self._call_api(f'/get/workflow/{workflow_id}', method='GET')
 
-    def delete_workflow(self, workflow_id):
-        return self._call_api(f'/delete/workflow/{workflow_id}', method='DELETE')
+    def delete_workflow(self, workflow_id: str) -> Dict[str, Any]:
+        """
+        Deletes a workflow by its ID by sending a DELETE request to the API.
 
+        Parameters:
+        - workflow_id (str): The ID of the workflow to delete.
+
+        Returns:
+        - Dict[str, Any]: The JSON response from the server, parsed into a dictionary.
+        """
+        return self._call_api(f'/delete/workflow/{workflow_id}', method='DELETE')

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "pygeoweaver"
-version = "1.0.9"
+version = "1.2.0"
 authors = [{ name="Geoweaver team", email="[email protected]" }]
 description = "This is a wrapper package of the Geoweaver app."
 readme = "README.md"

test/test_api_caller.py

@@ -0,0 +1,68 @@
+from pygeoweaver.server import start, stop
+import pytest
+from pygeoweaver.api_call.pgw_base_api_caller import BaseAPI
+from pygeoweaver.api_call.pgw_process_api_caller import GeoweaverProcessAPI  # Adjust the import path as needed
+
+# Define the base URL for the API
+BASE_URL = "http://localhost:8070"  # Update with your actual base URL
+
+@pytest.fixture(scope="session")
+def start_geoweaver():
+    # Start the Geoweaver server
+    start(exit_on_finish=False)
+
+    yield
+
+    # Stop the Geoweaver server
+    stop(exit_on_finish=False)
+
+@pytest.fixture
+def geoweaver_api():
+    return GeoweaverProcessAPI(base_url=BASE_URL)
+
+@pytest.fixture
+def create_process(geoweaver_api):
+
+    # Example process data
+    process_data = {
+        "name": "Test Process",
+        "description": "A test process for pytest",
+        "parameters": {"param1": "value1"}
+    }
+    response = geoweaver_api.add_process(process_data)
+    print(response)
+    assert response.status_code == 201, "Failed to create process"
+    process_id = response.json().get("id")
+    yield process_id, process_data
+    # Cleanup: Ensure process is deleted after test
+    geoweaver_api.delete_process(process_id)
+    # Verify the process is deleted
+    response = geoweaver_api.get_process(process_id)
+    assert response.status_code == 404, "Process should be deleted"
+
+def test_edit_process(geoweaver_api, create_process):
+    process_id, process_data = create_process
+    updated_data = {
+        "name": "Updated Test Process",
+        "description": "An updated test process",
+        "parameters": {"param1": "new_value"}
+    }
+    response = geoweaver_api.edit_process({**updated_data, "id": process_id})
+    assert response.status_code == 200, "Failed to edit process"
+    assert response.json().get("id") == process_id
+
+def test_get_process(geoweaver_api, create_process):
+    process_id, _ = create_process
+    response = geoweaver_api.get_process(process_id)
+    assert response.status_code == 200, "Failed to get process"
+    process = response.json()
+    assert process.get("id") == process_id
+
+def test_delete_process(geoweaver_api, create_process):
+    process_id, _ = create_process
+    response = geoweaver_api.delete_process(process_id)
+    assert response.status_code == 200, "Failed to delete process"
+
+    # Verify the process is deleted
+    response = geoweaver_api.get_process(process_id)
+    assert response.status_code == 404, "Process should be deleted"