feat: Enhance track model and import functionality
- Added new fields to TrackModel: status, is_bidirectional, and coordinates. - Updated network service to handle new track attributes and geometry extraction. - Introduced CLI scripts for importing and loading tracks from OpenStreetMap. - Implemented normalization of track elements to ensure valid geometries. - Enhanced tests for track model, network service, and import/load scripts. - Updated frontend to accommodate new track attributes and improve route computation. - Documented OSM ingestion process in architecture and runtime views.
This commit is contained in:
168
backend/tests/test_tracks_load.py
Normal file
168
backend/tests/test_tracks_load.py
Normal file
@@ -0,0 +1,168 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import List
|
||||
|
||||
import pytest
|
||||
from geoalchemy2.shape import from_shape
|
||||
from shapely.geometry import Point
|
||||
|
||||
from backend.scripts import tracks_load
|
||||
|
||||
|
||||
def test_parse_track_entries_returns_models() -> None:
|
||||
entries = [
|
||||
{
|
||||
"name": "Connector",
|
||||
"coordinates": [[52.5, 13.4], [52.6, 13.5]],
|
||||
"lengthMeters": 1500,
|
||||
"maxSpeedKph": 120,
|
||||
"status": "operational",
|
||||
"isBidirectional": True,
|
||||
}
|
||||
]
|
||||
|
||||
parsed = tracks_load._parse_track_entries(entries)
|
||||
|
||||
assert parsed[0].name == "Connector"
|
||||
assert parsed[0].coordinates[0] == (52.5, 13.4)
|
||||
assert parsed[0].length_meters == 1500
|
||||
assert parsed[0].max_speed_kph == 120
|
||||
|
||||
|
||||
def test_parse_track_entries_invalid_raises_value_error() -> None:
|
||||
entries = [
|
||||
{
|
||||
"coordinates": [[52.5, 13.4]],
|
||||
}
|
||||
]
|
||||
|
||||
with pytest.raises(ValueError):
|
||||
tracks_load._parse_track_entries(entries)
|
||||
|
||||
|
||||
@dataclass
|
||||
class DummySession:
|
||||
committed: bool = False
|
||||
rolled_back: bool = False
|
||||
|
||||
def __enter__(self) -> "DummySession":
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc, traceback) -> None:
|
||||
pass
|
||||
|
||||
def commit(self) -> None:
|
||||
self.committed = True
|
||||
|
||||
def rollback(self) -> None:
|
||||
self.rolled_back = True
|
||||
|
||||
|
||||
@dataclass
|
||||
class DummyStation:
|
||||
id: str
|
||||
location: object
|
||||
|
||||
|
||||
@dataclass
|
||||
class DummyStationRepository:
|
||||
session: DummySession
|
||||
stations: List[DummyStation]
|
||||
|
||||
def list_active(self) -> List[DummyStation]:
|
||||
return self.stations
|
||||
|
||||
|
||||
@dataclass
|
||||
class DummyTrackRepository:
|
||||
session: DummySession
|
||||
created: list = field(default_factory=list)
|
||||
existing: list = field(default_factory=list)
|
||||
|
||||
def list_all(self):
|
||||
return self.existing
|
||||
|
||||
def create(self, data): # pragma: no cover - simple delegation
|
||||
self.created.append(data)
|
||||
|
||||
|
||||
def _point(lat: float, lon: float) -> object:
|
||||
return from_shape(Point(lon, lat), srid=4326)
|
||||
|
||||
|
||||
def test_load_tracks_creates_entries(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
session_instance = DummySession()
|
||||
station_repo_instance = DummyStationRepository(
|
||||
session_instance,
|
||||
stations=[
|
||||
DummyStation(id="station-a", location=_point(52.5, 13.4)),
|
||||
DummyStation(id="station-b", location=_point(52.6, 13.5)),
|
||||
],
|
||||
)
|
||||
track_repo_instance = DummyTrackRepository(session_instance)
|
||||
|
||||
monkeypatch.setattr(tracks_load, "SessionLocal", lambda: session_instance)
|
||||
monkeypatch.setattr(tracks_load, "StationRepository",
|
||||
lambda session: station_repo_instance)
|
||||
monkeypatch.setattr(tracks_load, "TrackRepository",
|
||||
lambda session: track_repo_instance)
|
||||
|
||||
parsed = tracks_load._parse_track_entries(
|
||||
[
|
||||
{
|
||||
"name": "Connector",
|
||||
"coordinates": [[52.5, 13.4], [52.6, 13.5]],
|
||||
}
|
||||
]
|
||||
)
|
||||
|
||||
created = tracks_load.load_tracks(parsed, commit=True)
|
||||
|
||||
assert created == 1
|
||||
assert session_instance.committed is True
|
||||
assert track_repo_instance.created
|
||||
track = track_repo_instance.created[0]
|
||||
assert track.start_station_id == "station-a"
|
||||
assert track.end_station_id == "station-b"
|
||||
assert track.coordinates == [(52.5, 13.4), (52.6, 13.5)]
|
||||
|
||||
|
||||
def test_load_tracks_skips_existing_pairs(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
session_instance = DummySession()
|
||||
station_repo_instance = DummyStationRepository(
|
||||
session_instance,
|
||||
stations=[
|
||||
DummyStation(id="station-a", location=_point(52.5, 13.4)),
|
||||
DummyStation(id="station-b", location=_point(52.6, 13.5)),
|
||||
],
|
||||
)
|
||||
existing_track = type("ExistingTrack", (), {
|
||||
"start_station_id": "station-a",
|
||||
"end_station_id": "station-b",
|
||||
})
|
||||
track_repo_instance = DummyTrackRepository(
|
||||
session_instance,
|
||||
existing=[existing_track],
|
||||
)
|
||||
|
||||
monkeypatch.setattr(tracks_load, "SessionLocal", lambda: session_instance)
|
||||
monkeypatch.setattr(tracks_load, "StationRepository",
|
||||
lambda session: station_repo_instance)
|
||||
monkeypatch.setattr(tracks_load, "TrackRepository",
|
||||
lambda session: track_repo_instance)
|
||||
|
||||
parsed = tracks_load._parse_track_entries(
|
||||
[
|
||||
{
|
||||
"name": "Connector",
|
||||
"coordinates": [[52.5, 13.4], [52.6, 13.5]],
|
||||
}
|
||||
]
|
||||
)
|
||||
|
||||
created = tracks_load.load_tracks(parsed, commit=False)
|
||||
|
||||
assert created == 0
|
||||
assert session_instance.rolled_back is True
|
||||
assert not track_repo_instance.created
|
||||
Reference in New Issue
Block a user