root
403 lines
12 KiB
Python
403 lines
12 KiB
Python
"""
|
||
Tests for geo_service.py
|
||
|
||
Tests the geographic zone building and Overpass API integration.
|
||
"""
|
||
import pytest
|
||
import math
|
||
from unittest.mock import AsyncMock, patch, MagicMock
|
||
from pathlib import Path
|
||
from datetime import datetime, timedelta
|
||
import json
|
||
|
||
from services.geo_service import (
|
||
haversine,
|
||
get_sector_bounds,
|
||
get_cache_key,
|
||
get_cached_result,
|
||
save_to_cache,
|
||
calculate_road_length,
|
||
find_nearest_distance,
|
||
calculate_forest_coverage,
|
||
build_search_zones,
|
||
DIRECTIONS,
|
||
SEARCH_DISTANCES,
|
||
_build_search_distances,
|
||
CACHE_DIR
|
||
)
|
||
|
||
|
||
class TestHaversine:
|
||
"""Test haversine distance calculation."""
|
||
|
||
def test_same_point(self):
|
||
"""Test distance between same point is zero."""
|
||
distance = haversine(53.9, 27.5, 53.9, 27.5)
|
||
assert distance == pytest.approx(0.0, abs=0.01)
|
||
|
||
def test_known_distance(self):
|
||
"""Test known distance between cities."""
|
||
# Minsk to Brest approximately 350 km
|
||
minsk_lat, minsk_lon = 53.9, 27.5
|
||
brest_lat, brest_lon = 52.1, 23.7
|
||
|
||
distance = haversine(minsk_lat, minsk_lon, brest_lat, brest_lon)
|
||
|
||
# Should be around 350 km
|
||
assert 300 < distance < 400
|
||
|
||
def test_short_distance(self):
|
||
"""Test short distance calculation."""
|
||
# 1 km north
|
||
lat1, lon1 = 53.9, 27.5
|
||
lat2 = lat1 + 0.009 # ~1 km
|
||
lon2 = lon1
|
||
|
||
distance = haversine(lat1, lon1, lat2, lon2)
|
||
assert distance == pytest.approx(1.0, abs=0.1)
|
||
|
||
|
||
class TestSectorBounds:
|
||
"""Test sector boundary calculations."""
|
||
|
||
def test_north_sector(self):
|
||
"""Test north sector bounds."""
|
||
lat, lon = 53.9, 27.5
|
||
bounds = get_sector_bounds(lat, lon, "N", 1000)
|
||
|
||
min_lat, min_lon, max_lat, max_lon = bounds
|
||
|
||
# North sector should extend north
|
||
assert max_lat > lat
|
||
assert isinstance(min_lat, float)
|
||
assert isinstance(max_lat, float)
|
||
|
||
def test_all_directions(self):
|
||
"""Test all 8 directions return valid bounds."""
|
||
lat, lon = 53.9, 27.5
|
||
|
||
for direction in DIRECTIONS:
|
||
bounds = get_sector_bounds(lat, lon, direction, 1000)
|
||
min_lat, min_lon, max_lat, max_lon = bounds
|
||
|
||
assert min_lat < max_lat
|
||
assert min_lon < max_lon
|
||
assert all(isinstance(x, float) for x in bounds)
|
||
|
||
def test_different_radii(self):
|
||
"""Test different radii produce different bounds."""
|
||
lat, lon = 53.9, 27.5
|
||
|
||
bounds_500 = get_sector_bounds(lat, lon, "N", 500)
|
||
bounds_5000 = get_sector_bounds(lat, lon, "N", 5000)
|
||
|
||
# Larger radius should have larger bounds
|
||
assert (bounds_5000[2] - bounds_5000[0]) > (bounds_500[2] - bounds_500[0])
|
||
|
||
|
||
class TestCaching:
|
||
"""Test caching functionality."""
|
||
|
||
def test_cache_key_generation(self):
|
||
"""Test cache key is consistent."""
|
||
query = "test query"
|
||
key1 = get_cache_key(query)
|
||
key2 = get_cache_key(query)
|
||
|
||
assert key1 == key2
|
||
assert len(key1) == 32 # MD5 hash length
|
||
|
||
def test_cache_key_different_queries(self):
|
||
"""Test different queries produce different keys."""
|
||
key1 = get_cache_key("query 1")
|
||
key2 = get_cache_key("query 2")
|
||
|
||
assert key1 != key2
|
||
|
||
def test_save_and_get_cache(self):
|
||
"""Test saving and retrieving from cache."""
|
||
cache_key = "test_key_123"
|
||
test_data = {'elements': [{'id': 1, 'type': 'node'}]}
|
||
|
||
# Save to cache
|
||
save_to_cache(cache_key, test_data)
|
||
|
||
# Retrieve from cache
|
||
cached = get_cached_result(cache_key)
|
||
|
||
assert cached is not None
|
||
assert cached == test_data
|
||
|
||
# Cleanup
|
||
cache_file = CACHE_DIR / f"{cache_key}.json"
|
||
if cache_file.exists():
|
||
cache_file.unlink()
|
||
|
||
def test_cache_expiration(self):
|
||
"""Test cache expires after TTL."""
|
||
cache_key = "test_key_expired"
|
||
test_data = {'elements': []}
|
||
|
||
# Save to cache with old timestamp
|
||
CACHE_DIR.mkdir(exist_ok=True)
|
||
cache_file = CACHE_DIR / f"{cache_key}.json"
|
||
|
||
old_time = datetime.now() - timedelta(hours=25)
|
||
with open(cache_file, 'w') as f:
|
||
json.dump({
|
||
'timestamp': old_time.isoformat(),
|
||
'data': test_data
|
||
}, f)
|
||
|
||
# Should return None (expired)
|
||
cached = get_cached_result(cache_key)
|
||
assert cached is None
|
||
|
||
# Cleanup
|
||
if cache_file.exists():
|
||
cache_file.unlink()
|
||
|
||
def test_cache_not_found(self):
|
||
"""Test cache returns None for non-existent key."""
|
||
cached = get_cached_result("nonexistent_key_xyz")
|
||
assert cached is None
|
||
|
||
|
||
class TestRoadLength:
|
||
"""Test road length calculation."""
|
||
|
||
def test_empty_elements(self):
|
||
"""Test empty elements returns zero."""
|
||
length = calculate_road_length([])
|
||
assert length == 0.0
|
||
|
||
def test_single_way(self):
|
||
"""Test single way calculation."""
|
||
elements = [
|
||
{
|
||
'type': 'way',
|
||
'geometry': [
|
||
{'lat': 53.9, 'lon': 27.5},
|
||
{'lat': 53.91, 'lon': 27.5}
|
||
]
|
||
}
|
||
]
|
||
|
||
length = calculate_road_length(elements)
|
||
|
||
# Should be approximately 1.1 km
|
||
assert 0.5 < length < 2.0
|
||
|
||
def test_multiple_ways(self):
|
||
"""Test multiple ways are summed."""
|
||
elements = [
|
||
{
|
||
'type': 'way',
|
||
'geometry': [
|
||
{'lat': 53.9, 'lon': 27.5},
|
||
{'lat': 53.91, 'lon': 27.5}
|
||
]
|
||
},
|
||
{
|
||
'type': 'way',
|
||
'geometry': [
|
||
{'lat': 53.9, 'lon': 27.5},
|
||
{'lat': 53.9, 'lon': 27.51}
|
||
]
|
||
}
|
||
]
|
||
|
||
length = calculate_road_length(elements)
|
||
assert length > 0
|
||
|
||
def test_ignores_non_ways(self):
|
||
"""Test non-way elements are ignored."""
|
||
elements = [
|
||
{'type': 'node', 'lat': 53.9, 'lon': 27.5},
|
||
{
|
||
'type': 'way',
|
||
'geometry': [
|
||
{'lat': 53.9, 'lon': 27.5},
|
||
{'lat': 53.91, 'lon': 27.5}
|
||
]
|
||
}
|
||
]
|
||
|
||
length = calculate_road_length(elements)
|
||
assert length > 0
|
||
|
||
|
||
class TestNearestDistance:
|
||
"""Test nearest distance calculation."""
|
||
|
||
def test_empty_elements(self):
|
||
"""Test empty elements returns None."""
|
||
distance = find_nearest_distance(53.9, 27.5, [])
|
||
assert distance is None
|
||
|
||
def test_single_node(self):
|
||
"""Test single node distance."""
|
||
elements = [
|
||
{'type': 'node', 'lat': 53.91, 'lon': 27.5}
|
||
]
|
||
|
||
distance = find_nearest_distance(53.9, 27.5, elements)
|
||
|
||
assert distance is not None
|
||
assert distance > 0
|
||
|
||
def test_finds_nearest(self):
|
||
"""Test finds nearest among multiple nodes."""
|
||
elements = [
|
||
{'type': 'node', 'lat': 53.95, 'lon': 27.5}, # Far
|
||
{'type': 'node', 'lat': 53.901, 'lon': 27.5}, # Near
|
||
{'type': 'node', 'lat': 54.0, 'lon': 27.5} # Very far
|
||
]
|
||
|
||
distance = find_nearest_distance(53.9, 27.5, elements)
|
||
|
||
# Should find the nearest (53.901)
|
||
assert distance < 0.2
|
||
|
||
def test_ignores_non_nodes(self):
|
||
"""Test non-node elements are ignored."""
|
||
elements = [
|
||
{'type': 'way', 'geometry': []},
|
||
{'type': 'node', 'lat': 53.91, 'lon': 27.5}
|
||
]
|
||
|
||
distance = find_nearest_distance(53.9, 27.5, elements)
|
||
assert distance is not None
|
||
|
||
|
||
class TestForestCoverage:
|
||
"""Test forest coverage calculation."""
|
||
|
||
def test_no_forest(self):
|
||
"""Test no forest returns 0%."""
|
||
coverage = calculate_forest_coverage([], 1000)
|
||
assert coverage == 0.0
|
||
|
||
def test_some_forest(self):
|
||
"""Test forest coverage calculation."""
|
||
elements = [
|
||
{'type': 'way', 'tags': {'landuse': 'forest'}},
|
||
{'type': 'way', 'tags': {'natural': 'wood'}}
|
||
]
|
||
|
||
coverage = calculate_forest_coverage(elements, 1000)
|
||
|
||
assert 0 < coverage <= 100
|
||
|
||
def test_coverage_capped_at_100(self):
|
||
"""Test coverage is capped at 100%."""
|
||
# Many forest ways
|
||
elements = [{'type': 'way'} for _ in range(1000)]
|
||
|
||
coverage = calculate_forest_coverage(elements, 100)
|
||
|
||
assert coverage <= 100.0
|
||
|
||
|
||
class TestBuildSearchZones:
|
||
"""Test search zone building."""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_zone_count(self):
|
||
"""Test correct number of zones are created."""
|
||
with patch('services.geo_service.get_zone_features', new_callable=AsyncMock) as mock_features:
|
||
mock_features.return_value = {
|
||
'roads_km': 5.0,
|
||
'road_density': 2.0,
|
||
'water_distance_km': 1.5,
|
||
'settlement_distance_km': 3.0,
|
||
'forest_pct': 40.0
|
||
}
|
||
|
||
zones = await build_search_zones(53.9, 27.5, {})
|
||
|
||
# 8 directions × 4 distances = 32 zones
|
||
assert len(zones) == 32
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_all_directions_covered(self):
|
||
"""Test all 8 directions are included."""
|
||
with patch('services.geo_service.get_zone_features', new_callable=AsyncMock) as mock_features:
|
||
mock_features.return_value = {
|
||
'roads_km': 5.0,
|
||
'road_density': 2.0,
|
||
'water_distance_km': 1.5,
|
||
'settlement_distance_km': 3.0,
|
||
'forest_pct': 40.0
|
||
}
|
||
|
||
zones = await build_search_zones(53.9, 27.5, {})
|
||
|
||
directions_found = set(z.direction for z in zones)
|
||
assert directions_found == set(DIRECTIONS)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_all_distances_covered(self):
|
||
"""Test all 4 distances are included."""
|
||
with patch('services.geo_service.get_zone_features', new_callable=AsyncMock) as mock_features:
|
||
mock_features.return_value = {
|
||
'roads_km': 5.0,
|
||
'road_density': 2.0,
|
||
'water_distance_km': 1.5,
|
||
'settlement_distance_km': 3.0,
|
||
'forest_pct': 40.0
|
||
}
|
||
|
||
zones = await build_search_zones(53.9, 27.5, {})
|
||
|
||
distances_found = set(z.distance_km for z in zones)
|
||
# build_search_zones derives distances from max_distance_km (default 3.0 km),
|
||
# not from the legacy SEARCH_DISTANCES constant.
|
||
expected_distances = set(d / 1000 for d in _build_search_distances(3.0))
|
||
assert distances_found == expected_distances
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_zone_structure(self):
|
||
"""Test zone objects have correct structure."""
|
||
with patch('services.geo_service.get_zone_features', new_callable=AsyncMock) as mock_features:
|
||
mock_features.return_value = {
|
||
'roads_km': 5.0,
|
||
'road_density': 2.0,
|
||
'water_distance_km': 1.5,
|
||
'settlement_distance_km': 3.0,
|
||
'forest_pct': 40.0
|
||
}
|
||
|
||
zones = await build_search_zones(53.9, 27.5, {})
|
||
|
||
for zone in zones:
|
||
assert hasattr(zone, 'direction')
|
||
assert hasattr(zone, 'distance_km')
|
||
assert hasattr(zone, 'forest_pct')
|
||
assert hasattr(zone, 'road_density')
|
||
assert hasattr(zone, 'water_distance_km')
|
||
assert hasattr(zone, 'settlement_distance_km')
|
||
|
||
|
||
class TestConstants:
|
||
"""Test module constants."""
|
||
|
||
def test_directions_count(self):
|
||
"""Test there are 8 directions."""
|
||
assert len(DIRECTIONS) == 8
|
||
|
||
def test_directions_values(self):
|
||
"""Test direction values are correct."""
|
||
expected = ["N", "NE", "E", "SE", "S", "SW", "W", "NW"]
|
||
assert DIRECTIONS == expected
|
||
|
||
def test_search_distances(self):
|
||
"""Test search distances are correct."""
|
||
expected = [500, 1000, 2000, 5000]
|
||
assert SEARCH_DISTANCES == expected
|
||
|
||
def test_cache_dir_path(self):
|
||
"""Test cache directory path is set."""
|
||
assert isinstance(CACHE_DIR, Path)
|
||
assert str(CACHE_DIR) == "/tmp/overpass_cache"
|