watermaps/scripts/tests/test_build_local_fairways.py

from __future__ import annotations

import hashlib
import json
import os
from pathlib import Path
import stat
import subprocess
import sys
import tempfile
import unittest

try:
    import osmium
except ImportError:
    osmium = None


ROOT_DIR = Path(__file__).resolve().parents[2]
BUILDER = ROOT_DIR / "scripts" / "build-local-fairways.py"


def write_pbf(
    path: Path,
    nodes: list[tuple[int, float, float]],
    ways: list[tuple[int, int, list[int], dict[str, str]]],
) -> None:
    assert osmium is not None
    with osmium.SimpleWriter(str(path)) as writer:
        for node_id, lon, lat in nodes:
            writer.add_node(
                osmium.osm.mutable.Node(
                    id=node_id,
                    version=1,
                    location=(lon, lat),
                )
            )
        for way_id, version, node_ids, tags in ways:
            writer.add_way(
                osmium.osm.mutable.Way(
                    id=way_id,
                    version=version,
                    nodes=node_ids,
                    tags=tags,
                )
            )
    checksum = hashlib.md5(path.read_bytes()).hexdigest()
    Path(f"{path}.md5").write_text(
        f"{checksum}  {path.name}\n",
        encoding="utf-8",
    )


@unittest.skipIf(osmium is None, "pyosmium/osmium is not installed")
class BuildLocalFairwaysTest(unittest.TestCase):
    def test_merges_two_extracts_and_deduplicates_shared_osm_ways(self) -> None:
        with tempfile.TemporaryDirectory() as temporary_directory:
            work_dir = Path(temporary_directory)
            germany = work_dir / "germany-latest.osm.pbf"
            netherlands = work_dir / "netherlands-latest.osm.pbf"
            output = work_dir / "germany-netherlands-fairways.json"

            write_pbf(
                germany,
                [
                    (1, 7.0, 53.0),
                    (2, 7.1, 53.1),
                    (3, 7.2, 53.2),
                ],
                [
                    (100, 1, [1, 2], {"waterway": "canal", "name": "DE Kanal"}),
                    (
                        200,
                        1,
                        [2, 3],
                        {"seamark:type": "navigation_line", "name": "Alte Linie"},
                    ),
                    (300, 1, [1, 2, 1], {"seamark:type": "fairway"}),
                    (500, 1, [1, 9], {"waterway": "canal", "name": "Grenzkanal"}),
                    (600, 1, [1, 9999], {"waterway": "canal"}),
                ],
            )
            write_pbf(
                netherlands,
                [
                    (2, 7.1, 53.1),
                    (3, 7.2, 53.2),
                    (4, 7.3, 53.3),
                    (9, 7.05, 53.05),
                ],
                [
                    (
                        200,
                        2,
                        [2, 3, 4],
                        {"seamark:type": "navigation_line", "name": "Nieuwe lijn"},
                    ),
                    (400, 1, [3, 4], {"route": "ferry", "name": "Veerboot"}),
                ],
            )

            result = subprocess.run(
                [
                    sys.executable,
                    str(BUILDER),
                    str(germany),
                    str(netherlands),
                    "--output",
                    str(output),
                ],
                cwd=ROOT_DIR,
                env={
                    **os.environ,
                    "PYTHONPATH": str(ROOT_DIR / ".tools" / "python"),
                },
                check=False,
                capture_output=True,
                text=True,
            )

            self.assertEqual(result.returncode, 0, result.stderr)
            document = json.loads(output.read_text(encoding="utf-8"))
            self.assertEqual(
                stat.S_IMODE(output.stat().st_mode),
                0o644,
                "the unprivileged application container must be able to read the index",
            )
            self.assertEqual(document["version"], 1)
            self.assertEqual(document["generatorVersion"], 2)
            self.assertEqual(document["source"], "germany+netherlands")
            self.assertEqual(document["duplicateWaysMerged"], 1)
            self.assertEqual(document["incompleteWaysSkipped"], 1)
            self.assertEqual(document["closedFairwaysSkipped"], 1)
            self.assertEqual([way["id"] for way in document["ways"]], ["100", "200", "400", "500"])

            shared_way = next(way for way in document["ways"] if way["id"] == "200")
            self.assertEqual(shared_way["osmVersion"], 2)
            self.assertEqual(shared_way["regions"], ["germany", "netherlands"])
            self.assertEqual(
                shared_way["sourceFiles"],
                ["germany-latest.osm.pbf", "netherlands-latest.osm.pbf"],
            )
            self.assertEqual(shared_way["tags"]["name"], "Nieuwe lijn")
            self.assertEqual(len(shared_way["coordinates"]), 3)

            cross_border_way = next(
                way for way in document["ways"] if way["id"] == "500"
            )
            self.assertEqual(
                cross_border_way["coordinates"],
                [[53.0, 7.0], [53.05, 7.05]],
            )

            self.assertEqual(
                [source["region"] for source in document["sources"]],
                ["germany", "netherlands"],
            )
            self.assertTrue(
                all(source["checksumMd5"] for source in document["sources"])
            )

    def test_failed_source_read_keeps_existing_output(self) -> None:
        with tempfile.TemporaryDirectory() as temporary_directory:
            work_dir = Path(temporary_directory)
            broken_pbf = work_dir / "germany-latest.osm.pbf"
            output = work_dir / "germany-netherlands-fairways.json"
            broken_pbf.write_bytes(b"not an OSM PBF")
            previous_content = '{"version":1,"source":"previous","ways":[]}\n'
            output.write_text(previous_content, encoding="utf-8")

            result = subprocess.run(
                [
                    sys.executable,
                    str(BUILDER),
                    str(broken_pbf),
                    "--output",
                    str(output),
                ],
                cwd=ROOT_DIR,
                env={
                    **os.environ,
                    "PYTHONPATH": str(ROOT_DIR / ".tools" / "python"),
                },
                check=False,
                capture_output=True,
                text=True,
            )

            self.assertNotEqual(result.returncode, 0)
            self.assertEqual(output.read_text(encoding="utf-8"), previous_content)


if __name__ == "__main__":
    unittest.main()