enable tests and create test workflow

.github/workflows/tests.yml

@@ -0,0 +1,39 @@
+name: Run Tests
+
+on:
+  push:
+    branches:
+      - '*'
+  pull_request:
+    branches:
+      - '*'
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v3
+
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: '3.9' # specify the Python version you need
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+          pip install pytest
+
+      - name: Run tests
+        run: |
+          pytest tests/
+
+      - name: Upload pytest results
+        if: always()
+        uses: actions/upload-artifact@v3
+        with:
+          name: pytest-results
+          path: tests/results/

pygeoweaver/api_call/pgw_base_api_caller.py

@@ -0,0 +1,31 @@
+import requests
+
+
+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
+        headers = {
+            'Content-Type': 'application/json'
+        }
+
+        try:
+            if method.upper() == 'POST':
+                response = requests.post(url, headers=headers, data=json.dumps(data))
+            elif method.upper() == 'GET':
+                response = requests.get(url, headers=headers, params=data)
+            elif method.upper() == 'PUT':
+                response = requests.put(url, headers=headers, data=json.dumps(data))
+            elif method.upper() == 'DELETE':
+                response = requests.delete(url, headers=headers, data=json.dumps(data))
+            else:
+                raise ValueError(f"Unsupported HTTP method: {method}")
+
+            response.raise_for_status()
+            return response.json()
+        except requests.exceptions.HTTPError as http_err:
+            print(f"HTTP error occurred: {http_err}")
+        except Exception as err:
+            print(f"An error occurred: {err}")

pygeoweaver/api_call/pgw_process_api_caller.py

@@ -0,0 +1,19 @@
+
+import requests
+
+from pygeoweaver.api_call.pgw_base_api_caller import BaseAPI
+
+class GeoweaverProcessAPI(BaseAPI):
+
+    def edit_process(self, process_data):
+        return self._call_api('/edit/process', method='POST', data=process_data)
+
+    def add_process(self, process_data):
+        return self._call_api('/add/process', method='POST', data=process_data)
+
+    def get_process(self, process_id):
+        return self._call_api(f'/get/process/{process_id}', method='GET')
+
+    def delete_process(self, process_id):
+        return self._call_api(f'/delete/process/{process_id}', method='DELETE')
+

pygeoweaver/api_call/pyw_workflow_api_caller.py

@@ -0,0 +1,17 @@
+
+from pygeoweaver.api_call.pgw_base_api_caller import BaseAPI
+
+
+class GeoweaverWorkflowAPI(BaseAPI):
+    def add_workflow(self, workflow_data):
+        return self._call_api('/add/workflow', method='POST', data=workflow_data)
+
+    def edit_workflow(self, workflow_data):
+        return self._call_api('/edit/workflow', method='POST', data=workflow_data)
+
+    def get_workflow(self, workflow_id):
+        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')
+

pygeoweaver/server.py

@@ -4,6 +4,7 @@
 import sys
 import webbrowser
 import time
+import psutil
 import requests
 from halo import Halo
 
@@ -191,7 +192,32 @@ def stop_on_mac_linux(exit_on_finish: bool=False) -> int:
     with get_spinner(text=f'Stopping Geoweaver...', spinner='dots'):
         # Stop running Geoweaver if any
         logger.info("Stop running Geoweaver if any..")
-        subprocess.run(["pkill", "-f", "geoweaver.jar"])
+        # Find all processes running geoweaver.jar
+        processes = []
+        for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
+            # logger.info(proc)
+
+            if proc and proc.info and proc.info['cmdline'] and 'geoweaver.jar' in proc.info['cmdline']:
+                processes.append(proc)
+
+        if not processes:
+            print("No running Geoweaver processes found.")
+            return 0
+
+        # Attempt to kill each process
+        errors = []
+        for proc in processes:
+            try:
+                proc.terminate()
+                proc.wait(timeout=5)  # Wait for the process to terminate
+            except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.TimeoutExpired) as e:
+                errors.append(f"Failed to kill process {proc.info['pid']}: {e}")
+
+        if errors:
+            for error in errors:
+                logger.error(error)
+            print("Some processes could not be stopped.")
+            return 1
 
         # Check status
         status = subprocess.run(["curl", "-s", "-o", "/dev/null", 

pyproject.toml

@@ -23,6 +23,7 @@ dependencies = [
     "pandas",
     "halo",
     "tabulate",
+    "psutil",
 ]
 
 [project.urls]

test/test_server.py

@@ -33,19 +33,19 @@ def test_server_start_stop(self):
         stop(exit_on_finish=False)  # stop again should have no issue
 
     def test_windows(self):
-        with patch("pygeoweaver.server.checkOS") as mock_checkos:
+        with patch("pygeoweaver.server.check_os") as mock_checkos:
             mock_checkos.return_value = 3
-            with self.assertRaises(RuntimeError):
+            with self.assertRaises(FileNotFoundError):
                 start(exit_on_finish=False)
-            with self.assertRaises(RuntimeError):
+            with self.assertRaises(FileNotFoundError):
                 stop(exit_on_finish=False)
 
     def test_show_gui(self):
         with patch("pygeoweaver.webbrowser.open") as mock_browser_open:
             show()
             mock_browser_open.assert_called_once()
 
-            with patch("pygeoweaver.server.checkIPython") as mock_checkipython:
+            with patch("pygeoweaver.server.check_ipython") as mock_checkipython:
                 mock_checkipython.return_value = True
                 show()
                 mock_browser_open.assert_called_once()