refactor: tests for mapillary processing and street project

Author: Gigaszi

mapswipe_workers/tests/unittests/test_process_mapillary.py

@@ -1,16 +1,30 @@
 import unittest
 import os
 import json
-from shapely.geometry import Polygon, MultiPolygon, Point, LineString, MultiLineString, GeometryCollection
+from shapely.geometry import (
+    Polygon,
+    MultiPolygon,
+    Point,
+    LineString,
+    MultiLineString,
+    GeometryCollection,
+)
 import pandas as pd
 from unittest.mock import patch, MagicMock
-from mapswipe_workers.utils.process_mapillary import create_tiles, download_and_process_tile, coordinate_download, geojson_to_polygon, filter_by_timerange, filter_results
+from mapswipe_workers.utils.process_mapillary import (
+    create_tiles,
+    download_and_process_tile,
+    coordinate_download,
+    geojson_to_polygon,
+    filter_by_timerange,
+    filter_results,
+)
 
 
 # Assuming create_tiles, download_and_process_tile, and coordinate_download are imported
 
-class TestTileGroupingFunctions(unittest.TestCase):
 
+class TestTileGroupingFunctions(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         with open(
@@ -39,7 +53,9 @@ def setUp(self):
         self.token = "test_token"
         self.level = 14
         self.test_polygon = Polygon([(0, 0), (1, 0), (1, 1), (0, 1)])
-        self.test_multipolygon = MultiPolygon([self.test_polygon, Polygon([(2, 2), (3, 2), (3, 3), (2, 3)])])
+        self.test_multipolygon = MultiPolygon(
+            [self.test_polygon, Polygon([(2, 2), (3, 2), (3, 3), (2, 3)])]
+        )
         self.empty_polygon = Polygon()
         self.empty_geometry = GeometryCollection()
 
@@ -67,9 +83,21 @@ def test_geojson_to_polygon_feature_collection_with_multiple_polygons(self):
         geojson_data = {
             "type": "FeatureCollection",
             "features": [
-                {"type": "Feature", "geometry": {"type": "Polygon", "coordinates": [[(0, 0), (1, 0), (1, 1), (0, 1), (0, 0)]]}},
-                {"type": "Feature", "geometry": {"type": "Polygon", "coordinates": [[(2, 2), (3, 2), (3, 3), (2, 3), (2, 2)]]}}
-            ]
+                {
+                    "type": "Feature",
+                    "geometry": {
+                        "type": "Polygon",
+                        "coordinates": [[(0, 0), (1, 0), (1, 1), (0, 1), (0, 0)]],
+                    },
+                },
+                {
+                    "type": "Feature",
+                    "geometry": {
+                        "type": "Polygon",
+                        "coordinates": [[(2, 2), (3, 2), (3, 3), (2, 3), (2, 2)]],
+                    },
+                },
+            ],
         }
         result = geojson_to_polygon(geojson_data)
         self.assertIsInstance(result, MultiPolygon)
@@ -80,8 +108,8 @@ def test_geojson_to_polygon_single_feature_polygon(self):
             "type": "Feature",
             "geometry": {
                 "type": "Polygon",
-                "coordinates": [[(0, 0), (1, 0), (1, 1), (0, 1), (0, 0)]]
-            }
+                "coordinates": [[(0, 0), (1, 0), (1, 1), (0, 1), (0, 0)]],
+            },
         }
         result = geojson_to_polygon(geojson_data)
         self.assertIsInstance(result, Polygon)
@@ -93,9 +121,9 @@ def test_geojson_to_polygon_single_feature_multipolygon(self):
                 "type": "MultiPolygon",
                 "coordinates": [
                     [[(0, 0), (1, 0), (1, 1), (0, 1), (0, 0)]],
-                    [[(2, 2), (3, 2), (3, 3), (2, 3), (2, 2)]]
-                ]
-            }
+                    [[(2, 2), (3, 2), (3, 3), (2, 3), (2, 2)]],
+                ],
+            },
         }
         result = geojson_to_polygon(geojson_data)
         self.assertIsInstance(result, MultiPolygon)
