Fixed unit test cases

Author: sujata-m

src/osm_osw_reformatter/serializer/osm/osm_graph.py

@@ -632,7 +632,9 @@ def to_geojson(self, *args) -> None:
         polygon_features = []
         for n, d in self.G.nodes(data=True):
             d_copy = {**d}
-            d_copy["_id"] = str(n)[1:]
+            id_str = str(n)
+            trimmed_id = id_str[1:] if isinstance(n, str) else id_str
+            d_copy["_id"] = trimmed_id
             d_copy['ext:osm_id'] = str(d_copy.get('osm_id', d_copy["_id"]))
 
             if OSWPointNormalizer.osw_point_filter(d):
@@ -665,7 +667,7 @@ def to_geojson(self, *args) -> None:
                 polygon_features.append(
                     {"type": "Feature", "geometry": geometry, "properties": d_copy}
                 )
-            else:
+            elif OSWNodeNormalizer.osw_node_filter(d) or self.G.degree(n) > 0:
                 d_copy['_id'] = str(n)
 
                 geometry = mapping(d_copy.pop('geometry'))

src/osm_osw_reformatter/serializer/osm/osm_normalizer.py

@@ -87,22 +87,46 @@ def process_output(self, osmnodes, osmways, osmrelations):
         """
         mask_63bit = (1 << 63) - 1
 
+        def _set_id_tag(osm_obj, new_id):
+            tags = getattr(osm_obj, "tags", None)
+            if tags is None or not hasattr(tags, "__setitem__"):
+                return
+
+            value = str(new_id)
+            existing = tags.get("_id") if hasattr(tags, "get") else None
+
+            if isinstance(existing, list):
+                tags["_id"] = [value]
+            elif existing is None:
+                # Determine if the container generally stores values as lists
+                sample_value = None
+                if hasattr(tags, "values"):
+                    for sample_value in tags.values():
+                        if sample_value is not None:
+                            break
+                if isinstance(sample_value, list):
+                    tags["_id"] = [value]
+                else:
+                    # Default to list storage to match ogr2osm's internal structures
+                    tags["_id"] = [value]
+            else:
+                tags["_id"] = value
+
+        def _normalise_id(osm_obj):
+            if osm_obj.id < 0:
+                new_id = osm_obj.id & mask_63bit
+                osm_obj.id = new_id
+                _set_id_tag(osm_obj, new_id)
+                return new_id
+            return osm_obj.id
 
         # Fix node IDs
         for node in osmnodes:
-            if node.id < 0:
-                if mask_63bit == 3964026:
-                    print('node')
-                    print(node.id)
-                node.id = node.id & mask_63bit
+            _normalise_id(node)
 
         # Fix ways and their node references
         for way in osmways:
-            if way.id < 0:
-                if mask_63bit == 3964026:
-                    print('node')
-                    print(way.id)
-                way.id = way.id & mask_63bit
+            _normalise_id(way)
 
             # Detect how node references are stored
             node_refs = getattr(way, "nds", None) or getattr(way, "refs", None) or getattr(way, "nodeRefs", None) or getattr(way, "nodes", None)
@@ -115,10 +139,8 @@ def process_output(self, osmnodes, osmways, osmrelations):
                         new_refs.append(ref & mask_63bit if ref < 0 else ref)
                     elif hasattr(ref, "id"):
                         if ref.id < 0:
-                            if mask_63bit == 3964026:
-                                print('ref')
-                                print(ref.id)
                             ref.id = ref.id & mask_63bit
+                            _set_id_tag(ref, ref.id)
                         new_refs.append(ref)
                     else:
                         new_refs.append(ref)
@@ -136,24 +158,25 @@ def process_output(self, osmnodes, osmways, osmrelations):
         # Fix relation IDs and their member refs
         for rel in osmrelations:
             if rel.id < 0:
-                if mask_63bit == 3964026:
-                    print('rel')
-                    print(rel.id)
                 rel.id = rel.id & mask_63bit
+            _normalise_id(rel)
 
             if hasattr(rel, "members"):
                 for member in rel.members:
                     if hasattr(member, "ref"):
                         ref = member.ref
                         if isinstance(ref, int) and ref < 0:
-                            if mask_63bit == 3964026:
-                                print('members ref 1')
-                                print(ref.id)
                             member.ref = ref & mask_63bit
                         elif hasattr(ref, "id") and ref.id < 0:
-                            if mask_63bit == 3964026:
-                                print('members ref 2')
-                                print(ref.id)
                             ref.id = ref.id & mask_63bit
+                            _set_id_tag(ref, ref.id)
+
+        # Ensure deterministic ordering now that IDs have been normalised
+        if hasattr(osmnodes, "sort"):
+            osmnodes.sort(key=lambda n: n.id)
+        if hasattr(osmways, "sort"):
+            osmways.sort(key=lambda w: w.id)
+        if hasattr(osmrelations, "sort"):
+            osmrelations.sort(key=lambda r: r.id)
 
 

tests/unit_tests/test_osm_compliance/test_osm_compliance.py

@@ -51,18 +51,17 @@ async def test_incline_tag_preserved(self):
         osw_files = res.generated_files
 
         found_incline = False
-        if osw_files:
-            for f in osw_files:
-                if f.endswith('.geojson'):
-                    with open(f) as fh:
-                        data = json.load(fh)
-                        for feature in data.get('features', []):
-                            props = feature.get('properties', {})
-                            if 'incline' in props:
-                                found_incline = True
-                                break
-                if found_incline:
-                    break
+        for f in osw_files:
+            if f.endswith('.geojson'):
+                with open(f) as fh:
+                    data = json.load(fh)
+                    for feature in data.get('features', []):
+                        props = feature.get('properties', {})
+                        if 'incline' in props:
+                            found_incline = True
+                            break
+            if found_incline:
+                break
 
         self.assertTrue(found_incline, 'No incline tag found in OSW output')
 

tests/unit_tests/test_serializer/test_osm_graph.py

@@ -53,6 +53,70 @@ def filter_func(u, v, d):
             list(filtered_graph.get_graph().edges(data=True))[0][2]["property"], "A"
         )
 
+    def test_node_adds_tagged_node_with_coordinates(self):
+        class DummyNode:
+            def __init__(self, node_id, tags, lon, lat):
+                self.id = node_id
+                self.tags = tags
+
+                class Location:
+                    def __init__(self, lon, lat):
+                        self.lon = lon
+                        self.lat = lat
+
+                self.location = Location(lon, lat)
+
+        tagged = DummyNode(101, {"highway": "footway"}, -122.3, 47.6)
+
+        self.osm_graph.node(tagged)
+
+        self.assertIn(101, self.mock_graph.nodes)
+        node_data = self.mock_graph.nodes[101]
+        self.assertEqual(node_data["highway"], "footway")
+        self.assertEqual(node_data["lon"], -122.3)
+        self.assertEqual(node_data["lat"], 47.6)
+
+    def test_node_skips_existing_identifiers(self):
+        self.mock_graph.add_node(202, lon=0.0, lat=0.0, name="original")
+
+        class DummyNode:
+            def __init__(self):
+                self.id = 202
+                self.tags = {"name": "replacement"}
+
+                class Location:
+                    lon = 1.0
+                    lat = 1.0
+
+                self.location = Location()
+
+        duplicate = DummyNode()
+
+        self.osm_graph.node(duplicate)
+
+        node_data = self.mock_graph.nodes[202]
+        self.assertEqual(node_data["name"], "original")
+        self.assertEqual(node_data["lon"], 0.0)
+        self.assertEqual(node_data["lat"], 0.0)
+
+    def test_node_requires_tags(self):
+        class DummyNode:
+            def __init__(self):
+                self.id = 303
+                self.tags = {}
+
+                class Location:
+                    lon = 2.0
+                    lat = 2.0
+
+                self.location = Location()
+
+        empty = DummyNode()
+
+        self.osm_graph.node(empty)
+
+        self.assertNotIn(303, self.mock_graph.nodes)
+
     @patch('src.osm_osw_reformatter.serializer.osm.osm_graph.mapping')
     def test_to_geojson(self, mock_mapping):
         # Mock mapping function to return a sample GeoJSON structure
@@ -457,16 +521,29 @@ def test_construct_geometries_line_node(self):
 class TestFromGeoJSON(unittest.TestCase):
     def setUp(self):
         # Create valid test GeoJSON files
-        self.nodes_path = "test_nodes.geojson"
-        self.edges_path = "test_edges.geojson"
+        self.tempdir = TemporaryDirectory()
+        self.nodes_path = os.path.join(self.tempdir.name, "nodes.geojson")
+        self.edges_path = os.path.join(self.tempdir.name, "edges.geojson")
 
-        self.node_data = {
+    def tearDown(self):
+        self.tempdir.cleanup()
+
+    def _write_geojson(self, node_data, edge_data):
+        with open(self.nodes_path, "w") as f:
+            json.dump(node_data, f)
+
+        with open(self.edges_path, "w") as f:
+            json.dump(edge_data, f)
+
+
+    def test_from_geojson_populates_graph(self):
+        node_data = {
             "type": "FeatureCollection",
             "features": [
                 {
                     "type": "Feature",
                     "geometry": {"type": "Point", "coordinates": [1, 1]},
-                    "properties": {"_id": "1", "attribute": "value1"},
+                    "properties": {"_id": "1", "attribute": "value1", "ext:osm_id": "11"},
                 },
                 {
                     "type": "Feature",
@@ -476,45 +553,88 @@ def setUp(self):
             ],
         }
 
-        self.edge_data = {
+        edge_data = {
             "type": "FeatureCollection",
             "features": [
                 {
                     "type": "Feature",
                     "geometry": {"type": "LineString", "coordinates": [[1, 1], [2, 2]]},
-                    "properties": {"_id": "1", "_u_id": "1", "_v_id": "2", "attribute": "edge_value"},
+                    "properties": {
+                        "_id": "5",
+                        "_u_id": "1",
+                        "_v_id": "2",
+                        "attribute": "edge_value",
+                        "ext:osm_id": "99",
+                    },
                 },
             ],
         }
 
-        # Write the data to files
-        with open(self.nodes_path, "w") as f:
-            json.dump(self.node_data, f)
+        self._write_geojson(node_data, edge_data)
+        osm_graph = OSMGraph.from_geojson(self.nodes_path, self.edges_path)
+        graph = osm_graph.get_graph()
 
-        with open(self.edges_path, "w") as f:
-            json.dump(self.edge_data, f)
+        self.assertEqual(set(graph.nodes), {1, 2})
+        node_attrs = graph.nodes[1]
+        self.assertIsInstance(node_attrs["geometry"], Point)
+        self.assertEqual(node_attrs["lon"], 1)
+        self.assertEqual(node_attrs["lat"], 1)
+        self.assertEqual(node_attrs["osm_id"], 11)
 
-    def tearDown(self):
-        # Clean up files after tests
-        import os
-        if os.path.exists(self.nodes_path):
-            os.remove(self.nodes_path)
-        if os.path.exists(self.edges_path):
-            os.remove(self.edges_path)
-
-    @patch("src.osm_osw_reformatter.serializer.osm.osm_graph.OSMGraph.from_geojson")
-    def test_from_geojson(self, mock_from_geojson):
-        mock_graph = MagicMock()
-        mock_graph.get_graph.return_value.nodes = {"1": {"geometry": Point(1, 1)}}
-        mock_graph.get_graph.return_value.edges = {("1", "2"): {"geometry": LineString([(1, 1), (2, 2)])}}
-        mock_from_geojson.return_value = mock_graph
+        edges = list(graph.edges(keys=True, data=True))
+        self.assertEqual(len(edges), 1)
+        u, v, key, attrs = edges[0]
+        self.assertEqual((u, v, key), (1, 2, 5))
+        self.assertIsInstance(attrs["geometry"], LineString)
+        self.assertEqual(attrs["osm_id"], 99)
+        self.assertEqual(attrs["attribute"], "edge_value")
+
+    def test_from_geojson_preserves_non_numeric_identifiers(self):
+        node_data = {
+            "type": "FeatureCollection",
+            "features": [
+                {
+                    "type": "Feature",
+                    "geometry": {"type": "Point", "coordinates": [3, 3]},
+                    "properties": {"_id": "p123", "ext:osm_id": "node-1"},
+                },
+                {
+                    "type": "Feature",
+                    "geometry": {"type": "Point", "coordinates": [4, 4]},
+                    "properties": {"_id": "p456"},
+                },
+            ],
+        }
+
+        edge_data = {
+            "type": "FeatureCollection",
+            "features": [
+                {
+                    "type": "Feature",
+                    "geometry": {"type": "LineString", "coordinates": [[3, 3], [4, 4]]},
+                    "properties": {
+                        "_id": "edge-7",
+                        "_u_id": "p123",
+                        "_v_id": "p456",
+                        "ext:osm_id": "edge-A",
+                    },
+                },
+            ],
+        }
+
+        self._write_geojson(node_data, edge_data)
 
         osm_graph = OSMGraph.from_geojson(self.nodes_path, self.edges_path)
+        graph = osm_graph.get_graph()
+
+        self.assertIn("p123", graph.nodes)
+        self.assertEqual(graph.nodes["p123"]["osm_id"], "node-1")
 
-        # Assertions
-        self.assertIsNotNone(osm_graph, "OSMGraph object should not be None")
-        self.assertEqual(len(osm_graph.get_graph().nodes), 1)
-        self.assertEqual(len(osm_graph.get_graph().edges), 1)
+        edges = list(graph.edges(keys=True, data=True))
+        self.assertEqual(len(edges), 1)
+        u, v, key, attrs = edges[0]
+        self.assertEqual((u, v, key), ("p123", "p456", "edge-7"))
+        self.assertEqual(attrs["osm_id"], "edge-A")
 
     def test_tagged_node_parser_skips_non_osw_nodes(self):
         graph = nx.MultiDiGraph()