"""Helper methods for testing.""" import time import logging from functools import wraps from fastapi import HTTPException from ..cache import client as cache_client from ..structs.landmark import Landmark from ..structs.preferences import Preferences, Preference 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(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.error(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) -> 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: landmark = fetch_landmark(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) def trip_params( sightseeing: int, shopping: int, nature: int, max_time_minute: int, start_coords: tuple[float, float] = None, end_coords: tuple[float, float] = None, ): def decorator(test_func): @wraps(test_func) def wrapper(client, request): prefs = Preferences( sightseeing=Preference(type='sightseeing', score=sightseeing), shopping=Preference(type='shopping', score=shopping), nature=Preference(type='nature', score=nature), max_time_minute=max_time_minute, detour_tolerance_minute=0, ) start = start_coords end = end_coords # Inject into test function return test_func(client, request, prefs, start, end) return wrapper return decorator