@@ -105,102 +133,117 @@ def test_geojson_to_polygon_non_polygon_geometry_in_feature_collection(self):
         geojson_data = {
             "type": "FeatureCollection",
             "features": [
-                {"type": "Feature", "geometry": {"type": "LineString", "coordinates": [(0, 0), (1, 1)]}}
-            ]
+                {
+                    "type": "Feature",
+                    "geometry": {"type": "LineString", "coordinates": [(0, 0), (1, 1)]},
+                }
+            ],
         }
         with self.assertRaises(ValueError) as context:
             geojson_to_polygon(geojson_data)
         self.assertEqual(str(context.exception), "Non-polygon geometries cannot be combined into a MultiPolygon.")
 
     def test_geojson_to_polygon_empty_feature_collection(self):
-        geojson_data = {
-            "type": "FeatureCollection",
-            "features": []
-        }
+        geojson_data = {"type": "FeatureCollection", "features": []}
         result = geojson_to_polygon(geojson_data)
         self.assertTrue(result.is_empty)
 
     def test_geojson_to_polygon_contribution_geojson(self):
         result = geojson_to_polygon(self.fixture_data)
         self.assertIsInstance(result, Polygon)
 
-    @patch('mapswipe_workers.utils.process_mapillary.vt2geojson_tools.vt_bytes_to_geojson')
-    @patch('mapswipe_workers.utils.process_mapillary.requests.get')
+    @patch(
+        "mapswipe_workers.utils.process_mapillary.vt2geojson_tools.vt_bytes_to_geojson"
+    )
+    @patch("mapswipe_workers.utils.process_mapillary.requests.get")
     def test_download_and_process_tile_success(self, mock_get, mock_vt2geojson):
         # Mock the response from requests.get
         mock_response = MagicMock()
         mock_response.status_code = 200
-        mock_response.content = b'mock vector tile data'  # Example mock data
+        mock_response.content = b"mock vector tile data"  # Example mock data
         mock_get.return_value = mock_response
 
         # Mock the return value of vt_bytes_to_geojson
         mock_vt2geojson.return_value = {
             "features": [
-                {"geometry": {"type": "Point", "coordinates": [0, 0]}, "properties": {"id": 1}}
+                {
+                    "geometry": {"type": "Point", "coordinates": [0, 0]},
+                    "properties": {"id": 1},
+                }
             ]
         }
 
-        row = {'x': 1, 'y': 1, 'z': 14}
-        token = 'test_token'
+        row = {"x": 1, "y": 1, "z": 14}
+        token = "test_token"
 
         result, failed = download_and_process_tile(row, token)
 
         # Assertions
         self.assertIsNone(failed)
         self.assertIsInstance(result, pd.DataFrame)
         self.assertEqual(len(result), 1)
-        self.assertEqual(result['geometry'][0].wkt, 'POINT (0 0)')
+        self.assertEqual(result["geometry"][0].wkt, "POINT (0 0)")
 
-    @patch('mapswipe_workers.utils.process_mapillary.requests.get')
+    @patch("mapswipe_workers.utils.process_mapillary.requests.get")
     def test_download_and_process_tile_failure(self, mock_get):
         # Mock a failed response
         mock_response = MagicMock()
         mock_response.status_code = 500
         mock_get.return_value = mock_response
 
-        row = pd.Series({'x': 1, 'y': 1, 'z': self.level})
+        row = pd.Series({"x": 1, "y": 1, "z": self.level})
         result, failed = download_and_process_tile(row, self.token)
 
         self.assertIsNone(result)
         self.assertIsNotNone(failed)
 
-    @patch('mapswipe_workers.utils.process_mapillary.download_and_process_tile')
+    @patch("mapswipe_workers.utils.process_mapillary.download_and_process_tile")
     def test_coordinate_download(self, mock_download_and_process_tile):
-        mock_download_and_process_tile.return_value = (pd.DataFrame([{"geometry": None}]), None)
+        mock_download_and_process_tile.return_value = (
+            pd.DataFrame([{"geometry": None}]),
+            None,
+        )
 
