feat: add Docker/OpenTofu deployment and DE/NL routing
This commit is contained in:
BuTzZ
@@ -0,0 +1,194 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user