Dev to Stage

.github/workflows/unit_tests.yaml

@@ -0,0 +1,72 @@
+name: Unit Tests
+on:
+  workflow_dispatch:
+  push:
+    branches-ignore:
+      - '**'
+  pull_request:
+    branches: [ main, dev, stage ]
+
+jobs:
+  UnitTest:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v2
+
+      - name: Set up Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: '3.10'
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+
+      - name: Determine output folder
+        id: set_output_folder
+        run: |
+          if [[ $GITHUB_EVENT_NAME == "pull_request" ]]; then
+            branch_name=$GITHUB_BASE_REF
+          else
+            branch_name=$GITHUB_REF_NAME
+          fi
+
+          if [[ $branch_name == "main" ]]; then
+            echo "output_folder=prod" >> $GITHUB_ENV
+          elif [[ $branch_name == "stage" ]]; then
+            echo "output_folder=stage" >> $GITHUB_ENV
+          elif [[ $branch_name == "dev" ]]; then
+            echo "output_folder=dev" >> $GITHUB_ENV
+          else
+            echo "Unknown branch: $branch_name"
+            exit 1
+          fi
+
+      - name: Run tests with coverage
+        run: |
+          timestamp=$(date '+%Y-%m-%d_%H-%M-%S')
+          mkdir -p test_results
+          log_file="test_results/${timestamp}_report.log"
+          echo -e "\nTest Cases Report Report\n" >> $log_file
+          # Run the tests and append output to the log file
+          PYTHONPATH=$(pwd) python -m coverage run --source=src -m unittest discover -v tests/ >> $log_file 2>&1
+          echo -e "\nCoverage Report\n" >> $log_file
+          coverage report >> $log_file
+          coverage xml  
+
+      - name: Check coverage
+        run: |
+          coverage report --fail-under=85
+
+      - name: Upload report to Azure
+        uses: LanceMcCarthy/Action-AzureBlobUpload@v2
+        with:
+          source_folder: 'test_results'
+          destination_folder: '${{ env.output_folder }}'
+          connection_string: ${{ secrets.AZURE_STORAGE_CONNECTION_STRING }}
+          container_name: 'osw-quality-metric-service'
+          clean_destination_folder: false
+          delete_if_exists: false

.gitignore

@@ -162,3 +162,5 @@ cython_debug/
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
 .ve
+MagicMock
+test_results

src/config.py

@@ -12,14 +12,14 @@ class Config(BaseSettings):
     incoming_topic_subscription: str = os.environ.get('QUALITY_REQ_SUB', '')
     outgoing_topic_name: str = os.environ.get('QUALITY_RES_TOPIC', '')
     storage_container_name: str = os.environ.get('CONTAINER_NAME', 'osw')
-    algorithm_dictionary: dict = {"fixed":QMFixedCalculator,"ixn":QMXNLibCalculator}
+    algorithm_dictionary: dict = {"fixed": QMFixedCalculator, "ixn": QMXNLibCalculator}
     max_concurrent_messages: int = os.environ.get('MAX_CONCURRENT_MESSAGES', 1)
-    partition_count:int = os.environ.get('PARTITION_COUNT', 2)
+    partition_count: int = os.environ.get('PARTITION_COUNT', 2)
 
     def get_download_folder(self) -> str:
         root_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
         return os.path.join(root_dir, 'downloads')
-    
+
     def get_assets_folder(self) -> str:
         root_dir = os.path.dirname(os.path.abspath(__file__))
-        return os.path.join(root_dir, 'assets')
+        return os.path.join(root_dir, 'assets')

src/services/servicebus_service.py

@@ -45,6 +45,7 @@ def __init__(self) -> None:
 
     def process_message(self, msg: QueueMessage):
         logger.info(f"Processing message {msg}")
+        input_file_url = None
         try:
             logger.info(f"Processing message {msg.messageId}")
             # Parse the message

tests/calculators/test_qm_calculator.py

@@ -0,0 +1,49 @@
+import unittest
+from abc import ABC, abstractmethod
+from src.calculators import QMCalculator
+from src.calculators.qm_calculator import QualityMetricResult
+
+# A concrete implementation for testing purposes
+class MockQMCalculator(QMCalculator):
+    def calculate_quality_metric(self) -> QualityMetricResult:
+        return QualityMetricResult(success=True, message='Calculation succeeded.', output_file='output.geojson')
+
+    def algorithm_name(self) -> str:
+        return 'mock-algorithm'
+
+class TestQualityMetricResult(unittest.TestCase):
+    def test_quality_metric_result_initialization(self):
+        # Create a QualityMetricResult instance
+        result = QualityMetricResult(
+            success=True,
+            message='Calculation succeeded.',
+            output_file='output.geojson'
+        )
+
+        # Assertions
+        self.assertTrue(result.success)
+        self.assertEqual(result.message, 'Calculation succeeded.')
+        self.assertEqual(result.output_file, 'output.geojson')
+
+class TestQMCalculator(unittest.TestCase):
+    def test_qm_calculator_is_abstract(self):
+        # Ensure QMCalculator cannot be instantiated directly
+        with self.assertRaises(TypeError):
+            QMCalculator()
+
+    def test_concrete_implementation(self):
+        # Instantiate the mock implementation
+        calculator = MockQMCalculator()
+
+        # Test calculate_quality_metric
+        result = calculator.calculate_quality_metric()
+        self.assertIsInstance(result, QualityMetricResult)
+        self.assertTrue(result.success)
+        self.assertEqual(result.message, 'Calculation succeeded.')
+        self.assertEqual(result.output_file, 'output.geojson')
+
+        # Test algorithm_name
+        self.assertEqual(calculator.algorithm_name(), 'mock-algorithm')
+
+if __name__ == '__main__':
+    unittest.main()

