from __future__ import annotations import os import re import shutil import unittest from io import BytesIO from pathlib import Path from unittest.mock import patch os.environ["AMUNDSEN_TEST_STORE"] = "memory" TEST_UPLOAD_ROOT = Path(__file__).resolve().parents[1] / ".tmp" TEST_UPLOAD_ROOT.mkdir(exist_ok=True) TEST_UPLOAD_DIR = TEST_UPLOAD_ROOT / "test-uploads" shutil.rmtree(TEST_UPLOAD_DIR, ignore_errors=True) TEST_UPLOAD_DIR.mkdir(parents=True, exist_ok=True) os.environ["AMUNDSEN_UPLOAD_DIR"] = str(TEST_UPLOAD_DIR) from amundsen_app import app, project_store from amundsen.core.project_store import DatabaseConfig, ProjectStore class AppUploadTest(unittest.TestCase): @classmethod def tearDownClass(cls) -> None: shutil.rmtree(TEST_UPLOAD_DIR, ignore_errors=True) def _logged_in_client(self): client = app.test_client() with client.session_transaction() as active_session: active_session["user_id"] = "test" active_session["username"] = "test" return client def _mini_gpx(self) -> bytes: return b""" Mini route Start Finish Track 1000 1012 1004 """ def _line_gpx(self, name: str, points: list[tuple[float, float, int]]) -> bytes: trackpoints = "\n".join( f' {elevation}' for lat, lon, elevation in points ) return f""" {name} {name} {trackpoints} """.encode() def _last_route_id(self) -> str: return project_store.projects[-1]["routes"][-1]["id"] def test_upload_saves_route_without_rendering_report(self) -> None: client = self._logged_in_client() response = client.post( "/wandelen/analyze", data={"gpx_files": (BytesIO(self._mini_gpx()), "mini.gpx")}, content_type="multipart/form-data", follow_redirects=True, ) self.assertEqual(response.status_code, 200) self.assertIn(b"Mini route", response.data) self.assertIn(b"Analyse wordt pas gemaakt wanneer je de route opent.", response.data) self.assertNotIn(b"routeDirectionSelect", response.data) self.assertNotIn(b"route_directions", response.data) def test_upload_saves_original_gpx_on_server(self) -> None: client = self._logged_in_client() client.post( "/wandelen/analyze", data={"gpx_files": (BytesIO(self._mini_gpx()), "mini.gpx")}, content_type="multipart/form-data", ) saved_files = list(Path(TEST_UPLOAD_DIR).rglob("*mini.gpx")) self.assertTrue(saved_files) self.assertIn(b"Mini route", saved_files[0].read_bytes()) def test_project_page_shows_routes_and_upload(self) -> None: client = self._logged_in_client() client.post( "/wandelen/analyze", data={"gpx_files": (BytesIO(self._mini_gpx()), "mini.gpx")}, content_type="multipart/form-data", ) response = client.get("/wandelen/projects/default") self.assertEqual(response.status_code, 200) self.assertIn(b"Nieuwe route toevoegen", response.data) self.assertIn(b"Mini route", response.data) self.assertIn(b"Analyse wordt pas gemaakt wanneer je de route opent.", response.data) def test_projects_manager_endpoint_renders_project_overview(self) -> None: client = self._logged_in_client() response = client.get("/wandelen/projects/manage") self.assertEqual(response.status_code, 200) self.assertIn(b"Projecten beheren", response.data) self.assertIn(b"Nieuwe wandeling toevoegen", response.data) def test_route_metadata_can_be_updated(self) -> None: client = self._logged_in_client() upload_response = client.post( "/wandelen/analyze", data={"gpx_files": (BytesIO(self._mini_gpx()), "mini.gpx")}, content_type="multipart/form-data", ) route_id = self._last_route_id() update_response = client.post( f"/wandelen/routes/{route_id}/metadata", data={"name": "Nieuwe routenaam", "route_date": "2026-08-08", "next": "/wandelen/projects/default"}, follow_redirects=True, ) report_response = client.get(f"/wandelen/report/{route_id}") self.assertEqual(update_response.status_code, 200) self.assertIn(b"Nieuwe routenaam", update_response.data) self.assertIn(b"08-08-2026", update_response.data) self.assertIn(b"Nieuwe routenaam", report_response.data) self.assertIn(b"08-08-2026", report_response.data) def test_route_card_has_edit_and_delete_icons(self) -> None: client = self._logged_in_client() client.post( "/wandelen/analyze", data={"gpx_files": (BytesIO(self._line_gpx("Route met iconen", [(46.0, 7.0, 1000), (46.001, 7.0, 1010)])), "iconen.gpx")}, content_type="multipart/form-data", ) response = client.get("/wandelen/projects/default") self.assertEqual(response.status_code, 200) self.assertIn(b"data-toggle-route-edit", response.data) self.assertIn(b"Naam en datum wijzigen", response.data) self.assertIn(b"Route verwijderen", response.data) self.assertIn(b"/delete", response.data) def test_route_can_be_deleted(self) -> None: client = self._logged_in_client() client.post( "/wandelen/analyze", data={"gpx_files": (BytesIO(self._line_gpx("Route voor verwijderen", [(46.0, 7.0, 1000), (46.001, 7.0, 1010)])), "verwijderen.gpx")}, content_type="multipart/form-data", ) route_id = self._last_route_id() delete_response = client.post( f"/wandelen/routes/{route_id}/delete", data={"next": "/wandelen/projects/default"}, follow_redirects=True, ) report_response = client.get(f"/wandelen/api/report/{route_id}") self.assertEqual(delete_response.status_code, 200) self.assertNotIn(b"Route voor verwijderen", delete_response.data) self.assertEqual(report_response.status_code, 404) def test_sidebar_shows_gpx_combine_tool(self) -> None: client = self._logged_in_client() response = client.get("/wandelen") self.assertEqual(response.status_code, 200) self.assertIn(b"Tools", response.data) self.assertIn(b"GPX combineren", response.data) self.assertIn(b"/wandelen/tools/gpx-combine", response.data) self.assertIn(b"/wandelen/static/js/progress.js", response.data) def test_progress_indicator_asset_is_available(self) -> None: client = self._logged_in_client() response = client.get("/wandelen/static/js/progress.js") payload = response.get_data() response.close() self.assertEqual(response.status_code, 200) self.assertIn(b"Route laden", payload) self.assertIn(b"AmundsenProgress", payload) self.assertIn(b"doorgaan zonder indicator", payload) self.assertIn(b"is-waiting", payload) def test_report_print_layout_keeps_planning_and_waypoints_printable(self) -> None: template = Path("templates/report.html").read_text(encoding="utf-8") stylesheet = Path("static/css/styles.css").read_text(encoding="utf-8") report_script = Path("static/js/report.js").read_text(encoding="utf-8") self.assertIn('id="panel-planner" class="tab-panel"', template) self.assertIn('id="panel-waypoints" class="tab-panel"', template) self.assertLess(template.index('id="panel-planner"'), template.index('id="panel-waypoints"')) self.assertIn("@page", stylesheet) self.assertIn("size: A4 landscape", stylesheet) self.assertIn("page-orientation: landscape", stylesheet) self.assertNotIn("min-height: 190mm", stylesheet) self.assertIn("height: 80mm", stylesheet) self.assertIn("height: 52mm !important", stylesheet) self.assertIn("#panel-overview", stylesheet) self.assertIn("break-after: page", stylesheet) self.assertIn("#panel-planner", stylesheet) self.assertIn("break-before: auto", stylesheet) self.assertNotIn("break-before: page", stylesheet) self.assertIn("grid-template-columns: repeat(3, minmax(0, 1fr))", stylesheet) self.assertIn("grid-template-columns: 0.75fr 1.25fr 1.3fr 2.1fr", stylesheet) self.assertIn(".manual-planner", stylesheet) self.assertIn("display: none !important", stylesheet) self.assertIn(".day-plan-legs", stylesheet) self.assertIn('id="cachedMapImage"', template) self.assertIn('id="cachedRoute3dImage"', template) self.assertIn('id="cachedSlopeChartImage"', template) self.assertIn('id="cachedElevationChartImage"', template) self.assertIn('id="cachedElevationTimeChartImage"', template) self.assertIn(".has-cached-report-images", stylesheet) self.assertIn(".has-cached-report-image canvas", stylesheet) self.assertIn("function prepareReportPrint()", report_script) self.assertIn("function applyCachedReportImages()", report_script) self.assertIn("function findCachedReportImage", report_script) self.assertIn("report.report_images", report_script) self.assertIn('window.addEventListener("beforeprint", prepareReportPrint)', report_script) self.assertIn("map.invalidateSize()", report_script) self.assertIn('chart.update("none")', report_script) def test_gpx_combine_map_asset_is_available(self) -> None: client = self._logged_in_client() response = client.get("/wandelen/static/js/gpx_combine.js") payload = response.get_data() response.close() self.assertEqual(response.status_code, 200) self.assertIn(b"OpenTopoMap", payload) self.assertIn(b"Ontbrekend stuk", payload) self.assertIn(b"fitBounds", payload) self.assertIn(b"combine-route-row", payload) def test_gpx_combine_tool_previews_and_executes_smart_merge(self) -> None: client = self._logged_in_client() first_route = self._line_gpx("Route A", [(46.0000, 7.0000, 1000), (46.0010, 7.0000, 1010), (46.0020, 7.0000, 1020)]) second_route = self._line_gpx("Route B", [(46.0040, 7.0000, 1030), (46.0050, 7.0000, 1040)]) preview_response = client.post( "/wandelen/tools/gpx-combine", data={ "name": "Slimme route", "gpx_files": [ (BytesIO(second_route), "route-b.gpx"), (BytesIO(first_route), "route-a.gpx"), ], }, content_type="multipart/form-data", ) draft_token = re.search(rb'name="draft_token" value="([^"]+)"', preview_response.data).group(1).decode() project_before_execute = client.get("/wandelen/projects/default") self.assertEqual(preview_response.status_code, 200) self.assertIn(b"Voorvertoning", preview_response.data) self.assertLess(preview_response.data.index(b"Route A"), preview_response.data.index(b"Route B")) self.assertIn(b"Overlap verwijderd", preview_response.data) self.assertIn(b"combinePreviewMap", preview_response.data) self.assertIn(b"OpenTopoMap", preview_response.data) self.assertIn(b"Ontbrekend stuk", preview_response.data) self.assertIn(b"missing", preview_response.data) self.assertIn(b"combine-preview-card", preview_response.data) self.assertIn(b"combine-move-up", preview_response.data) self.assertIn(b"combine-reverse", preview_response.data) self.assertIn(b'name="source_index"', preview_response.data) self.assertIn(b'data-progress-download="true"', preview_response.data) self.assertNotIn(b"Slimme route", project_before_execute.data) files_before_execute = set(Path(TEST_UPLOAD_DIR).rglob("*.gpx")) response = client.post( "/wandelen/tools/gpx-combine/execute", data={ "draft_token": draft_token, "source_index": ["0", "1"], "reversed": ["1", "0"], }, ) self.assertEqual(response.status_code, 200) self.assertEqual(response.mimetype, "application/gpx+xml") self.assertIn(b"Slimme route", response.data) self.assertIn(b"Slim gecombineerd", response.data) self.assertNotIn(b"Dag 1:", response.data) self.assertLess(response.data.index(b'lat="46.00500000"'), response.data.index(b'lat="46.00400000"')) self.assertLess(response.data.index(b'lat="46.00400000"'), response.data.index(b'lat="46.00000000"')) project_response = client.get("/wandelen/projects/default") files_after_execute = set(Path(TEST_UPLOAD_DIR).rglob("*.gpx")) self.assertNotIn(b"Route A", project_response.data) self.assertNotIn(b"Route B", project_response.data) self.assertNotIn(b"Slimme route", project_response.data) self.assertEqual(files_before_execute, files_after_execute) def test_save_waypoints_persists_to_route_gpx(self) -> None: client = self._logged_in_client() client.post( "/wandelen/analyze", data={"gpx_files": (BytesIO(self._mini_gpx()), "mini.gpx")}, content_type="multipart/form-data", ) report_id = self._last_route_id() save_response = client.post( f"/wandelen/api/waypoints/{report_id}", json={ "waypoints": [ {"name": "Nieuwe hut", "lat": 46.001, "lon": 7.0, "elevation": 1010}, ] }, ) report_response = client.get(f"/wandelen/api/report/{report_id}") self.assertEqual(save_response.status_code, 200) self.assertEqual(report_response.get_json()["waypoints"][0]["name"], "Nieuwe hut") def test_database_connection_uses_timeouts(self) -> None: store = ProjectStore.__new__(ProjectStore) store.config = DatabaseConfig( host="db.example", port=3307, user="tester", password="secret", database="amundsen_test", connect_timeout=2, read_timeout=3, write_timeout=4, ) with patch("amundsen.core.project_store.pymysql.connect") as connect: store._connect() connect.assert_called_once_with( host="db.example", port=3307, user="tester", password="secret", database="amundsen_test", charset="utf8mb4", connect_timeout=2, read_timeout=3, write_timeout=4, ) if __name__ == "__main__": unittest.main()