845 lines
29 KiB
Python
845 lines
29 KiB
Python
"""
|
|
Comprehensive tests for all Geo GraphQL endpoints.
|
|
Tests use real API calls to production/staging GraphQL endpoint.
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import requests
|
|
import pytest
|
|
|
|
# GraphQL endpoint - override with TEST_GEO_URL env var
|
|
GEO_URL = os.getenv('TEST_GEO_URL', 'https://geo.optovia.ru/graphql/public/')
|
|
|
|
|
|
class TestBasicEndpoints:
|
|
"""Test basic list/query endpoints."""
|
|
|
|
def test_products_query(self):
|
|
"""Test products query - should return list of unique products."""
|
|
query = """
|
|
query GetProducts {
|
|
products {
|
|
uuid
|
|
name
|
|
offersCount
|
|
}
|
|
}
|
|
"""
|
|
response = requests.post(GEO_URL, json={'query': query})
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.text}"
|
|
|
|
data = response.json()
|
|
assert 'errors' not in data, f"GraphQL errors: {data.get('errors')}"
|
|
assert 'data' in data
|
|
assert 'products' in data['data']
|
|
|
|
products = data['data']['products']
|
|
assert isinstance(products, list), "products should be a list"
|
|
if len(products) > 0:
|
|
product = products[0]
|
|
assert 'uuid' in product
|
|
assert 'name' in product
|
|
assert 'offersCount' in product
|
|
assert isinstance(product['offersCount'], int)
|
|
|
|
print(f"✓ products query: {len(products)} products found")
|
|
|
|
def test_nodes_query_basic(self):
|
|
"""Test nodes query without filters."""
|
|
query = """
|
|
query GetNodes($limit: Int, $offset: Int) {
|
|
nodes(limit: $limit, offset: $offset) {
|
|
uuid
|
|
name
|
|
latitude
|
|
longitude
|
|
country
|
|
transportTypes
|
|
}
|
|
nodesCount
|
|
}
|
|
"""
|
|
variables = {'limit': 10, 'offset': 0}
|
|
|
|
response = requests.post(GEO_URL, json={'query': query, 'variables': variables})
|
|
assert response.status_code == 200, f"Expected 200, got {response.status_code}: {response.text}"
|
|
|
|
data = response.json()
|
|
assert 'errors' not in data, f"GraphQL errors: {data.get('errors')}"
|
|
|
|
nodes = data['data']['nodes']
|
|
count = data['data']['nodesCount']
|
|
|
|
assert isinstance(nodes, list)
|
|
assert isinstance(count, int)
|
|
assert count > 0, "Should have at least some nodes in database"
|
|
assert len(nodes) <= 10, "Should respect limit"
|
|
|
|
if len(nodes) > 0:
|
|
node = nodes[0]
|
|
assert 'uuid' in node
|
|
assert 'name' in node
|
|
# Coordinates might be null for some nodes
|
|
assert 'latitude' in node
|
|
assert 'longitude' in node
|
|
|
|
print(f"✓ nodes query: {len(nodes)}/{count} nodes found")
|
|
|
|
def test_nodes_query_with_filters(self):
|
|
"""Test nodes query with transport type and country filters."""
|
|
query = """
|
|
query GetNodes($transportType: String, $country: String, $limit: Int) {
|
|
nodes(transportType: $transportType, country: $country, limit: $limit) {
|
|
uuid
|
|
name
|
|
country
|
|
transportTypes
|
|
}
|
|
nodesCount(transportType: $transportType, country: $country)
|
|
}
|
|
"""
|
|
variables = {'transportType': 'sea', 'limit': 5}
|
|
|
|
response = requests.post(GEO_URL, json={'query': query, 'variables': variables})
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert 'errors' not in data, f"GraphQL errors: {data.get('errors')}"
|
|
|
|
nodes = data['data']['nodes']
|
|
# All nodes should have 'sea' in their transportTypes
|
|
for node in nodes:
|
|
if node.get('transportTypes'):
|
|
assert 'sea' in node['transportTypes'], f"Node {node['uuid']} missing 'sea' transport type"
|
|
|
|
print(f"✓ nodes query with filters: {len(nodes)} sea nodes found")
|
|
|
|
def test_nodes_query_with_bounds(self):
|
|
"""Test nodes query with geographic bounds."""
|
|
query = """
|
|
query GetNodes($west: Float, $south: Float, $east: Float, $north: Float, $limit: Int) {
|
|
nodes(west: $west, south: $south, east: $east, north: $north, limit: $limit) {
|
|
uuid
|
|
name
|
|
latitude
|
|
longitude
|
|
}
|
|
}
|
|
"""
|
|
# Bounds for central Europe
|
|
variables = {
|
|
'west': 5.0,
|
|
'south': 45.0,
|
|
'east': 15.0,
|
|
'north': 55.0,
|
|
'limit': 20
|
|
}
|
|
|
|
response = requests.post(GEO_URL, json={'query': query, 'variables': variables})
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert 'errors' not in data, f"GraphQL errors: {data.get('errors')}"
|
|
|
|
nodes = data['data']['nodes']
|
|
# Verify all nodes are within bounds
|
|
for node in nodes:
|
|
if node.get('latitude') and node.get('longitude'):
|
|
lat = float(node['latitude'])
|
|
lon = float(node['longitude'])
|
|
assert variables['south'] <= lat <= variables['north'], \
|
|
f"Node {node['uuid']} latitude {lat} outside bounds"
|
|
assert variables['west'] <= lon <= variables['east'], \
|
|
f"Node {node['uuid']} longitude {lon} outside bounds"
|
|
|
|
print(f"✓ nodes with bounds: {len(nodes)} nodes in central Europe")
|
|
|
|
def test_clustered_nodes_query(self):
|
|
"""Test clusteredNodes query for map clustering."""
|
|
query = """
|
|
query GetClusteredNodes($west: Float!, $south: Float!, $east: Float!, $north: Float!, $zoom: Int!) {
|
|
clusteredNodes(west: $west, south: $south, east: $east, north: $north, zoom: $zoom) {
|
|
id
|
|
latitude
|
|
longitude
|
|
count
|
|
expansionZoom
|
|
name
|
|
}
|
|
}
|
|
"""
|
|
# World view
|
|
variables = {
|
|
'west': -180.0,
|
|
'south': -90.0,
|
|
'east': 180.0,
|
|
'north': 90.0,
|
|
'zoom': 2
|
|
}
|
|
|
|
response = requests.post(GEO_URL, json={'query': query, 'variables': variables})
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert 'errors' not in data, f"GraphQL errors: {data.get('errors')}"
|
|
|
|
clusters = data['data']['clusteredNodes']
|
|
assert isinstance(clusters, list)
|
|
assert len(clusters) > 0, "Should have clusters at zoom level 2"
|
|
|
|
for cluster in clusters:
|
|
assert 'id' in cluster
|
|
assert 'latitude' in cluster
|
|
assert 'longitude' in cluster
|
|
assert 'count' in cluster
|
|
assert cluster['count'] >= 1
|
|
|
|
print(f"✓ clusteredNodes: {len(clusters)} clusters/points at zoom 2")
|
|
|
|
|
|
class TestNearestEndpoints:
|
|
"""Test new coordinate-based 'nearest' endpoints."""
|
|
|
|
def test_nearest_hubs(self):
|
|
"""Test nearestHubs query - find hubs near coordinates."""
|
|
query = """
|
|
query NearestHubs($lat: Float!, $lon: Float!, $radius: Float, $limit: Int) {
|
|
nearestHubs(lat: $lat, lon: $lon, radius: $radius, limit: $limit) {
|
|
uuid
|
|
name
|
|
latitude
|
|
longitude
|
|
country
|
|
transportTypes
|
|
distanceKm
|
|
}
|
|
}
|
|
"""
|
|
# Rotterdam coordinates (major European port)
|
|
variables = {
|
|
'lat': 51.9244,
|
|
'lon': 4.4777,
|
|
'radius': 200,
|
|
'limit': 5
|
|
}
|
|
|
|
response = requests.post(GEO_URL, json={'query': query, 'variables': variables})
|
|
assert response.status_code == 200, f"Status: {response.status_code}, Body: {response.text}"
|
|
|
|
data = response.json()
|
|
assert 'errors' not in data, f"GraphQL errors: {data.get('errors')}"
|
|
|
|
hubs = data['data']['nearestHubs']
|
|
assert isinstance(hubs, list)
|
|
|
|
# Should find some hubs in 200km radius of Rotterdam
|
|
if len(hubs) > 0:
|
|
hub = hubs[0]
|
|
assert 'uuid' in hub
|
|
assert 'name' in hub
|
|
assert 'distanceKm' in hub
|
|
assert hub['distanceKm'] <= 200, f"Hub {hub['uuid']} distance {hub['distanceKm']} exceeds radius"
|
|
|
|
# Verify hubs are sorted by distance
|
|
distances = [h['distanceKm'] for h in hubs]
|
|
assert distances == sorted(distances), "Hubs should be sorted by distance"
|
|
|
|
print(f"✓ nearestHubs: {len(hubs)} hubs near Rotterdam")
|
|
|
|
def test_nearest_hubs_with_product_filter(self):
|
|
"""Test nearestHubs with product filter."""
|
|
# First get a product UUID
|
|
products_query = "query { products { uuid } }"
|
|
prod_response = requests.post(GEO_URL, json={'query': products_query})
|
|
products = prod_response.json()['data']['products']
|
|
|
|
if not products:
|
|
pytest.skip("No products in database")
|
|
|
|
product_uuid = products[0]['uuid']
|
|
|
|
query = """
|
|
query NearestHubs($lat: Float!, $lon: Float!, $radius: Float, $productUuid: String) {
|
|
nearestHubs(lat: $lat, lon: $lon, radius: $radius, productUuid: $productUuid) {
|
|
uuid
|
|
name
|
|
distanceKm
|
|
}
|
|
}
|
|
"""
|
|
variables = {
|
|
'lat': 50.0,
|
|
'lon': 10.0,
|
|
'radius': 1000,
|
|
'productUuid': product_uuid
|
|
}
|
|
|
|
response = requests.post(GEO_URL, json={'query': query, 'variables': variables})
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert 'errors' not in data, f"GraphQL errors: {data.get('errors')}"
|
|
|
|
hubs = data['data']['nearestHubs']
|
|
print(f"✓ nearestHubs with product filter: {len(hubs)} hubs for product {product_uuid[:8]}")
|
|
|
|
def test_nearest_offers(self):
|
|
"""Test nearestOffers query - find offers near coordinates."""
|
|
query = """
|
|
query NearestOffers($lat: Float!, $lon: Float!, $radius: Float, $limit: Int) {
|
|
nearestOffers(lat: $lat, lon: $lon, radius: $radius, limit: $limit) {
|
|
uuid
|
|
productUuid
|
|
productName
|
|
latitude
|
|
longitude
|
|
pricePerUnit
|
|
currency
|
|
distanceKm
|
|
}
|
|
}
|
|
"""
|
|
# Central Europe
|
|
variables = {
|
|
'lat': 50.0,
|
|
'lon': 10.0,
|
|
'radius': 500,
|
|
'limit': 10
|
|
}
|
|
|
|
response = requests.post(GEO_URL, json={'query': query, 'variables': variables})
|
|
assert response.status_code == 200, f"Status: {response.status_code}, Body: {response.text}"
|
|
|
|
data = response.json()
|
|
assert 'errors' not in data, f"GraphQL errors: {data.get('errors')}"
|
|
|
|
offers = data['data']['nearestOffers']
|
|
assert isinstance(offers, list)
|
|
|
|
if len(offers) > 0:
|
|
offer = offers[0]
|
|
assert 'uuid' in offer
|
|
assert 'productUuid' in offer
|
|
assert 'distanceKm' in offer
|
|
assert offer['distanceKm'] <= 500
|
|
|
|
# Verify offers are sorted by distance
|
|
distances = [o['distanceKm'] for o in offers]
|
|
assert distances == sorted(distances), "Offers should be sorted by distance"
|
|
|
|
print(f"✓ nearestOffers: {len(offers)} offers in Central Europe")
|
|
|
|
def test_nearest_offers_with_product_filter(self):
|
|
"""Test nearestOffers with product UUID filter."""
|
|
# First get a product UUID
|
|
products_query = "query { products { uuid name } }"
|
|
prod_response = requests.post(GEO_URL, json={'query': products_query})
|
|
products = prod_response.json()['data']['products']
|
|
|
|
if not products:
|
|
pytest.skip("No products in database")
|
|
|
|
product_uuid = products[0]['uuid']
|
|
product_name = products[0]['name']
|
|
|
|
query = """
|
|
query NearestOffers($lat: Float!, $lon: Float!, $radius: Float, $productUuid: String) {
|
|
nearestOffers(lat: $lat, lon: $lon, radius: $radius, productUuid: $productUuid) {
|
|
uuid
|
|
productUuid
|
|
productName
|
|
distanceKm
|
|
}
|
|
}
|
|
"""
|
|
# Global search with large radius
|
|
variables = {
|
|
'lat': 0.0,
|
|
'lon': 0.0,
|
|
'radius': 20000,
|
|
'productUuid': product_uuid
|
|
}
|
|
|
|
response = requests.post(GEO_URL, json={'query': query, 'variables': variables})
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert 'errors' not in data, f"GraphQL errors: {data.get('errors')}"
|
|
|
|
offers = data['data']['nearestOffers']
|
|
# All offers should be for the requested product
|
|
for offer in offers:
|
|
assert offer['productUuid'] == product_uuid, \
|
|
f"Offer {offer['uuid']} has wrong product UUID"
|
|
|
|
print(f"✓ nearestOffers with product: {len(offers)} offers for '{product_name}'")
|
|
|
|
def test_nearest_suppliers(self):
|
|
"""Test nearestSuppliers query - find suppliers near coordinates."""
|
|
query = """
|
|
query NearestSuppliers($lat: Float!, $lon: Float!, $radius: Float, $limit: Int) {
|
|
nearestSuppliers(lat: $lat, lon: $lon, radius: $radius, limit: $limit) {
|
|
uuid
|
|
name
|
|
latitude
|
|
longitude
|
|
distanceKm
|
|
}
|
|
}
|
|
"""
|
|
variables = {
|
|
'lat': 52.52, # Berlin
|
|
'lon': 13.405,
|
|
'radius': 300,
|
|
'limit': 10
|
|
}
|
|
|
|
response = requests.post(GEO_URL, json={'query': query, 'variables': variables})
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert 'errors' not in data, f"GraphQL errors: {data.get('errors')}"
|
|
|
|
suppliers = data['data']['nearestSuppliers']
|
|
assert isinstance(suppliers, list)
|
|
|
|
if len(suppliers) > 0:
|
|
supplier = suppliers[0]
|
|
assert 'uuid' in supplier
|
|
assert 'name' in supplier
|
|
assert 'distanceKm' in supplier
|
|
assert supplier['distanceKm'] <= 300
|
|
|
|
# Verify sorted by distance
|
|
distances = [s['distanceKm'] for s in suppliers]
|
|
assert distances == sorted(distances)
|
|
|
|
print(f"✓ nearestSuppliers: {len(suppliers)} suppliers near Berlin")
|
|
|
|
def test_nearest_suppliers_with_product_filter(self):
|
|
"""Test nearestSuppliers with product filter."""
|
|
# Get a product UUID
|
|
products_query = "query { products { uuid } }"
|
|
prod_response = requests.post(GEO_URL, json={'query': products_query})
|
|
products = prod_response.json()['data']['products']
|
|
|
|
if not products:
|
|
pytest.skip("No products in database")
|
|
|
|
product_uuid = products[0]['uuid']
|
|
|
|
query = """
|
|
query NearestSuppliers($lat: Float!, $lon: Float!, $radius: Float, $productUuid: String) {
|
|
nearestSuppliers(lat: $lat, lon: $lon, radius: $radius, productUuid: $productUuid) {
|
|
uuid
|
|
name
|
|
distanceKm
|
|
}
|
|
}
|
|
"""
|
|
variables = {
|
|
'lat': 50.0,
|
|
'lon': 10.0,
|
|
'radius': 1000,
|
|
'productUuid': product_uuid
|
|
}
|
|
|
|
response = requests.post(GEO_URL, json={'query': query, 'variables': variables})
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert 'errors' not in data, f"GraphQL errors: {data.get('errors')}"
|
|
|
|
suppliers = data['data']['nearestSuppliers']
|
|
print(f"✓ nearestSuppliers with product: {len(suppliers)} suppliers for product {product_uuid[:8]}")
|
|
|
|
def test_nearest_offers_with_hub_uuid(self):
|
|
"""Test nearestOffers with hubUuid - should return offers with calculated routes.
|
|
|
|
This tests the fix for the bug where resolve_offer_to_hub was called incorrectly
|
|
(self was None in graphene resolvers).
|
|
"""
|
|
# First, get a hub UUID from the database
|
|
hubs_query = """
|
|
query {
|
|
nearestHubs(lat: 0, lon: 0, radius: 20000, limit: 5) {
|
|
uuid
|
|
name
|
|
latitude
|
|
longitude
|
|
}
|
|
}
|
|
"""
|
|
hubs_response = requests.post(GEO_URL, json={'query': hubs_query})
|
|
hubs_data = hubs_response.json()
|
|
|
|
if not hubs_data.get('data', {}).get('nearestHubs'):
|
|
pytest.skip("No hubs found in database")
|
|
|
|
hub = hubs_data['data']['nearestHubs'][0]
|
|
hub_uuid = hub['uuid']
|
|
hub_lat = hub['latitude']
|
|
hub_lon = hub['longitude']
|
|
|
|
# Now test nearestOffers with this hub UUID
|
|
query = """
|
|
query NearestOffers($lat: Float!, $lon: Float!, $radius: Float, $hubUuid: String, $limit: Int) {
|
|
nearestOffers(lat: $lat, lon: $lon, radius: $radius, hubUuid: $hubUuid, limit: $limit) {
|
|
uuid
|
|
productUuid
|
|
productName
|
|
supplierUuid
|
|
supplierName
|
|
latitude
|
|
longitude
|
|
pricePerUnit
|
|
currency
|
|
distanceKm
|
|
routes {
|
|
totalDistanceKm
|
|
totalTimeSeconds
|
|
stages {
|
|
fromUuid
|
|
fromName
|
|
fromLat
|
|
fromLon
|
|
toUuid
|
|
toName
|
|
toLat
|
|
toLon
|
|
distanceKm
|
|
travelTimeSeconds
|
|
transportType
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
# Search around the hub location with large radius
|
|
variables = {
|
|
'lat': float(hub_lat) if hub_lat else 0.0,
|
|
'lon': float(hub_lon) if hub_lon else 0.0,
|
|
'radius': 5000, # 5000km radius to find offers
|
|
'hubUuid': hub_uuid,
|
|
'limit': 10
|
|
}
|
|
|
|
response = requests.post(GEO_URL, json={'query': query, 'variables': variables})
|
|
assert response.status_code == 200, f"Status: {response.status_code}, Body: {response.text}"
|
|
|
|
data = response.json()
|
|
assert 'errors' not in data, f"GraphQL errors: {data.get('errors')}"
|
|
|
|
offers = data['data']['nearestOffers']
|
|
assert isinstance(offers, list), "nearestOffers should return a list"
|
|
|
|
# The key assertion: with hubUuid, we should get offers with routes calculated
|
|
# (This was the bug - resolve_offer_to_hub was failing silently)
|
|
print(f"✓ nearestOffers with hubUuid: {len(offers)} offers for hub '{hub['name']}'")
|
|
|
|
if len(offers) > 0:
|
|
# Check first offer structure
|
|
offer = offers[0]
|
|
assert 'uuid' in offer
|
|
assert 'productUuid' in offer
|
|
assert 'routes' in offer, "Offer should have routes field when hubUuid is provided"
|
|
|
|
# If routes exist, verify structure
|
|
if offer['routes'] and len(offer['routes']) > 0:
|
|
route = offer['routes'][0]
|
|
assert 'totalDistanceKm' in route
|
|
assert 'totalTimeSeconds' in route
|
|
assert 'stages' in route
|
|
|
|
if route['stages'] and len(route['stages']) > 0:
|
|
stage = route['stages'][0]
|
|
assert 'fromUuid' in stage
|
|
assert 'toUuid' in stage
|
|
assert 'transportType' in stage
|
|
assert 'distanceKm' in stage
|
|
print(f" Route has {len(route['stages'])} stages, total {route['totalDistanceKm']:.1f}km")
|
|
|
|
def test_nearest_offers_with_hub_and_product(self):
|
|
"""Test nearestOffers with both hubUuid and productUuid filters."""
|
|
# Get a product and hub
|
|
products_query = "query { products { uuid name } }"
|
|
prod_response = requests.post(GEO_URL, json={'query': products_query})
|
|
products = prod_response.json().get('data', {}).get('products', [])
|
|
|
|
hubs_query = """
|
|
query {
|
|
nearestHubs(lat: 0, lon: 0, radius: 20000, limit: 1) {
|
|
uuid
|
|
name
|
|
latitude
|
|
longitude
|
|
}
|
|
}
|
|
"""
|
|
hubs_response = requests.post(GEO_URL, json={'query': hubs_query})
|
|
hubs = hubs_response.json().get('data', {}).get('nearestHubs', [])
|
|
|
|
if not products or not hubs:
|
|
pytest.skip("No products or hubs in database")
|
|
|
|
product = products[0]
|
|
hub = hubs[0]
|
|
|
|
query = """
|
|
query NearestOffers($lat: Float!, $lon: Float!, $radius: Float, $productUuid: String, $hubUuid: String, $limit: Int) {
|
|
nearestOffers(lat: $lat, lon: $lon, radius: $radius, productUuid: $productUuid, hubUuid: $hubUuid, limit: $limit) {
|
|
uuid
|
|
productUuid
|
|
productName
|
|
routes {
|
|
totalDistanceKm
|
|
stages {
|
|
transportType
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
variables = {
|
|
'lat': float(hub['latitude']) if hub['latitude'] else 0.0,
|
|
'lon': float(hub['longitude']) if hub['longitude'] else 0.0,
|
|
'radius': 10000,
|
|
'productUuid': product['uuid'],
|
|
'hubUuid': hub['uuid'],
|
|
'limit': 5 # Limit to avoid timeout when calculating routes
|
|
}
|
|
|
|
# Use longer timeout as route calculation takes time
|
|
response = requests.post(GEO_URL, json={'query': query, 'variables': variables}, timeout=120)
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert 'errors' not in data, f"GraphQL errors: {data.get('errors')}"
|
|
|
|
offers = data['data']['nearestOffers']
|
|
|
|
# All offers should be for the requested product
|
|
for offer in offers:
|
|
assert offer['productUuid'] == product['uuid'], \
|
|
f"Offer has wrong productUuid: {offer['productUuid']} != {product['uuid']}"
|
|
|
|
print(f"✓ nearestOffers with hub+product: {len(offers)} offers for '{product['name']}' via hub '{hub['name']}'")
|
|
|
|
|
|
class TestRoutingEndpoints:
|
|
"""Test routing and pathfinding endpoints."""
|
|
|
|
def test_route_to_coordinate(self):
|
|
"""Test routeToCoordinate query - find route from offer to coordinates."""
|
|
# First, get an offer UUID with coordinates
|
|
offers_query = """
|
|
query {
|
|
nearestOffers(lat: 50.0, lon: 10.0, radius: 1000, limit: 1) {
|
|
uuid
|
|
latitude
|
|
longitude
|
|
}
|
|
}
|
|
"""
|
|
offers_response = requests.post(GEO_URL, json={'query': offers_query})
|
|
offers_data = offers_response.json()
|
|
|
|
if not offers_data.get('data', {}).get('nearestOffers'):
|
|
pytest.skip("No offers found for routing test")
|
|
|
|
offer = offers_data['data']['nearestOffers'][0]
|
|
offer_uuid = offer['uuid']
|
|
|
|
query = """
|
|
query RouteToCoordinate($offerUuid: String!, $lat: Float!, $lon: Float!) {
|
|
routeToCoordinate(offerUuid: $offerUuid, lat: $lat, lon: $lon) {
|
|
offerUuid
|
|
distanceKm
|
|
routes {
|
|
totalDistanceKm
|
|
totalTimeSeconds
|
|
stages {
|
|
fromUuid
|
|
fromName
|
|
fromLat
|
|
fromLon
|
|
toUuid
|
|
toName
|
|
toLat
|
|
toLon
|
|
distanceKm
|
|
travelTimeSeconds
|
|
transportType
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
# Route to Amsterdam
|
|
variables = {
|
|
'offerUuid': offer_uuid,
|
|
'lat': 52.3676,
|
|
'lon': 4.9041
|
|
}
|
|
|
|
response = requests.post(GEO_URL, json={'query': query, 'variables': variables})
|
|
assert response.status_code == 200, f"Status: {response.status_code}, Body: {response.text}"
|
|
|
|
data = response.json()
|
|
assert 'errors' not in data, f"GraphQL errors: {data.get('errors')}"
|
|
|
|
route_data = data['data']['routeToCoordinate']
|
|
assert route_data is not None
|
|
assert route_data['offerUuid'] == offer_uuid
|
|
|
|
if route_data.get('routes'):
|
|
route = route_data['routes'][0]
|
|
assert 'totalDistanceKm' in route
|
|
assert 'totalTimeSeconds' in route
|
|
assert 'stages' in route
|
|
assert len(route['stages']) > 0
|
|
|
|
# Verify each stage has required fields
|
|
for stage in route['stages']:
|
|
assert 'fromUuid' in stage
|
|
assert 'toUuid' in stage
|
|
assert 'distanceKm' in stage
|
|
assert 'transportType' in stage
|
|
|
|
print(f"✓ routeToCoordinate: {len(route['stages'])} stages, {route['totalDistanceKm']:.1f}km")
|
|
else:
|
|
print(f"✓ routeToCoordinate: no routes found (offer may be isolated)")
|
|
|
|
def test_auto_route(self):
|
|
"""Test autoRoute query - calculate road route between coordinates."""
|
|
query = """
|
|
query AutoRoute($fromLat: Float!, $fromLon: Float!, $toLat: Float!, $toLon: Float!) {
|
|
autoRoute(fromLat: $fromLat, fromLon: $fromLon, toLat: $toLat, toLon: $toLon) {
|
|
distanceKm
|
|
geometry
|
|
}
|
|
}
|
|
"""
|
|
# Route from Amsterdam to Rotterdam
|
|
variables = {
|
|
'fromLat': 52.3676,
|
|
'fromLon': 4.9041,
|
|
'toLat': 51.9244,
|
|
'toLon': 4.4777
|
|
}
|
|
|
|
response = requests.post(GEO_URL, json={'query': query, 'variables': variables})
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
# Auto route might not be available (requires OSRM service)
|
|
if 'errors' in data:
|
|
print("⚠ autoRoute: service not available (expected if OSRM not configured)")
|
|
else:
|
|
route = data['data']['autoRoute']
|
|
if route:
|
|
assert 'distanceKm' in route
|
|
assert route['distanceKm'] > 0
|
|
print(f"✓ autoRoute: {route['distanceKm']:.1f}km Amsterdam → Rotterdam")
|
|
else:
|
|
print("⚠ autoRoute: returned null")
|
|
|
|
def test_rail_route(self):
|
|
"""Test railRoute query - calculate rail route between coordinates."""
|
|
query = """
|
|
query RailRoute($fromLat: Float!, $fromLon: Float!, $toLat: Float!, $toLon: Float!) {
|
|
railRoute(fromLat: $fromLat, fromLon: $fromLon, toLat: $toLat, toLon: $toLon) {
|
|
distanceKm
|
|
geometry
|
|
}
|
|
}
|
|
"""
|
|
variables = {
|
|
'fromLat': 52.3676,
|
|
'fromLon': 4.9041,
|
|
'toLat': 51.9244,
|
|
'toLon': 4.4777
|
|
}
|
|
|
|
response = requests.post(GEO_URL, json={'query': query, 'variables': variables})
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
# Rail route might not be available
|
|
if 'errors' in data:
|
|
print("⚠ railRoute: service not available")
|
|
else:
|
|
route = data['data']['railRoute']
|
|
if route:
|
|
assert 'distanceKm' in route
|
|
print(f"✓ railRoute: {route['distanceKm']:.1f}km")
|
|
else:
|
|
print("⚠ railRoute: returned null")
|
|
|
|
|
|
class TestEdgeCases:
|
|
"""Test edge cases and error handling."""
|
|
|
|
def test_nearest_with_zero_radius(self):
|
|
"""Test nearest queries with very small radius."""
|
|
query = """
|
|
query NearestHubs($lat: Float!, $lon: Float!, $radius: Float) {
|
|
nearestHubs(lat: $lat, lon: $lon, radius: $radius) {
|
|
uuid
|
|
}
|
|
}
|
|
"""
|
|
variables = {'lat': 50.0, 'lon': 10.0, 'radius': 0.001}
|
|
|
|
response = requests.post(GEO_URL, json={'query': query, 'variables': variables})
|
|
assert response.status_code == 200
|
|
|
|
data = response.json()
|
|
assert 'errors' not in data
|
|
# Should return empty list or very few results
|
|
hubs = data['data']['nearestHubs']
|
|
assert isinstance(hubs, list)
|
|
print(f"✓ nearestHubs with tiny radius: {len(hubs)} hubs")
|
|
|
|
def test_invalid_coordinates(self):
|
|
"""Test behavior with invalid latitude/longitude values."""
|
|
query = """
|
|
query NearestHubs($lat: Float!, $lon: Float!) {
|
|
nearestHubs(lat: $lat, lon: $lon) {
|
|
uuid
|
|
}
|
|
}
|
|
"""
|
|
# Latitude > 90 is invalid
|
|
variables = {'lat': 100.0, 'lon': 10.0}
|
|
|
|
response = requests.post(GEO_URL, json={'query': query, 'variables': variables})
|
|
# Should either return error or empty results
|
|
assert response.status_code in [200, 400]
|
|
print("✓ invalid coordinates handled")
|
|
|
|
def test_nonexistent_uuid(self):
|
|
"""Test route query with non-existent offer UUID."""
|
|
query = """
|
|
query RouteToCoordinate($offerUuid: String!, $lat: Float!, $lon: Float!) {
|
|
routeToCoordinate(offerUuid: $offerUuid, lat: $lat, lon: $lon) {
|
|
offerUuid
|
|
}
|
|
}
|
|
"""
|
|
variables = {
|
|
'offerUuid': 'nonexistent-uuid-12345',
|
|
'lat': 50.0,
|
|
'lon': 10.0
|
|
}
|
|
|
|
response = requests.post(GEO_URL, json={'query': query, 'variables': variables})
|
|
# Should handle gracefully (null result or error)
|
|
assert response.status_code in [200, 400]
|
|
print("✓ nonexistent UUID handled")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
pytest.main([__file__, '-v', '-s'])
|