249 lines
9.3 KiB
Python
249 lines
9.3 KiB
Python
"""Unit тесты для OSM GPS-источника (ET-008).
|
||
|
||
U-42: split_bbox_for_osm разбивает правильно
|
||
U-43: длина через Haversine
|
||
U-44: защита от XXE через defusedxml
|
||
"""
|
||
import os
|
||
import pytest
|
||
|
||
from src.api.gps_tracks.sources.osm import (
|
||
OsmParser,
|
||
split_bbox_for_osm,
|
||
_haversine_m,
|
||
_parse_gpx_trackpoints,
|
||
)
|
||
|
||
|
||
# ─── U-42: split_bbox_for_osm ────────────────────────────────────────────────
|
||
|
||
def test_u42_split_bbox_basic():
|
||
"""U-42: корректное разбиение на ячейки."""
|
||
bbox = (37.0, 55.0, 38.0, 56.0) # 1° x 1°
|
||
cells = split_bbox_for_osm(bbox, cell_size=0.25)
|
||
|
||
# 1° / 0.25° = 4 ячейки по каждой оси = 16 ячеек
|
||
assert len(cells) == 16
|
||
|
||
|
||
def test_u42_split_bbox_cell_size():
|
||
"""U-42: каждая ячейка не больше cell_size по размеру."""
|
||
bbox = (29.0, 49.5, 47.5, 60.0) # ЦФО
|
||
cells = split_bbox_for_osm(bbox, cell_size=0.25)
|
||
|
||
for cell in cells:
|
||
west, south, east, north = cell
|
||
assert east - west <= 0.25 + 1e-9
|
||
assert north - south <= 0.25 + 1e-9
|
||
|
||
|
||
def test_u42_split_bbox_covers_region():
|
||
"""U-42: все ячейки вместе покрывают весь регион."""
|
||
bbox = (37.0, 55.0, 38.0, 56.0)
|
||
cells = split_bbox_for_osm(bbox, cell_size=0.25)
|
||
|
||
min_lon = min(c[0] for c in cells)
|
||
min_lat = min(c[1] for c in cells)
|
||
max_lon = max(c[2] for c in cells)
|
||
max_lat = max(c[3] for c in cells)
|
||
|
||
assert abs(min_lon - 37.0) < 1e-9
|
||
assert abs(min_lat - 55.0) < 1e-9
|
||
assert abs(max_lon - 38.0) < 0.25 + 1e-9 # последняя ячейка обрезается
|
||
assert abs(max_lat - 56.0) < 0.25 + 1e-9
|
||
|
||
|
||
def test_u42_split_small_bbox():
|
||
"""U-42: bbox меньше cell_size даёт одну ячейку."""
|
||
bbox = (37.0, 55.0, 37.1, 55.1)
|
||
cells = split_bbox_for_osm(bbox, cell_size=0.25)
|
||
assert len(cells) == 1
|
||
|
||
|
||
def test_u42_split_bbox_no_overlap():
|
||
"""U-42: ячейки не перекрываются (west следующей = east предыдущей)."""
|
||
bbox = (37.0, 55.0, 37.5, 55.25)
|
||
cells = split_bbox_for_osm(bbox, cell_size=0.25)
|
||
|
||
# При bbox шириной 0.5° и cell_size=0.25 должно быть 2 ячейки по оси lon
|
||
assert len(cells) == 2
|
||
# Восток первой ячейки = запад второй
|
||
cells_sorted = sorted(cells, key=lambda c: c[0])
|
||
assert abs(cells_sorted[0][2] - cells_sorted[1][0]) < 1e-9
|
||
|
||
|
||
# ─── U-43: Haversine длина ───────────────────────────────────────────────────
|
||
|
||
def test_u43_haversine_known_distance():
|
||
"""U-43: проверка haversine на известном расстоянии."""
|
||
# Москва (37.617, 55.755) → Химки (37.425, 55.889) ≈ 20 км
|
||
dist = _haversine_m(37.617, 55.755, 37.425, 55.889)
|
||
assert 18000 < dist < 22000
|
||
|
||
|
||
def test_u43_haversine_zero_distance():
|
||
"""U-43: одна точка → расстояние 0."""
|
||
dist = _haversine_m(37.617, 55.755, 37.617, 55.755)
|
||
assert dist == pytest.approx(0.0, abs=1e-6)
|
||
|
||
|
||
def test_u43_haversine_symmetry():
|
||
"""U-43: расстояние A→B = B→A."""
|
||
d1 = _haversine_m(37.617, 55.755, 37.425, 55.889)
|
||
d2 = _haversine_m(37.425, 55.889, 37.617, 55.755)
|
||
assert abs(d1 - d2) < 1e-6
|
||
|
||
|
||
def test_u43_haversine_short_distance():
|
||
"""U-43: короткое расстояние (~111 м на экваторе при 0.001° по lon)."""
|
||
dist = _haversine_m(0.0, 0.0, 0.001, 0.0)
|
||
assert 100 < dist < 120
|
||
|
||
|
||
# ─── U-44: защита от XXE ─────────────────────────────────────────────────────
|
||
|
||
def test_u44_xxe_protection():
|
||
"""U-44: defusedxml блокирует XXE атаку."""
|
||
fixture_path = os.path.join(
|
||
os.path.dirname(__file__),
|
||
"../../tests/fixtures/gps-tracks/xxe-payload.gpx",
|
||
)
|
||
|
||
with open(fixture_path, "rb") as f:
|
||
content = f.read()
|
||
|
||
# Должен либо выбросить исключение, либо вернуть пустой список без чтения /etc/passwd
|
||
try:
|
||
tracks = _parse_gpx_trackpoints(content, "osm", "")
|
||
# Если парсинг прошёл без ошибки — проверяем что /etc/passwd не попал в данные
|
||
for track in tracks:
|
||
assert "root:" not in str(track)
|
||
assert "/bin/" not in str(track)
|
||
except Exception:
|
||
# defusedxml выбросил исключение — это ожидаемое поведение
|
||
pass
|
||
|
||
|
||
def test_u44_valid_gpx_parsed_correctly():
|
||
"""U-44: корректный GPX с gpx_id парсится правильно."""
|
||
fixture_path = os.path.join(
|
||
os.path.dirname(__file__),
|
||
"../../tests/fixtures/gps-tracks/osm-trackpoints-bbox-moscow.gpx",
|
||
)
|
||
|
||
with open(fixture_path, "rb") as f:
|
||
content = f.read()
|
||
|
||
tracks = _parse_gpx_trackpoints(content, "osm", "")
|
||
|
||
assert len(tracks) == 1
|
||
track = tracks[0]
|
||
assert track.points_count == 3
|
||
assert abs(track.min_lat - 55.751) < 0.001
|
||
assert abs(track.max_lat - 55.753) < 0.001
|
||
assert track.source_id == "osm"
|
||
|
||
|
||
def test_u44_anonymous_trackpoints_skipped():
|
||
"""U-44: анонимные точки без gpx_id пропускаются."""
|
||
gpx_without_ids = b"""<?xml version="1.0"?>
|
||
<gpx version="1.0" creator="OpenStreetMap.org" xmlns="http://www.topografix.com/GPX/1/0">
|
||
<trk>
|
||
<trkseg>
|
||
<trkpt lat="55.751" lon="37.618"><time>2024-05-12T10:00:00Z</time></trkpt>
|
||
<trkpt lat="55.752" lon="37.619"><time>2024-05-12T10:01:00Z</time></trkpt>
|
||
</trkseg>
|
||
</trk>
|
||
</gpx>"""
|
||
|
||
tracks = _parse_gpx_trackpoints(gpx_without_ids, "osm", "")
|
||
assert len(tracks) == 0
|
||
|
||
|
||
def test_u44_multiple_tracks_in_gpx():
|
||
"""U-44: несколько gpx_id в одном ответе парсятся как разные треки."""
|
||
gpx_multi = b"""<?xml version="1.0"?>
|
||
<gpx version="1.0" creator="OpenStreetMap.org" xmlns="http://www.topografix.com/GPX/1/0">
|
||
<trk>
|
||
<trkseg>
|
||
<trkpt lat="55.751" lon="37.618" gpx_id="111"><time>2024-05-12T10:00:00Z</time></trkpt>
|
||
<trkpt lat="55.752" lon="37.619" gpx_id="111"><time>2024-05-12T10:01:00Z</time></trkpt>
|
||
<trkpt lat="55.760" lon="37.700" gpx_id="222"><time>2024-05-13T08:00:00Z</time></trkpt>
|
||
<trkpt lat="55.765" lon="37.710" gpx_id="222"><time>2024-05-13T08:05:00Z</time></trkpt>
|
||
</trkseg>
|
||
</trk>
|
||
</gpx>"""
|
||
|
||
tracks = _parse_gpx_trackpoints(gpx_multi, "osm", "")
|
||
assert len(tracks) == 2
|
||
|
||
ids = {t.external_id for t in tracks}
|
||
assert "111" in ids
|
||
assert "222" in ids
|
||
|
||
|
||
# ─── U-45: _parse_gpx_meta_response ──────────────────────────────────────────
|
||
|
||
def test_u45_meta_response_with_known_tag():
|
||
"""U-45: _parse_gpx_meta_response extracts activity via MAPPING."""
|
||
from src.api.gps_tracks.sources.osm import _parse_gpx_meta_response
|
||
content = b"""<?xml version="1.0"?>
|
||
<osm version="0.6">
|
||
<gpx_file id="123" name="my_ride.gpx" user="alice">
|
||
<description>Weekend ride</description>
|
||
<tag>enduro</tag>
|
||
<tag>motorcycle</tag>
|
||
</gpx_file>
|
||
</osm>"""
|
||
meta = _parse_gpx_meta_response(content)
|
||
assert meta is not None
|
||
assert meta["activity_type"] == "enduro"
|
||
assert meta["name"] == "my_ride.gpx"
|
||
assert meta["user"] == "alice"
|
||
assert meta["description"] == "Weekend ride"
|
||
|
||
|
||
def test_u45_meta_response_unknown_tag_returns_none_activity():
|
||
"""U-45: unknown tag → activity_type is None."""
|
||
from src.api.gps_tracks.sources.osm import _parse_gpx_meta_response
|
||
content = b"""<?xml version="1.0"?>
|
||
<osm version="0.6">
|
||
<gpx_file id="99" name="trip.gpx" user="bob">
|
||
<tag>unknown-sport</tag>
|
||
</gpx_file>
|
||
</osm>"""
|
||
meta = _parse_gpx_meta_response(content)
|
||
assert meta is not None
|
||
assert meta["activity_type"] is None
|
||
|
||
|
||
def test_u45_meta_response_motorcycle_maps_to_moto():
|
||
"""U-45: 'motorcycle' tag maps to 'moto' via MAPPING."""
|
||
from src.api.gps_tracks.sources.osm import _parse_gpx_meta_response
|
||
content = b"""<?xml version="1.0"?>
|
||
<osm version="0.6">
|
||
<gpx_file id="77" name="ride.gpx" user="carl">
|
||
<tag>motorcycle</tag>
|
||
</gpx_file>
|
||
</osm>"""
|
||
meta = _parse_gpx_meta_response(content)
|
||
assert meta["activity_type"] == "moto"
|
||
|
||
|
||
def test_u45_meta_response_invalid_xml_returns_none():
|
||
"""U-45: malformed XML returns None."""
|
||
from src.api.gps_tracks.sources.osm import _parse_gpx_meta_response
|
||
meta = _parse_gpx_meta_response(b"not xml at all <<<")
|
||
assert meta is None
|
||
|
||
|
||
def test_u45_meta_response_no_gpx_file_element():
|
||
"""U-45: valid XML but no gpx_file element → result has all None values."""
|
||
from src.api.gps_tracks.sources.osm import _parse_gpx_meta_response
|
||
content = b"""<?xml version="1.0"?><osm version="0.6"></osm>"""
|
||
meta = _parse_gpx_meta_response(content)
|
||
# Function should return the dict with None values, not None itself
|
||
assert meta is not None
|
||
assert meta["activity_type"] is None
|
||
assert meta["name"] is None
|