From 5cd41dca73515ac8926b8883b90b3f9f8353078f Mon Sep 17 00:00:00 2001 From: Ziheng Sun Date: Sun, 21 Jul 2024 03:46:46 -0400 Subject: [PATCH] api caller should work --- pygeoweaver/api_call/pgw_base_api_caller.py | 25 ++++++- .../api_call/pyw_workflow_api_caller.py | 46 +++++++++++-- pyproject.toml | 2 +- test/test_api_caller.py | 68 +++++++++++++++++++ 4 files changed, 133 insertions(+), 8 deletions(-) create mode 100644 test/test_api_caller.py diff --git a/pygeoweaver/api_call/pgw_base_api_caller.py b/pygeoweaver/api_call/pgw_base_api_caller.py index 8183b7d..a3982da 100644 --- a/pygeoweaver/api_call/pgw_base_api_caller.py +++ b/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 \ No newline at end of file diff --git a/pygeoweaver/api_call/pyw_workflow_api_caller.py b/pygeoweaver/api_call/pyw_workflow_api_caller.py index cdf6882..c5247e2 100644 --- a/pygeoweaver/api_call/pyw_workflow_api_caller.py +++ b/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') diff --git a/pyproject.toml b/pyproject.toml index 1d61bb2..aaeba90 100644 --- a/pyproject.toml +++ b/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="geoweaver.app@gmail.com" }] description = "This is a wrapper package of the Geoweaver app." readme = "README.md" diff --git a/test/test_api_caller.py b/test/test_api_caller.py new file mode 100644 index 0000000..ddf01b1 --- /dev/null +++ b/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"