Merge pull request #107 from mapswipe/feature/unit-testing-additional

Author: tnagorra

project_types/validate/api_calls.py

@@ -37,6 +37,7 @@ def retry_get(url: str, retries: int | None = 3, timeout: int | None = 4, to_osm
         return session.get(url, timeout=timeout)
 
 
+# FIXME(rup): Not used anywhere
 def geojsonToFeatureCollection(geojson: dict) -> dict:
     """Take a GeoJson and wrap it in a FeatureCollection."""
     if geojson["type"] != "FeatureCollection":

project_types/validate/tests/__init__.py

@@ -0,0 +1,276 @@
+import typing
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from main.config import Config
+from main.tests import TestCase
+from project_types.validate.api_calls import (
+    ValidateApiCallError,
+    ohsome,
+    query_osm,
+    query_osmcha,
+    remove_noise_and_add_user_info,
+    remove_troublesome_chars,
+)
+
+
+class TestValidateProject(TestCase):
+    @typing.override
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+
+    def test_remove_troublesome_chars(self):
+        data = [
+            ("Hello\nWorld", "Hello World"),
+            ('She said "Hello"', "She said Hello"),
+            ("It's fine", "Its fine"),
+            ("Hello 'World'", "Hello World"),
+            (None, None),
+            (123, 123),
+        ]
+        for input_str, expected in data:
+            result = remove_troublesome_chars(input_str)
+            assert result == expected
+
+    @patch("project_types.validate.api_calls.retry_get")
+    def test_query_osmcha(self, mock_retry_get):
+        changeset_ids = [12345, 67890]
+        changeset_results = {}
+
+        # Fake API JSON response
+        mock_response_data = {
+            "features": [
+                {
+                    "id": "12345",
+                    "properties": {
+                        "user": "Sita Devi",
+                        "uid": 1001,
+                        "comment": "Looks good!",
+                        "editor": "Ram",
+                    },
+                },
+                {
+                    "id": "67890",
+                    "properties": {
+                        "user": "Hari",
+                        "uid": 1002,
+                        "comment": "It's not a bridge",
+                        "editor": "Shyam 'Bahadur'",
+                    },
+                },
+            ],
+        }
+
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = mock_response_data
+        mock_retry_get.return_value = mock_response
+
+        query_osmcha(changeset_ids, changeset_results)
+
+        assert changeset_results == {
+            12345: {
+                "username": "Sita Devi",
+                "userid": 1001,
+                "comment": "Looks good!",
+                "editor": "Ram",
+            },
+            67890: {
+                "username": "Hari",
+                "userid": 1002,
+                "comment": "Its not a bridge",
+                "editor": "Shyam Bahadur",
+            },
+        }
+
+        # Check request is made for osmcha
+        mock_retry_get.assert_called_once()
+        called_url = mock_retry_get.call_args[0][0]
+        assert called_url.startswith(Config.OSMCHA_API_LINK)
+        assert "changesets/?ids=12345,67890" in called_url
+
+        # check for other status code to raise error
+        mock_response.status_code = 403
+        with pytest.raises(ValidateApiCallError):
+            query_osmcha(changeset_ids, changeset_results)
+
+    @patch("project_types.validate.api_calls.retry_get")
+    def test_query_osm(self, mock_retry_get):
+        changeset_ids = [12345, 67890]
+        changeset_results = {}
+
+        xml_response = """
+        <osm>
+            <changeset id="12345" user='Sita "Devi"' uid="1001">
+                <tag k="comment" v="Looks good!"/>
+                <tag k="created_by" v="Ram"/>
+            </changeset>
+            <changeset id="67890" user="Hari" uid="1002">
+                <tag k="comment" v="It's not a bridge"/>
+                <tag k="created_by" v="Shyam Bahadur"/>
+            </changeset>
+        </osm>
+        """
+
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.content = xml_response.encode("utf-8")
+        mock_retry_get.return_value = mock_response
+
+        result = query_osm(changeset_ids, changeset_results)
+
+        assert result == {
+            12345: {
+                "username": "Sita Devi",
+                "userid": "1001",
+                "comment": "Looks good!",
+                "editor": "Ram",
+            },
+            67890: {
+                "username": "Hari",
+                "userid": "1002",
+                "comment": "Its not a bridge",
+                "editor": "Shyam Bahadur",
+            },
+        }
+
+        # check for other status code to raise error
+        mock_response.status_code = 500
+        with pytest.raises(ValidateApiCallError):
+            query_osm([12345], {})
+
+    @patch("requests.post")
+    @patch("project_types.validate.api_calls.remove_noise_and_add_user_info")
+    def test_ohsome(self, mock_remove_noise, mock_post):
+        sample_request = {
+            "endpoint": "elements/geometry",
+            "filter": "highway=primary",
+        }
+        sample_area = "POLYGON((8.67 49.39,8.68 49.39,8.68 49.40,8.67 49.40,8.67 49.39))"
+        sample_properties = "tags,timestamp"
+
+        mock_response_data = {
+            "attribution": {
+                "url": "https://ohsome.org/copyrights",
+                "text": "© OpenStreetMap contributors",
+            },
+            "apiVersion": "1.10.4",
+            "type": "FeatureCollection",
+            "features": [
+                {
+                    "type": "Feature",
+                    "geometry": {
+                        "type": "Point",
+                        "coordinates": [
+                            8.6861,
+                            49.4051089,
+                        ],
+                    },
+                    "properties": {
+                        "@osmId": "node/385941986",
+                        "@snapshotTimestamp": "2019-09-01T00: 00: 00Z",
+                    },
+                },
+                {
+                    "type": "Feature",
+                    "geometry": {
+                        "type": "Point",
+                        "coordinates": [
+                            8.6819524,
+                            49.3825748,
+                        ],
+                    },
+                    "properties": {
+                        "@osmId": "node/699583613",
+                        "@snapshotTimestamp": "2019-09-01T00: 00: 00Z",
+                    },
+                },
+            ],
+        }
+
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = mock_response_data
+        mock_post.return_value = mock_response
+
+        processed_data = {"processed": True, "data": mock_response_data}
+        mock_remove_noise.return_value = processed_data
+
+        result = ohsome(sample_request, sample_area, sample_properties)
+
+        mock_post.assert_called_once()
+        mock_remove_noise.assert_called_once_with(mock_response_data)
+        assert result == processed_data
+
+        # test for error
+        status_codes = [400, 401, 403, 404, 500]
+        for status_code in status_codes:
+            mock_response.status_code = status_code
+            with pytest.raises(ValidateApiCallError):
+                result = ohsome(sample_request, sample_area, sample_properties)
+
+    @patch("project_types.validate.api_calls.query_osmcha")
+    @patch("project_types.validate.api_calls.query_osm")
+    def test_remove_noise_and_add_user_info(self, mock_query_osm, mock_query_osmcha):
+        input_data = {
+            "type": "FeatureCollection",
+            "features": [
+                {
+                    "type": "Feature",
+                    "geometry": {"type": "Point", "coordinates": [0, 0]},
+                    "properties": {
+                        "@changesetId": 12345,
+                        "@lastEdit": 1234567890,
+                        "@osmId": 111,
+                        "@version": 1,
+                        "unwanted_field": "should_be_removed",
+                        "another_noise": 999,
+                    },
+                },
+                {
+                    "type": "Feature",
+                    "geometry": {"type": "Point", "coordinates": [1, 1]},
+                    "properties": {
+                        "@changesetId": 12346,
+                        "@lastEdit": 1234567891,
+                        "@osmId": 222,
+                        "@version": 2,
+                        "extra_data": "also_removed",
+                    },
+                },
+            ],
+        }
+        mock_osmcha_response = {
+            12345: {
+                "username": "Sita",
+                "comment": "This is a test comment",
+                "editor": "iD",
+                "userid": 1001,
+            },
+            12346: None,
+        }
+
+        mock_query_osmcha.return_value = mock_osmcha_response
+
+        mock_query_osm.return_value = {
+            12345: {
+                "username": "Sita",
+                "userid": "1001",
+                "comment": "This is an updated test comment",
+                "editor": "Ram",
+            },
+            12346: {
+                "username": "Kiran",
+                "userid": "1002",
+                "comment": "This is a test comment",
+                "editor": "Hari",
+            },
+        }
+
+        remove_noise_and_add_user_info(input_data)
+        mock_query_osmcha.assert_called()
+        called_ids = mock_query_osmcha.call_args[0][0]
+        assert set(called_ids) == {12345, 12346}
+        mock_query_osm.assert_called()

project_types/validate/tests/api_calls_test.py

@@ -7,6 +7,7 @@
 from django.core.files.base import ContentFile
 from pydantic_core import ValidationError as PydanticValidationError
 
+from apps.tutorial.models import generate_tutorial_firebase_id
 from main.tests import TestCase
 from utils.common import (
     compress_tasks,
@@ -262,3 +263,15 @@ def test_parse_b64gzjson_to_dict(self):
         compressed_task = compress_tasks(original_json)
         result = parse_b64gzjson_to_dict(compressed_task)
         assert result == original_json
+
+    def test_generate_tutorial_firebase_id(self):
+        tutorial_firebase_id = generate_tutorial_firebase_id()
+        assert tutorial_firebase_id.startswith("tutorial_")
+
+        # check is_valid
+        split_id = tutorial_firebase_id.split("_")
+        validate_ulid(split_id[1])
+
+        # check unique tutorial id
+        tutorial_firebase_id_2 = generate_tutorial_firebase_id()
+        assert tutorial_firebase_id != tutorial_firebase_id_2

utils/tests/common_test.py

@@ -0,0 +1,72 @@
+import typing
+
+from main.tests import TestCase
+from utils.geo.tile_functions import (
+    lat_long_zoom_to_pixel_coords,
+    pixel_coords_zoom_to_lat_lon,
+    tile_coords_and_zoom_to_quad_key,
+)
+
+
+class TestGeoUtils(TestCase):
+    @typing.override
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+
+    def test_lat_long_zoom_to_pixel_coords(self):
+        # Lat=0, Lon=0 at zoom=0 (whole world 256x256)
+        p = lat_long_zoom_to_pixel_coords(0, 0, 0)
+        assert p.x == 128
+        assert p.y == 128
+
+        # test with zoom level 1
+        p = lat_long_zoom_to_pixel_coords(0, 0, 1)
+        # At zoom 1, map is 512x512 pixels, middle is 256,256
+        assert p.x == 256
+        assert p.y == 256
+
+        # test with zoom level 2
+        p = lat_long_zoom_to_pixel_coords(0, 0, 2)
+        assert p.x == 512
+        assert p.y == 512
+
+        # test with zoom level 5
+        p = lat_long_zoom_to_pixel_coords(0, 0, 5)
+        assert p.x == 4096
+        assert p.y == 4096
+
+    def test_pixel_coords_zoom_to_lat_lon(self):
+        # Center of the map at zoom 0
+        lon, lat = pixel_coords_zoom_to_lat_lon(128, 128, 0)
+        assert lon == 0.0
+        assert lat == 0.0
+
+        # Known point (London: 51.5074° N, 0.1278° W)
+        lon, lat = pixel_coords_zoom_to_lat_lon(512, 341.3333, 2)
+        assert lon == 0
+        assert lat == 51.32604237282497
+
+    def test_tile_coords_and_zoom_to_quadKey(self):
+        # At zoom 0, only one tile: quadkey is empty
+        assert not tile_coords_and_zoom_to_quad_key(0, 0, 0)
+
+        # Zoom 1: top-left, top-right, bottom-left, bottom-right
+        assert tile_coords_and_zoom_to_quad_key(0, 0, 1) == "0"
+        assert tile_coords_and_zoom_to_quad_key(1, 0, 1) == "1"
+        assert tile_coords_and_zoom_to_quad_key(0, 1, 1) == "2"
+        assert tile_coords_and_zoom_to_quad_key(1, 1, 1) == "3"
+
+        # Zoom 3: arbitrary
+        assert tile_coords_and_zoom_to_quad_key(3, 5, 3) == "213"
+
+        # Zoom 5: arbitrary
+        assert tile_coords_and_zoom_to_quad_key(10, 12, 5) == "03210"
+
+        # Zoom 22: top-left, bottom-right
+        assert tile_coords_and_zoom_to_quad_key(0, 0, 22) == "0000000000000000000000"
+        assert tile_coords_and_zoom_to_quad_key(4194303, 4194303, 22) == "3333333333333333333333"  # max tile = (2**22) - 1
+
+        # Zoom 23: top-left, bottom-right
+        assert tile_coords_and_zoom_to_quad_key(0, 0, 23) == "00000000000000000000000"
+        assert tile_coords_and_zoom_to_quad_key(8388607, 8388607, 23) == "33333333333333333333333"  # maxtile = (2*23) - 1