-        metadata, failed = coordinate_download(self.test_polygon, self.level, self.token)
+        metadata, failed = coordinate_download(
+            self.test_polygon, self.level, self.token
+        )
 
         self.assertIsInstance(metadata, pd.DataFrame)
         self.assertTrue(failed.empty)
 
-    @patch('mapswipe_workers.utils.process_mapillary.download_and_process_tile')
+    @patch("mapswipe_workers.utils.process_mapillary.download_and_process_tile")
     def test_coordinate_download_with_failures(self, mock_download_and_process_tile):
-        mock_download_and_process_tile.return_value = (None, pd.Series({"x": 1, "y": 1, "z": self.level}))
+        mock_download_and_process_tile.return_value = (
+            None,
+            pd.Series({"x": 1, "y": 1, "z": self.level}),
+        )
 
-        metadata, failed = coordinate_download(self.test_polygon, self.level, self.token)
+        metadata, failed = coordinate_download(
+            self.test_polygon, self.level, self.token
+        )
 
         self.assertTrue(metadata.empty)
         self.assertFalse(failed.empty)
 
     def test_filter_within_time_range(self):
-        start_time = '2016-01-20 00:00:00'
-        end_time = '2022-01-21 23:59:59'
+        start_time = "2016-01-20 00:00:00"
+        end_time = "2022-01-21 23:59:59"
         filtered_df = filter_by_timerange(self.fixture_df, start_time, end_time)
 
         self.assertEqual(len(filtered_df), 3)
-        self.assertTrue(all(filtered_df['captured_at'] >= pd.to_datetime(start_time)))
-        self.assertTrue(all(filtered_df['captured_at'] <= pd.to_datetime(end_time)))
+        self.assertTrue(all(filtered_df["captured_at"] >= pd.to_datetime(start_time)))
+        self.assertTrue(all(filtered_df["captured_at"] <= pd.to_datetime(end_time)))
 
     def test_filter_without_end_time(self):
-        start_time = '2020-01-20 00:00:00'
+        start_time = "2020-01-20 00:00:00"
         filtered_df = filter_by_timerange(self.fixture_df, start_time)
 
         self.assertEqual(len(filtered_df), 3)
-        self.assertTrue(all(filtered_df['captured_at'] >= pd.to_datetime(start_time)))
+        self.assertTrue(all(filtered_df["captured_at"] >= pd.to_datetime(start_time)))
 
     def test_filter_time_no_data(self):
-        start_time = '2016-01-30 00:00:00'
-        end_time = '2016-01-31 00:00:00'
+        start_time = "2016-01-30 00:00:00"
+        end_time = "2016-01-31 00:00:00"
         filtered_df = filter_by_timerange(self.fixture_df, start_time, end_time)
         self.assertTrue(filtered_df.empty)
 
@@ -221,10 +264,13 @@ def test_filter_organization_id(self):
         self.assertEqual(len(filtered_df), 1)
 
     def test_filter_time_range(self):
-        start_time = '2016-01-20 00:00:00'
-        end_time = '2022-01-21 23:59:59'
-        filtered_df = filter_results(self.fixture_df, start_time=start_time, end_time=end_time)
+        start_time = "2016-01-20 00:00:00"
+        end_time = "2022-01-21 23:59:59"
+        filtered_df = filter_results(
+            self.fixture_df, start_time=start_time, end_time=end_time
+        )
         self.assertEqual(len(filtered_df), 3)
 
-if __name__ == '__main__':
+
+if __name__ == "__main__":
     unittest.main()

mapswipe_workers/tests/unittests/test_project_type_street.py

@@ -26,11 +26,11 @@ def test_create_group(self):
         self.assertTrue(self.project.groups)
 
     def test_create_tasks(self):
-        imageId = self.project.imageList[-1]
+        imageId = self.project.imageIds[-1]
         self.project.create_groups()
         self.project.create_tasks()
         self.assertEqual(self.project.tasks["g0"][0].taskId, imageId)
-        #self.assertEqual(self.project.groups["g0"].numberOfTasks, 1)
+        # self.assertEqual(self.project.groups["g0"].numberOfTasks, 1)
 
 
 if __name__ == "__main__":