.gitea/workflows/hub-test.yaml

@@ -0,0 +1,21 @@
+name: Hub component testing
+on: [push, workflow_dispatch]
+
+jobs:
+  hub-test:
+    runs-on: host-arch-x86_64
+    steps:
+      - name: Clone repository
+        run: git clone --revision ${{ gitea.sha }} --depth 1 ${{ gitea.server_url }}/${{ gitea.repository }}
+
+      - name: Build Hub testing container
+        working-directory: IoT-Systems
+        run: docker build -t local/hub/${{gitea.sha}} -f hub/Dockerfile-test .
+
+      - name: Run Hub tests
+        working-directory: IoT-Systems
+        run: docker run --rm -it local/hub/${{gitea.sha}}
+
+      - name: Clean up containers
+        if: ${{always()}}
+        run: docker image rm local/hub/${{gitea.sha}}

hub/Dockerfile-test

@@ -0,0 +1,12 @@
+# Use the official Python image as the base image
+FROM python:3.9-slim
+# Set the working directory inside the container
+WORKDIR /app
+# Copy the requirements.txt file and install dependencies
+COPY hub/requirements.txt .
+
+RUN pip install --no-cache-dir -r requirements.txt
+# Copy the entire application into the container
+COPY hub/. .
+# Run the main.py script inside the container when it starts
+CMD ["./test-entry.sh"]

hub/app/adapters/store_api_adapter.py

@@ -13,7 +13,7 @@ class StoreApiAdapter(StoreGateway):
     def __init__(self, api_base_url):
         self.api_base_url = api_base_url
 
-    def save_data(self, processed_agent_data_batch: List[ProcessedAgentData]):
+    def processed_agent_data_batch_to_json(self, processed_agent_data_batch: List[ProcessedAgentData]):
         if not processed_agent_data_batch:
             return False
 
@@ -25,6 +25,14 @@ class StoreApiAdapter(StoreGateway):
             "user_id": user_id
         }
 
+        return payload
+
+    def save_data(self, processed_agent_data_batch: List[ProcessedAgentData]):
+        payload = self.processed_agent_data_batch_to_json(processed_agent_data_batch)
+
+        if payload == False:
+            return False
+
         try:
             # Perform a POST request to the Store API with a 10-second timeout
             response = requests.post(

hub/app/adapters/store_api_adapter_test.py

@@ -0,0 +1,39 @@
+from app.adapters.store_api_adapter import StoreApiAdapter
+from app.entities.agent_data import AccelerometerData, AgentData, GpsData
+from app.entities.processed_agent_data import ProcessedAgentData
+
+def _test_processed_agent_data_batch_to_json():
+    processed_data_batch = [
+            ProcessedAgentData(road_state = "normal",
+                               agent_data = AgentData(user_id = 1,
+                                                      accelerometer = AccelerometerData(x = 0.1, y = 0.2, z = 0.3),
+                                                      gps = GpsData(latitude = 10.123, longitude = 20.456),
+                                                      timestamp = "2023-07-21T12:34:56Z")
+            ),
+            ProcessedAgentData(road_state = "normal",
+                               agent_data = AgentData(user_id = 2,
+                                                      accelerometer = AccelerometerData(x = 0.1, y = 0.2, z = 0.3),
+                                                      gps = GpsData(latitude = 10.123, longitude = 20.456),
+                                                      timestamp = "2023-07-21T12:34:56Z")
+            ),
+            ProcessedAgentData(road_state = "normal",
+                               agent_data = AgentData(user_id = 3,
+                                                      accelerometer = AccelerometerData(x = 0.1, y = 0.2, z = 0.3),
+                                                      gps = GpsData(latitude = 10.123, longitude = 20.456),
+                                                      timestamp = "2023-07-21T12:34:56Z")
+            ),
+    ]
+
+    res = StoreApiAdapter(None).processed_agent_data_batch_to_json(processed_data_batch)
+
+    assert res["data"][0]["agent_data"]["user_id"] == 1
+    assert res["data"][1]["agent_data"]["user_id"] == 2
+    assert res["data"][2]["agent_data"]["user_id"] == 3
+
+
+if __name__ == "__main__":
+    test_functions = [i for i in dir() if i.startswith('_test_')]
+
+    for i in test_functions:
+        print(i)
+        eval(i)()

hub/test-entry.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+
+PYTHONPATH=$PWD python3 app/adapters/store_api_adapter_test.py