Helldragon67
11bbf34375
Some checks failed
Build and deploy the backend to staging / Build and push image (pull_request) Successful in 2m20s
Run linting on the backend code / Build (pull_request) Failing after 26s
Run testing on the backend code / Build (pull_request) Failing after 2m40s
Build and deploy the backend to staging / Deploy to staging (pull_request) Successful in 17s
135 lines
4.6 KiB
Python
135 lines
4.6 KiB
Python
"""Helper methods for testing."""
|
|
import logging
|
|
from fastapi import HTTPException
|
|
from pydantic import ValidationError
|
|
|
|
from ..structs.landmark import Landmark
|
|
from ..cache import client as cache_client
|
|
|
|
|
|
def landmarks_to_osmid(landmarks: list[Landmark]) -> list[int] :
|
|
"""
|
|
Convert the list of landmarks into a list containing their osm ids for quick landmark checking.
|
|
|
|
Args :
|
|
landmarks (list): the list of landmarks
|
|
|
|
Returns :
|
|
ids (list) : the list of corresponding OSM ids
|
|
"""
|
|
ids = []
|
|
for landmark in landmarks :
|
|
ids.append(landmark.osm_id)
|
|
|
|
return ids
|
|
|
|
def fetch_landmark(client, landmark_uuid: str):
|
|
"""
|
|
Fetch landmark data from the API based on the landmark UUID.
|
|
|
|
Args:
|
|
landmark_uuid (str): The UUID of the landmark.
|
|
|
|
Returns:
|
|
dict: Landmark data fetched from the API.
|
|
"""
|
|
logger = logging.getLogger(__name__)
|
|
response = client.get(f'/landmark/{landmark_uuid}')
|
|
|
|
if response.status_code != 200:
|
|
raise HTTPException(status_code=500,
|
|
detail=f'Failed to fetch landmark with UUID {landmark_uuid}: {response.status_code}')
|
|
|
|
try:
|
|
json_data = response.json()
|
|
logger.info(f'API Response: {json_data}')
|
|
except ValueError as e:
|
|
logger.error(f'Failed to parse response as JSON: {response.text}')
|
|
raise HTTPException(status_code=500, detail="Invalid response format from API")
|
|
|
|
# Try validating against the Landmark model here to ensure consistency
|
|
try:
|
|
landmark = Landmark(**json_data)
|
|
except ValidationError as ve:
|
|
logging.error(f'Validation error: {ve}')
|
|
raise HTTPException(status_code=500, detail="Invalid data format received from API")
|
|
|
|
if "detail" in json_data:
|
|
raise HTTPException(status_code=500, detail=json_data["detail"])
|
|
|
|
return Landmark(**json_data)
|
|
|
|
|
|
def fetch_landmark_cache(landmark_uuid: str):
|
|
"""
|
|
Fetch landmark data from the cache based on the landmark UUID.
|
|
|
|
Args:
|
|
landmark_uuid (str): The UUID of the landmark.
|
|
|
|
Returns:
|
|
dict: Landmark data fetched from the cache or raises an HTTP exception.
|
|
"""
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Try to fetch the landmark data from the cache
|
|
try:
|
|
landmark = cache_client.get(f'landmark_{landmark_uuid}')
|
|
if not landmark :
|
|
logger.warning(f'Cache miss for landmark UUID: {landmark_uuid}')
|
|
raise HTTPException(status_code=404, detail=f'Landmark with UUID {landmark_uuid} not found in cache.')
|
|
|
|
# Validate that the fetched data is a dictionary
|
|
if not isinstance(landmark, Landmark):
|
|
logger.error(f'Invalid cache data format for landmark UUID: {landmark_uuid}. Expected dict, got {type(landmark).__name__}.')
|
|
raise HTTPException(status_code=500, detail="Invalid cache data format.")
|
|
|
|
return landmark
|
|
|
|
except Exception as exc:
|
|
logger.error(f'Unexpected error occurred while fetching landmark UUID {landmark_uuid}: {exc}')
|
|
raise HTTPException(status_code=500, detail="An unexpected error occurred while fetching the landmark from the cache") from exc
|
|
|
|
|
|
def load_trip_landmarks(client, first_uuid: str, from_cache=None) -> list[Landmark]:
|
|
"""
|
|
Load all landmarks for a trip using the response from the API.
|
|
|
|
Args:
|
|
first_uuid (str) : The first UUID of the landmark.
|
|
|
|
Returns:
|
|
landmarks (list) : An list containing all landmarks for the trip.
|
|
"""
|
|
landmarks = []
|
|
next_uuid = first_uuid
|
|
|
|
while next_uuid is not None:
|
|
if from_cache :
|
|
landmark = fetch_landmark_cache(next_uuid)
|
|
else :
|
|
landmark = fetch_landmark(client, next_uuid)
|
|
|
|
landmarks.append(landmark)
|
|
next_uuid = landmark.next_uuid # Prepare for the next iteration
|
|
|
|
return landmarks
|
|
|
|
|
|
def log_trip_details(request, landmarks: list[Landmark], duration: int, target_duration: int) :
|
|
"""
|
|
Allows to show the detailed trip in the html test report.
|
|
|
|
Args:
|
|
request:
|
|
landmarks (list): the ordered list of visited landmarks
|
|
duration (int): the total duration of this trip
|
|
target_duration(int): the target duration of this trip
|
|
"""
|
|
trip_string = [f'{landmark.name} ({landmark.attractiveness} | {landmark.duration}) - {landmark.time_to_reach_next}' for landmark in landmarks]
|
|
|
|
# Pass additional info to pytest for reporting
|
|
request.node.trip_details = trip_string
|
|
request.node.trip_duration = str(duration) # result['total_time']
|
|
request.node.target_duration = str(target_duration)
|