Add comprehensive tests for all geo GraphQL endpoints
All checks were successful
Build Docker Image / build (push) Successful in 1m22s
All checks were successful
Build Docker Image / build (push) Successful in 1m22s
Created test suite covering all 8 main geo service endpoints: - Basic: products, nodes (with filters/bounds), clusteredNodes - Nearest: nearestHubs, nearestOffers, nearestSuppliers (with product filters) - Routing: routeToCoordinate, autoRoute, railRoute - Edge cases: invalid coordinates, zero radius, nonexistent UUIDs Test suite uses real API calls to production GraphQL endpoint. 16 tests total across 4 test classes. Files: - tests/test_graphql_endpoints.py: Main test suite (600+ lines) - tests/README.md: Documentation and usage guide - pytest.ini: Pytest configuration - run_tests.sh: Convenience script to run tests - pyproject.toml: Added pytest and requests as dev dependencies
This commit is contained in:
@@ -22,6 +22,10 @@ dependencies = [
|
||||
[tool.poetry]
|
||||
package-mode = false
|
||||
|
||||
[tool.poetry.group.dev.dependencies]
|
||||
pytest = "^8.0.0"
|
||||
requests = "^2.32.0"
|
||||
|
||||
[build-system]
|
||||
requires = ["poetry-core>=2.0.0,<3.0.0"]
|
||||
build-backend = "poetry.core.masonry.api"
|
||||
|
||||
12
pytest.ini
Normal file
12
pytest.ini
Normal file
@@ -0,0 +1,12 @@
|
||||
[pytest]
|
||||
testpaths = tests
|
||||
python_files = test_*.py
|
||||
python_classes = Test*
|
||||
python_functions = test_*
|
||||
addopts =
|
||||
-v
|
||||
--tb=short
|
||||
--strict-markers
|
||||
markers =
|
||||
slow: marks tests as slow (deselect with '-m "not slow"')
|
||||
integration: marks tests as integration tests
|
||||
36
run_tests.sh
Executable file
36
run_tests.sh
Executable file
@@ -0,0 +1,36 @@
|
||||
#!/bin/bash
|
||||
# Run geo service GraphQL endpoint tests
|
||||
|
||||
set -e
|
||||
|
||||
cd "$(dirname "$0")"
|
||||
|
||||
echo "🧪 Running Geo Service GraphQL Tests"
|
||||
echo "===================================="
|
||||
echo ""
|
||||
|
||||
# Check if TEST_GEO_URL is set, otherwise use production
|
||||
if [ -z "$TEST_GEO_URL" ]; then
|
||||
export TEST_GEO_URL="https://geo.optovia.ru/graphql/public/"
|
||||
echo "📍 Testing against: $TEST_GEO_URL (production)"
|
||||
else
|
||||
echo "📍 Testing against: $TEST_GEO_URL"
|
||||
fi
|
||||
|
||||
echo ""
|
||||
|
||||
# Install dependencies if needed
|
||||
if ! poetry run python -c "import pytest" 2>/dev/null; then
|
||||
echo "📦 Installing dependencies..."
|
||||
poetry install --with dev
|
||||
echo ""
|
||||
fi
|
||||
|
||||
# Run tests
|
||||
echo "🚀 Running tests..."
|
||||
echo ""
|
||||
|
||||
poetry run pytest tests/test_graphql_endpoints.py -v -s "$@"
|
||||
|
||||
echo ""
|
||||
echo "✅ Test run complete"
|
||||
130
tests/README.md
Normal file
130
tests/README.md
Normal file
@@ -0,0 +1,130 @@
|
||||
# Geo Service Tests
|
||||
|
||||
Comprehensive test suite for all GraphQL endpoints in the geo service.
|
||||
|
||||
## Test Coverage
|
||||
|
||||
### Basic Endpoints (4 tests)
|
||||
- `test_products_query` - List all unique products
|
||||
- `test_nodes_query_basic` - List hubs/nodes without filters
|
||||
- `test_nodes_query_with_filters` - Filter nodes by transport type and country
|
||||
- `test_nodes_query_with_bounds` - Filter nodes by geographic bounds
|
||||
- `test_clustered_nodes_query` - Map clustering for visualization
|
||||
|
||||
### Nearest Endpoints (6 tests)
|
||||
- `test_nearest_hubs` - Find hubs near coordinates
|
||||
- `test_nearest_hubs_with_product_filter` - Find hubs with specific product
|
||||
- `test_nearest_offers` - Find offers near coordinates
|
||||
- `test_nearest_offers_with_product_filter` - Find offers for specific product
|
||||
- `test_nearest_suppliers` - Find suppliers near coordinates
|
||||
- `test_nearest_suppliers_with_product_filter` - Find suppliers with product
|
||||
|
||||
### Routing Endpoints (3 tests)
|
||||
- `test_route_to_coordinate` - Multi-hop route from offer to destination
|
||||
- `test_auto_route` - Road route between coordinates (requires OSRM)
|
||||
- `test_rail_route` - Rail route between coordinates
|
||||
|
||||
### Edge Cases (3 tests)
|
||||
- `test_nearest_with_zero_radius` - Very small search radius
|
||||
- `test_invalid_coordinates` - Invalid lat/lon values
|
||||
- `test_nonexistent_uuid` - Non-existent offer UUID
|
||||
|
||||
**Total: 16 tests covering 8 main endpoints**
|
||||
|
||||
## Running Tests
|
||||
|
||||
### Local Testing (against production)
|
||||
|
||||
```bash
|
||||
cd backends/geo
|
||||
poetry install
|
||||
poetry run pytest tests/test_graphql_endpoints.py -v
|
||||
```
|
||||
|
||||
### Testing against different endpoint
|
||||
|
||||
```bash
|
||||
export TEST_GEO_URL=https://geo-staging.example.com/graphql/public/
|
||||
poetry run pytest tests/test_graphql_endpoints.py -v
|
||||
```
|
||||
|
||||
### Run specific test class
|
||||
|
||||
```bash
|
||||
poetry run pytest tests/test_graphql_endpoints.py::TestNearestEndpoints -v
|
||||
```
|
||||
|
||||
### Run single test
|
||||
|
||||
```bash
|
||||
poetry run pytest tests/test_graphql_endpoints.py::TestNearestEndpoints::test_nearest_offers -v
|
||||
```
|
||||
|
||||
### Show print output
|
||||
|
||||
```bash
|
||||
poetry run pytest tests/test_graphql_endpoints.py -v -s
|
||||
```
|
||||
|
||||
## CI Integration
|
||||
|
||||
Tests should be run on each deployment:
|
||||
|
||||
```yaml
|
||||
# .gitea/workflows/test.yml
|
||||
- name: Run geo endpoint tests
|
||||
run: |
|
||||
cd backends/geo
|
||||
poetry install
|
||||
export TEST_GEO_URL=https://geo.optovia.ru/graphql/public/
|
||||
poetry run pytest tests/test_graphql_endpoints.py -v
|
||||
```
|
||||
|
||||
## Test Data Requirements
|
||||
|
||||
Tests use real data from the production/staging database. Required data:
|
||||
- At least one product in `products` collection
|
||||
- At least one hub node with coordinates
|
||||
- At least one offer with coordinates
|
||||
- Graph edges for routing tests
|
||||
|
||||
## Expected Test Results
|
||||
|
||||
All tests should pass on production environment. Some tests may be skipped if:
|
||||
- No products exist: `test_nearest_hubs_with_product_filter`, `test_nearest_offers_with_product_filter`
|
||||
- No offers exist: `test_route_to_coordinate`
|
||||
- OSRM not configured: `test_auto_route`, `test_rail_route` (warnings, not failures)
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### All nearest* tests return 0 results
|
||||
|
||||
Check that nodes collection has documents with:
|
||||
- Valid `latitude` and `longitude` fields (not null)
|
||||
- Correct `node_type` field (`'hub'`, `'offer'`, `'supplier'`)
|
||||
|
||||
Query ArangoDB directly:
|
||||
|
||||
```javascript
|
||||
// Count offers with coordinates
|
||||
db._query(`
|
||||
FOR node IN nodes
|
||||
FILTER node.node_type == 'offer'
|
||||
FILTER node.latitude != null AND node.longitude != null
|
||||
RETURN node
|
||||
`).toArray().length
|
||||
```
|
||||
|
||||
### Test failures with 400 errors
|
||||
|
||||
Check GraphQL schema matches test queries. GraphQL validation errors indicate:
|
||||
- Missing required arguments
|
||||
- Wrong argument types
|
||||
- Invalid field names
|
||||
|
||||
### Connection errors
|
||||
|
||||
Verify:
|
||||
- TEST_GEO_URL points to correct endpoint
|
||||
- Endpoint is accessible (not behind VPN/firewall)
|
||||
- GraphQL endpoint is `/graphql/public/` not `/graphql/`
|
||||
1
tests/__init__.py
Normal file
1
tests/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
# Geo service tests
|
||||
672
tests/test_graphql_endpoints.py
Normal file
672
tests/test_graphql_endpoints.py
Normal file
@@ -0,0 +1,672 @@
|
||||
"""
|
||||
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]}")
|
||||
|
||||
|
||||
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'])
|
||||
Reference in New Issue
Block a user