tests/calculators/test_qm_fixed_calculator.py

@@ -0,0 +1,130 @@
+import json
+import tempfile
+import unittest
+import geopandas as gpd
+from subprocess import run, PIPE
+from unittest.mock import patch, MagicMock
+from src.calculators.qm_fixed_calculator import QMFixedCalculator
+from src.calculators.qm_calculator import QualityMetricResult
+
+
+class TestQMFixedCalculator(unittest.TestCase):
+
+    def setUp(self):
+        self.edges_file_path = 'test_edges.geojson'
+        self.output_file_path = 'test_output.geojson'
+        self.polygon_file_path = 'test_polygon.geojson'
+        self.calculator = QMFixedCalculator(
+            self.edges_file_path,
+            self.output_file_path,
+            self.polygon_file_path
+        )
+
+    def test_initialization(self):
+        self.assertEqual(self.calculator.edges_file_path, self.edges_file_path)
+        self.assertEqual(self.calculator.output_file_path, self.output_file_path)
+        self.assertEqual(self.calculator.polygon_file_path, self.polygon_file_path)
+
+    @patch('src.calculators.qm_fixed_calculator.gpd.read_file')
+    @patch('src.calculators.qm_fixed_calculator.gpd.GeoDataFrame.to_file', autospec=True)
+    @patch('src.calculators.qm_fixed_calculator.random.randint')
+    def test_calculate_quality_metric(self, mock_randint, mock_to_file, mock_read_file):
+        # Setup mocks
+        mock_randint.return_value = 42
+        mock_gdf = MagicMock(spec=gpd.GeoDataFrame)
+        mock_read_file.return_value = mock_gdf
+
+        # Call the method under test
+        result = self.calculator.calculate_quality_metric()
+
+        # Assertions
+        mock_read_file.assert_called_once_with(self.edges_file_path)
+        mock_gdf.__setitem__.assert_called_once_with('fixed_score', 42)
+        mock_gdf.to_file.assert_called_once_with(self.output_file_path)
+        self.assertIsInstance(result, QualityMetricResult)
+        self.assertTrue(result.success)
+        self.assertEqual(result.message, 'QMFixedCalculator')
+        self.assertEqual(result.output_file, self.output_file_path)
+
+    def test_algorithm_name(self):
+        self.assertEqual(self.calculator.algorithm_name(), 'QMFixedCalculator')
+
+    def test_main_with_polygon(self):
+        # Create temporary files for edges and polygon
+        with tempfile.NamedTemporaryFile(suffix='.geojson') as edges_file, \
+                tempfile.NamedTemporaryFile(suffix='.geojson') as output_file, \
+                tempfile.NamedTemporaryFile(suffix='.geojson') as polygon_file:
+            # Write dummy GeoJSON data to edges and polygon files
+            dummy_geojson = {
+                'type': 'FeatureCollection',
+                'features': [
+                    {
+                        'type': 'Feature',
+                        'geometry': {'type': 'LineString', 'coordinates': [[0, 0], [1, 1]]},
+                        'properties': {}
+                    }
+                ]
+            }
+
+            edges_file.write(json.dumps(dummy_geojson).encode())
+            edges_file.flush()
+
+            polygon_file.write(json.dumps(dummy_geojson).encode())
+            polygon_file.flush()
+
+            # Simulate running the script as a standalone process
+            result = run(
+                [
+                    'python',
+                    'src/calculators/qm_fixed_calculator.py',
+                    edges_file.name,
+                    output_file.name,
+                    polygon_file.name,
+                ],
+                stdout=PIPE,
+                stderr=PIPE,
+                text=True,
+            )
+
+            # Debugging: Print stderr if test fails
+            if result.returncode != 0:
+                print('Error:', result.stderr)
+
+            # Assertions
+            self.assertEqual(result.returncode, 0)
+            self.assertIn('QMFixedCalculator', result.stdout)
+
+    def test_main_without_polygon(self):
+        # Create temporary files for edges
+        with tempfile.NamedTemporaryFile(suffix='.geojson') as edges_file, \
+                tempfile.NamedTemporaryFile(suffix='.geojson') as output_file:
+            # Write dummy GeoJSON data to edges file
+            dummy_geojson = {
+                'type': 'FeatureCollection',
+                'features': []
+            }
+
+            edges_file.write(json.dumps(dummy_geojson).encode())
+            edges_file.flush()
+
+            # Simulate running the script as a standalone process
+            result = run(
+                [
+                    'python',
+                    'src/calculators/qm_fixed_calculator.py',
+                    edges_file.name,
+                    output_file.name,
+                ],
+                stdout=PIPE,
+                stderr=PIPE,
+                text=True,
+            )
+
+            # Assertions
+            self.assertEqual(result.returncode, 0)
+            self.assertIn('QMFixedCalculator', result.stdout)
+
+
+
+if __name__ == '__main__':
+    unittest.main()