Ors client

.github/workflows/ci.yml

@@ -43,13 +43,15 @@ jobs:
       - name: Preparing docker-compose environment
         env:
           QGIS_VERSION_TAG: ${{ matrix.qgis_version_tag }}
+          ORS_API_KEY: ${{ secrets.ORS_API_KEY }}
         run: |
           cat << EOF > .env
           QGIS_VERSION_TAG=${QGIS_VERSION_TAG}
           IMAGE=${IMAGE}
           ON_TRAVIS=true
           MUTE_LOGS=${MUTE_LOGS}
           WITH_PYTHON_PEP=${WITH_PYTHON_PEP}
+          ORS_API_KEY=${ORS_API_KEY}
           EOF
       - name: Install python
         uses: actions/setup-python@v4
@@ -61,7 +63,6 @@ jobs:
 
       - name: Preparing test environment
         run: |
-          cat .env
           docker pull "${IMAGE}":${{ matrix.qgis_version_tag }}
           python admin.py build --tests
           docker compose up -d

docker-compose.yml

@@ -9,6 +9,7 @@ services:
       ON_TRAVIS: "${ON_TRAVIS}"
       MUTE_LOGS: "${MUTE_LOGS}"
       DISPLAY: ":99"
+      ORS_API_KEY: "${ORS_API_KEY}"
     working_dir: /tests_directory
     entrypoint: /tests_directory/scripts/docker/qgis-testing-entrypoint.sh
     # Enable "command:" line below to immediately run unittests upon docker-compose up

geest/core/ors_client.py

@@ -0,0 +1,62 @@
+import os
+from PyQt5.QtCore import QUrl, QByteArray
+from PyQt5.QtNetwork import QNetworkRequest
+from qgis.core import QgsMessageLog, QgsNetworkAccessManager
+import json
+
+
+class ORSClient:
+    def __init__(self, base_url):
+        self.base_url = base_url
+        self.network_manager = QgsNetworkAccessManager.instance()
+        self.api_key = os.getenv("ORS_API_KEY")
+
+        # Ensure the API key is available
+        if not self.api_key:
+            raise EnvironmentError(
+                "ORS API key is missing. Set it in the environment variable 'ORS_API_KEY'."
+            )
+
+    def make_request(self, endpoint, params):
+        url = f"{self.base_url}/{endpoint}"
+        request = QNetworkRequest(QUrl(url))
+
+        # Set necessary headers for the ORS API
+        request.setHeader(QNetworkRequest.ContentTypeHeader, "application/json")
+        if self.api_key:
+            request.setRawHeader(b"Authorization", self.api_key.encode())
+        else:
+            QgsMessageLog.logMessage(
+                "API Key is missing", "ORS", QgsMessageLog.CRITICAL
+            )
+            return
+
+        # Convert parameters (Python dict) to JSON
+        data = QByteArray(json.dumps(params).encode("utf-8"))
+
+        # Send the request and connect to the response handler
+        reply = self.network_manager.post(request, data)
+        reply.finished.connect(lambda: self.handle_response(reply))
+
+    def handle_response(self, reply):
+        if reply.error() == reply.NoError:
+            response_data = reply.readAll().data().decode()
+            try:
+                # Parse the JSON response
+                response_json = json.loads(response_data)
+                QgsMessageLog.logMessage(f"ORS Response: {response_json}", "ORS")
+                return (
+                    response_json  # Return the parsed response for further processing
+                )
+            except json.JSONDecodeError as e:
+                QgsMessageLog.logMessage(
+                    f"Failed to decode JSON: {e}", "ORS", QgsMessageLog.CRITICAL
+                )
+                return None  # Return None in case of failure
+        else:
+            # Handle error
+            QgsMessageLog.logMessage(
+                f"Error: {reply.errorString()}", "ORS", QgsMessageLog.CRITICAL
+            )
+            return None  # Return None in case of error
+        reply.deleteLater()

geest/gui/tree_panel.py

@@ -28,7 +28,6 @@
 from geest.core.workflow_queue_manager import WorkflowQueueManager
 
 
-
 class TreePanel(QWidget):
     def __init__(self, parent=None, json_file=None):
         super().__init__(parent)

test/test_ors_client.py

@@ -0,0 +1,94 @@
+import os
+import unittest
+import json
+from geest.core.ors_client import ORSClient
+
+
+class TestORSClientRealRequest(unittest.TestCase):
+    def setUp(self):
+        """Ensure the API key is available in the environment before running the test."""
+        self.api_key = os.getenv("ORS_API_KEY")
+        self.assertIsNotNone(
+            self.api_key,
+            "API key is missing. Please set the ORS_API_KEY environment variable.",
+        )
+
+        # Instantiate the ORSClient
+        self.ors = ORSClient("https://api.openrouteservice.org/v2/isochrones")
+
+        # Prepare parameters (for isochrones request)
+        self.params = {
+            "locations": [[8.681495, 49.41461], [8.687872, 49.420318]],
+            "range": [300, 200],
+        }
+
+    def test_make_request(self):
+        """Test the ORSClient with an API request and assert the response is correct."""
+        try:
+            # Make the actual request to the ORS API
+            self.ors.make_request("walking", self.params)
+
+            # Ensure the API key is used and the endpoint is correctly reached
+            def check_response(reply):
+                # Check if there was an error with the request
+                self.assertEqual(
+                    reply.error(),
+                    reply.NoError,
+                    f"Error in network request: {reply.errorString()}",
+                )
+
+                # Parse the response and validate the data structure
+                response_data = reply.readAll().data().decode()
+                response_json = json.loads(response_data)
+
+                # Check that the response contains a 'features' key
+                self.assertIn(
+                    "features",
+                    response_json,
+                    "Response does not contain 'features' key",
+                )
+                self.assertIsInstance(
+                    response_json["features"], list, "'features' is not a list"
+                )
+
+                # Ensure at least one feature is returned
+                self.assertGreater(
+                    len(response_json["features"]),
+                    0,
+                    "No features found in the response",
+                )
+
+                # Check that the first feature has the expected structure
+                first_feature = response_json["features"][0]
+                self.assertIn(
+                    "geometry", first_feature, "Feature does not contain 'geometry' key"
+                )
+                self.assertEqual(
+                    first_feature["geometry"]["type"],
+                    "Polygon",
+                    "Expected geometry type 'Polygon'",
+                )
+                self.assertIn(
+                    "coordinates",
+                    first_feature["geometry"],
+                    "'geometry' does not contain 'coordinates' key",
+                )
+
+                # Ensure the 'coordinates' field is not empty
+                self.assertGreater(
+                    len(first_feature["geometry"]["coordinates"]),
+                    0,
+                    "'coordinates' list is empty",
+                )
+
+            # Ensure the response handler gets the data and processes it correctly
+            reply = self.ors.network_manager.finished.connect(
+                lambda: check_response(reply)
+            )
+
+        except Exception as e:
+            self.fail(f"Test failed due to unexpected exception: {e}")
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_polygons_per_grid_cell.py

@@ -29,9 +29,7 @@ def test_raster_polygon_grid_score(self):
         )
         self.country_boundary = os.path.join(self.test_data_dir, "admin/Admin0.shp")
 
-        self.assertTrue(
-            self.polygon_layer.isValid(), "The polygon layer is not valid."
-        )
+        self.assertTrue(self.polygon_layer.isValid(), "The polygon layer is not valid.")
 
         # Define output path for the generated raster
         self.output_path = os.path.join(