diff --git a/.env.example b/.env.example index 575da779..8281161b 100644 --- a/.env.example +++ b/.env.example @@ -64,3 +64,10 @@ # SUPADATA_API_KEY= # -- LinkedIn -- # LINKEDIN_ACCESS_TOKEN= + +# -- App Business Reviews -- +# APP_BUSINESS_REVIEWS_API_KEY= + +# -- Gong -- +# GONG_ACCESS_KEY= +# GONG_ACCESS_KEY_SECRET= diff --git a/app-business-reviews/app_business_reviews.py b/app-business-reviews/app_business_reviews.py index d735ba17..1d238b5f 100644 --- a/app-business-reviews/app_business_reviews.py +++ b/app-business-reviews/app_business_reviews.py @@ -1,4 +1,10 @@ -from autohive_integrations_sdk import Integration, ExecutionContext, ActionHandler +from autohive_integrations_sdk import ( + Integration, + ExecutionContext, + ActionHandler, + ActionResult, + ActionError, +) from typing import Dict, Any # Create the integration using the config.json @@ -10,10 +16,14 @@ @app_business_reviews.action("search_apps_ios") class SearchAppsIOS(ActionHandler): async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): - api_key = context.auth.get("credentials", {}).get("api_key", {}) + api_key = context.auth.get("credentials", {}).get("api_key", "") # Build SerpApi request parameters for app search - params = {"api_key": api_key, "engine": "apple_app_store", "term": inputs["term"]} + params = { + "api_key": api_key, + "engine": "apple_app_store", + "term": inputs["term"], + } # Add optional parameters if inputs.get("country"): @@ -29,7 +39,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): response = await context.fetch("https://serpapi.com/search", method="GET", params=params) # Extract apps from organic results - organic_results = response.get("organic_results", []) + organic_results = response.data.get("organic_results", []) limit = inputs.get("num", 10) apps = [] @@ -48,24 +58,29 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): } apps.append(app) - return {"apps": apps, "total_results": len(apps)} + return ActionResult(data={"apps": apps, "total_results": len(apps)}, cost_usd=0.0) @app_business_reviews.action("get_reviews_app_store") class GetReviewsAppStore(ActionHandler): async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): - api_key = context.auth.get("credentials", {}).get("api_key", {}) + api_key = context.auth.get("credentials", {}).get("api_key", "") # Get product_id - either provided directly or search by app name product_id = inputs.get("product_id") app_name = inputs.get("app_name") if not product_id and not app_name: - raise ValueError("Either product_id or app_name must be provided") + return ActionError(message="Either product_id or app_name must be provided") # If app_name is provided but no product_id, search for the app first if app_name and not product_id: - search_params = {"api_key": api_key, "engine": "apple_app_store", "term": app_name, "num": 1} + search_params = { + "api_key": api_key, + "engine": "apple_app_store", + "term": app_name, + "num": 1, + } # Add country if provided if inputs.get("country"): @@ -73,19 +88,23 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): search_response = await context.fetch("https://serpapi.com/search", method="GET", params=search_params) - organic_results = search_response.get("organic_results", []) + organic_results = search_response.data.get("organic_results", []) if not organic_results: - raise ValueError(f"No apps found for search term: {app_name}") + return ActionError(message=f"No apps found for search term: {app_name}") # Get the first result's product ID first_result = organic_results[0] product_id = str(first_result.get("id")) if not product_id: - raise ValueError(f"Could not extract product ID for app: {app_name}") + return ActionError(message=f"Could not extract product ID for app: {app_name}") # Build SerpApi request parameters for reviews - params = {"api_key": api_key, "engine": "apple_reviews", "product_id": product_id} + params = { + "api_key": api_key, + "engine": "apple_reviews", + "product_id": product_id, + } # Add optional filters if inputs.get("country"): @@ -110,7 +129,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): response = await context.fetch("https://serpapi.com/search", method="GET", params=params) # Extract reviews data from current page - page_reviews = response.get("reviews", []) + page_reviews = response.data.get("reviews", []) if not page_reviews: break @@ -139,16 +158,19 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): current_page += 1 # Check if there are more pages using pagination info - pagination_info = response.get("serpapi_pagination", {}) + pagination_info = response.data.get("serpapi_pagination", {}) if not pagination_info.get("next"): break - return { - "reviews": all_reviews, - "total_reviews": len(all_reviews), - "app_name": app_title, - "product_id": product_id, - } + return ActionResult( + data={ + "reviews": all_reviews, + "total_reviews": len(all_reviews), + "app_name": app_title, + "product_id": product_id, + }, + cost_usd=0.0, + ) # ---- Google Play Store Actions ---- @@ -157,16 +179,21 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): @app_business_reviews.action("search_apps_android") class SearchAppsAndroid(ActionHandler): async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): - api_key = context.auth.get("credentials", {}).get("api_key", {}) + api_key = context.auth.get("credentials", {}).get("api_key", "") # Build SerpApi request parameters for app search - params = {"api_key": api_key, "engine": "google_play", "store": "apps", "q": inputs["query"]} + params = { + "api_key": api_key, + "engine": "google_play", + "store": "apps", + "q": inputs["query"], + } # Make API request to SerpApi response = await context.fetch("https://serpapi.com/search", method="GET", params=params) # Extract apps from organic results - organic_results = response.get("organic_results", []) + organic_results = response.data.get("organic_results", []) limit = inputs.get("limit", 10) apps = [] @@ -189,30 +216,35 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): if len(apps) >= limit: break - return {"apps": apps, "total_results": len(apps)} + return ActionResult(data={"apps": apps, "total_results": len(apps)}, cost_usd=0.0) @app_business_reviews.action("get_reviews_google_play") class GetReviewsGooglePlay(ActionHandler): async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): - api_key = context.auth.get("credentials", {}).get("api_key", {}) + api_key = context.auth.get("credentials", {}).get("api_key", "") # Get product_id - either provided directly or search by app name product_id = inputs.get("product_id") app_name = inputs.get("app_name") if not product_id and not app_name: - raise ValueError("Either product_id or app_name must be provided") + return ActionError(message="Either product_id or app_name must be provided") # If app_name is provided but no product_id, search for the app first if app_name and not product_id: - search_params = {"api_key": api_key, "engine": "google_play", "store": "apps", "q": app_name} + search_params = { + "api_key": api_key, + "engine": "google_play", + "store": "apps", + "q": app_name, + } search_response = await context.fetch("https://serpapi.com/search", method="GET", params=search_params) - organic_results = search_response.get("organic_results", []) + organic_results = search_response.data.get("organic_results", []) if not organic_results: - raise ValueError(f"No apps found for search term: {app_name}") + return ActionError(message=f"No apps found for search term: {app_name}") # Get the first result's product ID from nested items structure product_id = None @@ -224,7 +256,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): break if not product_id: - raise ValueError(f"Could not extract product ID for app: {app_name}") + return ActionError(message=f"Could not extract product ID for app: {app_name}") # Build SerpApi request parameters params = { @@ -251,6 +283,9 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): all_reviews = [] next_page_token = None pages_fetched = 0 + # Sentinel: stays None when max_pages == 0 so the post-loop + # extraction below can fall back to an empty dict safely. + response = None # Fetch reviews with pagination while pages_fetched < max_pages: @@ -265,7 +300,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): response = await context.fetch("https://serpapi.com/search", method="GET", params=params) # Extract reviews data from current page - page_reviews = response.get("reviews", []) + page_reviews = response.data.get("reviews", []) if not page_reviews: break @@ -285,21 +320,24 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): pages_fetched += 1 # Check if there's a next page - pagination_info = response.get("serpapi_pagination", {}) + pagination_info = response.data.get("serpapi_pagination", {}) next_page_token = pagination_info.get("next_page_token") if not next_page_token: break - # Extract app information from the response - app_info = response.get("product_info", {}) + # Extract app information from the last response + app_info = response.data.get("product_info", {}) if response is not None else {} - return { - "reviews": all_reviews, - "total_reviews": len(all_reviews), - "app_name": app_info.get("title", ""), - "app_rating": app_info.get("rating") or 0.0, - "product_id": product_id, - } + return ActionResult( + data={ + "reviews": all_reviews, + "total_reviews": len(all_reviews), + "app_name": app_info.get("title", ""), + "app_rating": app_info.get("rating") or 0.0, + "product_id": product_id, + }, + cost_usd=0.0, + ) # ---- Google Maps Actions ---- @@ -308,7 +346,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): @app_business_reviews.action("search_places_google_maps") class SearchPlacesGoogleMaps(ActionHandler): async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): - api_key = context.auth.get("credentials", {}).get("api_key", {}) + api_key = context.auth.get("credentials", {}).get("api_key", "") # Build SerpApi request parameters for place search query_string = inputs["query"] @@ -317,13 +355,18 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): if inputs.get("location"): query_string = f"{query_string} in {inputs['location']}" - params = {"api_key": api_key, "engine": "google_maps", "type": "search", "q": query_string} + params = { + "api_key": api_key, + "engine": "google_maps", + "type": "search", + "q": query_string, + } # Make API request to SerpApi response = await context.fetch("https://serpapi.com/search", method="GET", params=params) # Extract places from local results - local_results = response.get("local_results", []) + local_results = response.data.get("local_results", []) limit = inputs.get("num_results", 5) places = [] @@ -340,13 +383,13 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): } places.append(place) - return {"places": places, "total_results": len(places)} + return ActionResult(data={"places": places, "total_results": len(places)}, cost_usd=0.0) @app_business_reviews.action("get_reviews_google_maps") class GetReviewsGoogleMaps(ActionHandler): async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): - api_key = context.auth.get("credentials", {}).get("api_key", {}) + api_key = context.auth.get("credentials", {}).get("api_key", "") # Get place_id and data_id - either provided directly or search by business name place_id = inputs.get("place_id") @@ -355,7 +398,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): local_results = [] # Initialize to store search results if not place_id and not data_id and not query: - raise ValueError("Either place_id, data_id, or query (business name) must be provided") + return ActionError(message="Either place_id, data_id, or query (business name) must be provided") # If query is provided but no place_id/data_id, search for the place first if query and not place_id and not data_id: @@ -364,12 +407,17 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): if inputs.get("location"): search_query = f"{query} in {inputs['location']}" - search_params = {"api_key": api_key, "engine": "google_maps", "type": "search", "q": search_query} + search_params = { + "api_key": api_key, + "engine": "google_maps", + "type": "search", + "q": search_query, + } # Search for the place to get place_id and data_id search_response = await context.fetch("https://serpapi.com/search", method="GET", params=search_params) - local_results = search_response.get("local_results", []) + local_results = search_response.data.get("local_results", []) if not local_results: # Provide helpful error message suggestion = ( @@ -377,7 +425,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): "Visit: https://developers.google.com/maps/documentation/places/web-service/" "place-id to find Place ID manually." ) - raise ValueError(suggestion) + return ActionError(message=suggestion) # Get the first result's place_id and data_id first_result = local_results[0] @@ -385,9 +433,11 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): data_id = first_result.get("data_id") if not place_id and not data_id: - raise ValueError( - f"Could not extract place_id or data_id for business: {search_query}. " - "The search returned results but they don't contain required identifiers." + return ActionError( + message=( + f"Could not extract place_id or data_id for business: {search_query}. " + "The search returned results but they don't contain required identifiers." + ) ) # Build SerpApi request parameters for reviews @@ -399,7 +449,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): elif data_id: params["data_id"] = data_id else: - raise ValueError("Could not resolve place_id or data_id from the provided query") + return ActionError(message="Could not resolve place_id or data_id from the provided query") # Add sort parameter if provided if inputs.get("sort_by"): @@ -411,6 +461,9 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): all_reviews = [] next_page_token = None pages_fetched = 0 + # Sentinel: stays None when max_pages == 0 so the post-loop + # extraction below can fall back to an empty dict safely. + response = None # Fetch reviews with pagination while pages_fetched < max_pages: @@ -426,7 +479,7 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): response = await context.fetch("https://serpapi.com/search", method="GET", params=params) # Extract reviews data from current page - page_reviews = response.get("reviews", []) + page_reviews = response.data.get("reviews", []) if not page_reviews: break @@ -444,12 +497,12 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): pages_fetched += 1 # Check if there's a next page - next_page_token = response.get("serpapi_pagination", {}).get("next_page_token") + next_page_token = response.data.get("serpapi_pagination", {}).get("next_page_token") if not next_page_token: break # Extract business information from the last response - place_info = response.get("place_info", {}) + place_info = response.data.get("place_info", {}) if response is not None else {} # Use business name from search result if we searched, otherwise from place_info business_name = place_info.get("title", "") @@ -458,10 +511,13 @@ async def execute(self, inputs: Dict[str, Any], context: ExecutionContext): if local_results: business_name = local_results[0].get("title", business_name) - return { - "reviews": all_reviews, - "total_reviews": len(all_reviews), - "average_rating": place_info.get("rating") or 0.0, - "business_name": business_name, - "place_id": place_id or place_info.get("place_id", inputs.get("place_id", "")), - } + return ActionResult( + data={ + "reviews": all_reviews, + "total_reviews": len(all_reviews), + "average_rating": place_info.get("rating") or 0.0, + "business_name": business_name, + "place_id": place_id or place_info.get("place_id", inputs.get("place_id", "")), + }, + cost_usd=0.0, + ) diff --git a/app-business-reviews/config.json b/app-business-reviews/config.json index 0f5285cd..73d87833 100644 --- a/app-business-reviews/config.json +++ b/app-business-reviews/config.json @@ -1,7 +1,7 @@ { "name": "App and Business Reviews", "display_name": "App and Business Reviews", - "version": "1.0.0", + "version": "2.0.0", "description": "Access reviews from App Store, Google Play Store, and Google Maps using SerpAPI", "entry_point": "app_business_reviews.py", "auth": { diff --git a/app-business-reviews/requirements.txt b/app-business-reviews/requirements.txt index b56fee2e..1af9591f 100644 --- a/app-business-reviews/requirements.txt +++ b/app-business-reviews/requirements.txt @@ -1 +1 @@ -autohive-integrations-sdk~=1.0.2 +autohive-integrations-sdk~=2.0.0 diff --git a/app-business-reviews/tests/conftest.py b/app-business-reviews/tests/conftest.py new file mode 100644 index 00000000..f0509d86 --- /dev/null +++ b/app-business-reviews/tests/conftest.py @@ -0,0 +1,5 @@ +import os +import sys + +# Put the integration root on sys.path so test files can use plain imports +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) diff --git a/app-business-reviews/tests/context.py b/app-business-reviews/tests/context.py deleted file mode 100644 index a7265d89..00000000 --- a/app-business-reviews/tests/context.py +++ /dev/null @@ -1,8 +0,0 @@ -# -*- coding: utf-8 -*- -import sys -import os - -sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) -sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../dependencies"))) - -from app_business_reviews import app_business_reviews diff --git a/app-business-reviews/tests/test_app_business_reviews.py b/app-business-reviews/tests/test_app_business_reviews.py deleted file mode 100644 index e4827182..00000000 --- a/app-business-reviews/tests/test_app_business_reviews.py +++ /dev/null @@ -1,330 +0,0 @@ -# Comprehensive testbed for the unified SerpAPI integration -# Tests all three review sources: App Store, Google Play, and Google Maps -import asyncio -from context import app_business_reviews -from autohive_integrations_sdk import ExecutionContext - -# ============ Apple App Store Tests ============ - - -async def test_search_apps_ios(): - """Test searching for iOS apps""" - auth = {"api_key": "your_serpapi_key_here"} - - inputs = {"term": "WhatsApp", "country": "us", "num": 5} - - async with ExecutionContext(auth=auth) as context: - try: - result = await app_business_reviews.execute_action("search_apps_ios", inputs, context) - print(f"[iOS] Found {result['total_results']} apps for 'WhatsApp'") - - for i, app in enumerate(result["apps"]): - print(f" {i + 1}. {app['title']} (ID: {app['id']})") - print(f" Developer: {app['developer']['name']}") - if app["rating"]: - print(f" Rating: {app['rating'][0].get('rating', 'N/A')}") - print(f" Link: {app['link']}") - - except Exception as e: - print(f"[iOS] Error testing search_apps_ios: {e}") - - -async def test_get_reviews_app_store_by_id(): - """Test getting App Store reviews by product ID""" - auth = {"api_key": "your_serpapi_key_here"} - - inputs = { - "product_id": "310633997", # WhatsApp - "country": "us", - "sort": "mostrecent", - "max_pages": 2, - } - - async with ExecutionContext(auth=auth) as context: - try: - result = await app_business_reviews.execute_action("get_reviews_app_store", inputs, context) - print(f"[iOS] Found {result['total_reviews']} reviews for {result['app_name']}") - print(f"[iOS] Product ID: {result['product_id']}") - - # Print first few reviews - for i, review in enumerate(result["reviews"][:3]): - print(f" Review {i + 1}: {review['rating']} stars - {review['title']}") - print(f" By: {review['author']['name']}") - print(f" Date: {review['review_date']}") - print(f" {review['text'][:80]}...") - - except Exception as e: - print(f"[iOS] Error testing get_reviews_app_store: {e}") - - -async def test_get_reviews_app_store_by_name(): - """Test getting App Store reviews by app name (auto-resolve ID)""" - auth = {"api_key": "your_serpapi_key_here"} - - inputs = {"app_name": "Instagram", "country": "us", "sort": "mostfavorable", "max_pages": 1} - - async with ExecutionContext(auth=auth) as context: - try: - result = await app_business_reviews.execute_action("get_reviews_app_store", inputs, context) - print(f"[iOS] Auto-resolved '{inputs['app_name']}' to ID: {result['product_id']}") - print(f"[iOS] Fetched {result['total_reviews']} favorable reviews") - - if result["reviews"]: - first_review = result["reviews"][0] - print(f" Top review: {first_review['rating']} stars - {first_review['title']}") - - except Exception as e: - print(f"[iOS] Error testing reviews by name: {e}") - - -# ============ Google Play Store Tests ============ - - -async def test_search_apps_android(): - """Test searching for Android apps""" - auth = {"api_key": "your_serpapi_key_here"} - - inputs = {"query": "Spotify", "limit": 5} - - async with ExecutionContext(auth=auth) as context: - try: - result = await app_business_reviews.execute_action("search_apps_android", inputs, context) - print(f"[Android] Found {result['total_results']} apps for 'Spotify'") - - for i, app in enumerate(result["apps"]): - print(f" {i + 1}. {app['title']}") - print(f" ID: {app['product_id']}") - print(f" Developer: {app['developer']}") - print(f" Rating: {app['rating']} | Price: {app['price']}") - - except Exception as e: - print(f"[Android] Error testing search_apps_android: {e}") - - -async def test_get_reviews_google_play_basic(): - """Test getting Google Play reviews with basic params""" - auth = {"api_key": "your_serpapi_key_here"} - - inputs = { - "product_id": "com.whatsapp", - "sort_by": 2, # Newest - "num_reviews": 20, - "max_pages": 2, - } - - async with ExecutionContext(auth=auth) as context: - try: - result = await app_business_reviews.execute_action("get_reviews_google_play", inputs, context) - print(f"[Android] Found {result['total_reviews']} reviews for {result['app_name']}") - print(f"[Android] App rating: {result['app_rating']}") - print(f"[Android] Product ID: {result['product_id']}") - - # Print first few reviews - for i, review in enumerate(result["reviews"][:3]): - print(f" Review {i + 1}: {review['rating']} stars by {review['author']}") - print(f" Date: {review['date']} | Likes: {review['likes']}") - print(f" {review['text'][:80]}...") - - except Exception as e: - print(f"[Android] Error testing get_reviews_google_play: {e}") - - -async def test_get_reviews_google_play_with_filters(): - """Test getting Google Play reviews with filters""" - auth = {"api_key": "your_serpapi_key_here"} - - inputs = { - "product_id": "com.instagram.android", - "rating": 5, # Only 5-star reviews - "platform": "phone", - "sort_by": 1, # Most relevant - "num_reviews": 15, - "max_pages": 1, - } - - async with ExecutionContext(auth=auth) as context: - try: - result = await app_business_reviews.execute_action("get_reviews_google_play", inputs, context) - print(f"[Android] Found {result['total_reviews']} 5-star phone reviews for {result['app_name']}") - - # Verify filter worked - ratings = [review["rating"] for review in result["reviews"]] - print(f"[Android] Review ratings (should all be 5): {set(ratings)}") - - except Exception as e: - print(f"[Android] Error testing filtered reviews: {e}") - - -async def test_get_reviews_google_play_by_name(): - """Test getting Google Play reviews by app name (auto-resolve ID)""" - auth = {"api_key": "your_serpapi_key_here"} - - inputs = { - "app_name": "Bumble", - "sort_by": 3, # Rating - "num_reviews": 10, - "max_pages": 1, - } - - async with ExecutionContext(auth=auth) as context: - try: - result = await app_business_reviews.execute_action("get_reviews_google_play", inputs, context) - print(f"[Android] Auto-resolved '{inputs['app_name']}' to: {result['product_id']}") - print(f"[Android] Fetched {result['total_reviews']} reviews") - print(f"[Android] App: {result['app_name']} (Rating: {result['app_rating']})") - - except Exception as e: - print(f"[Android] Error testing reviews by name: {e}") - - -# ============ Google Maps Tests ============ - - -async def test_search_places_google_maps(): - """Test searching for places on Google Maps""" - auth = {"api_key": "your_serpapi_key_here"} - - inputs = {"query": "Pizza restaurant", "location": "San Francisco, CA", "num_results": 5} - - async with ExecutionContext(auth=auth) as context: - try: - result = await app_business_reviews.execute_action("search_places_google_maps", inputs, context) - print(f"[Maps] Found {result['total_results']} places for 'Pizza restaurant' in San Francisco") - - for i, place in enumerate(result["places"]): - print(f" {i + 1}. {place['title']}") - print(f" Address: {place['address']}") - print(f" Rating: {place['rating']} ({place['reviews']} reviews)") - print(f" Place ID: {place['place_id']}") - - except Exception as e: - print(f"[Maps] Error testing search_places_google_maps: {e}") - - -async def test_get_reviews_google_maps_by_place_id(): - """Test getting Google Maps reviews by place ID""" - auth = {"api_key": "your_serpapi_key_here"} - - inputs = { - "place_id": "ChIJN1t_tDeuEmsRUsoyG83frY4", # Example place ID - "sort_by": "newestFirst", - "num_reviews": 20, - "max_pages": 3, - } - - async with ExecutionContext(auth=auth) as context: - try: - result = await app_business_reviews.execute_action("get_reviews_google_maps", inputs, context) - print(f"[Maps] Found {result['total_reviews']} reviews for {result['business_name']}") - print(f"[Maps] Average rating: {result['average_rating']}") - print(f"[Maps] Place ID: {result['place_id']}") - - # Print first few reviews - for i, review in enumerate(result["reviews"][:3]): - print(f" Review {i + 1}: {review['rating']} stars by {review['author']}") - print(f" Date: {review['date']} | Likes: {review['likes']}") - print(f" {review['text'][:80]}...") - - except Exception as e: - print(f"[Maps] Error testing get_reviews_google_maps with place_id: {e}") - - -async def test_get_reviews_google_maps_by_query(): - """Test getting Google Maps reviews by business name (auto-resolve place ID)""" - auth = {"api_key": "your_serpapi_key_here"} - - inputs = { - "query": "Starbucks Reserve Roastery", - "location": "Seattle, WA", - "sort_by": "qualityScore", - "num_reviews": 15, - "max_pages": 2, - } - - async with ExecutionContext(auth=auth) as context: - try: - result = await app_business_reviews.execute_action("get_reviews_google_maps", inputs, context) - print(f"[Maps] Auto-resolved business to: {result['business_name']}") - print(f"[Maps] Place ID: {result['place_id']}") - print(f"[Maps] Fetched {result['total_reviews']} reviews") - print(f"[Maps] Average rating: {result['average_rating']}") - - if result["reviews"]: - first_review = result["reviews"][0] - print(f" Top review: {first_review['rating']} stars") - - except Exception as e: - print(f"[Maps] Error testing reviews by query: {e}") - - -async def test_get_reviews_google_maps_with_data_id(): - """Test getting Google Maps reviews by data ID""" - auth = {"api_key": "your_serpapi_key_here"} - - inputs = { - "data_id": "0x808fb9fe2f8abe5f:0x3d1e3c3b33e4c4d8", # Example data ID - "sort_by": "ratingHigh", - "num_reviews": 10, - "max_pages": 1, - } - - async with ExecutionContext(auth=auth) as context: - try: - result = await app_business_reviews.execute_action("get_reviews_google_maps", inputs, context) - print(f"[Maps] Found {result['total_reviews']} high-rated reviews") - print(f"[Maps] Business: {result['business_name']}") - - except Exception as e: - print(f"[Maps] Error testing get_reviews_google_maps with data_id: {e}") - - -# ============ Main Test Runner ============ - - -async def main(): - print("=" * 60) - print("TESTING UNIFIED SERPAPI INTEGRATION") - print("=" * 60) - print() - - print("=" * 60) - print("APPLE APP STORE TESTS") - print("=" * 60) - await test_search_apps_ios() - print() - await test_get_reviews_app_store_by_id() - print() - await test_get_reviews_app_store_by_name() - print() - - print("=" * 60) - print("GOOGLE PLAY STORE TESTS") - print("=" * 60) - await test_search_apps_android() - print() - await test_get_reviews_google_play_basic() - print() - await test_get_reviews_google_play_with_filters() - print() - await test_get_reviews_google_play_by_name() - print() - - print("=" * 60) - print("GOOGLE MAPS TESTS") - print("=" * 60) - await test_search_places_google_maps() - print() - await test_get_reviews_google_maps_by_place_id() - print() - await test_get_reviews_google_maps_by_query() - print() - await test_get_reviews_google_maps_with_data_id() - print() - - print("=" * 60) - print("ALL TESTS COMPLETED") - print("=" * 60) - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/app-business-reviews/tests/test_app_business_reviews_integration.py b/app-business-reviews/tests/test_app_business_reviews_integration.py new file mode 100644 index 00000000..fba401a8 --- /dev/null +++ b/app-business-reviews/tests/test_app_business_reviews_integration.py @@ -0,0 +1,217 @@ +""" +End-to-end integration tests for the App Business Reviews integration. + +These tests call the real SerpApi API and require a valid API key +set in the APP_BUSINESS_REVIEWS_API_KEY environment variable (via .env or export). + +Run with: + pytest app-business-reviews/tests/test_app_business_reviews_integration.py -m integration + +Never runs in CI -- the default pytest marker filter (-m unit) excludes these, +and the file naming (test_*_integration.py) is not matched by python_files. +""" + +import os + +import pytest +from unittest.mock import MagicMock, AsyncMock +from autohive_integrations_sdk import FetchResponse + +from app_business_reviews import app_business_reviews as abr + +pytestmark = pytest.mark.integration + +API_KEY = os.environ.get("APP_BUSINESS_REVIEWS_API_KEY", "") + + +@pytest.fixture +def live_context(): + if not API_KEY: + pytest.skip("APP_BUSINESS_REVIEWS_API_KEY not set - skipping integration tests") + + import aiohttp + + async def real_fetch(url, *, method="GET", json=None, headers=None, params=None, **kwargs): + async with aiohttp.ClientSession() as session: + async with session.request(method, url, json=json, headers=headers or {}, params=params) as resp: + data = await resp.json(content_type=None) + return FetchResponse(status=resp.status, headers=dict(resp.headers), data=data) + + ctx = MagicMock(name="ExecutionContext") + ctx.fetch = AsyncMock(side_effect=real_fetch) + ctx.auth = {"credentials": {"api_key": API_KEY}} + return ctx + + +# ---- Read-Only Tests ---- + + +class TestSearchAppsIos: + async def test_returns_apps(self, live_context): + result = await abr.execute_action( + "search_apps_ios", + {"term": "WhatsApp", "country": "us", "num": 3}, + live_context, + ) + + data = result.result.data + assert "apps" in data + assert isinstance(data["apps"], list) + + async def test_num_limit_respected(self, live_context): + result = await abr.execute_action( + "search_apps_ios", + {"term": "Instagram", "num": 2}, + live_context, + ) + + data = result.result.data + assert data["total_results"] <= 2 + + +class TestSearchAppsAndroid: + async def test_returns_apps(self, live_context): + result = await abr.execute_action( + "search_apps_android", + {"query": "WhatsApp"}, + live_context, + ) + + data = result.result.data + assert "apps" in data + assert isinstance(data["apps"], list) + + async def test_limit_respected(self, live_context): + result = await abr.execute_action( + "search_apps_android", + {"query": "Spotify", "limit": 2}, + live_context, + ) + + data = result.result.data + assert data["total_results"] <= 2 + + +class TestSearchPlacesGoogleMaps: + async def test_returns_places(self, live_context): + result = await abr.execute_action( + "search_places_google_maps", + {"query": "coffee Sydney"}, + live_context, + ) + + data = result.result.data + assert "places" in data + assert isinstance(data["places"], list) + + async def test_location_filters_results(self, live_context): + result = await abr.execute_action( + "search_places_google_maps", + {"query": "pizza", "location": "New York, NY", "num_results": 3}, + live_context, + ) + + data = result.result.data + assert "places" in data + assert data["total_results"] <= 3 + + +class TestGetReviewsAppStore: + async def test_returns_reviews_for_whatsapp(self, live_context): + # First get a real product ID from search + search_result = await abr.execute_action("search_apps_ios", {"term": "WhatsApp", "num": 1}, live_context) + apps = search_result.result.data["apps"] + if not apps: + pytest.skip("No iOS apps returned from search") + + product_id = str(apps[0]["id"]) + + result = await abr.execute_action( + "get_reviews_app_store", + {"product_id": product_id, "max_pages": 1}, + live_context, + ) + + data = result.result.data + assert "reviews" in data + assert "total_reviews" in data + assert "product_id" in data + + async def test_auto_resolve_by_app_name(self, live_context): + result = await abr.execute_action( + "get_reviews_app_store", + {"app_name": "WhatsApp", "max_pages": 1}, + live_context, + ) + + data = result.result.data + assert "reviews" in data + assert "app_name" in data + + +class TestGetReviewsGooglePlay: + async def test_returns_reviews_for_whatsapp(self, live_context): + result = await abr.execute_action( + "get_reviews_google_play", + {"product_id": "com.whatsapp", "max_pages": 1}, + live_context, + ) + + data = result.result.data + assert "reviews" in data + assert "total_reviews" in data + assert data["product_id"] == "com.whatsapp" + + async def test_auto_resolve_by_app_name(self, live_context): + result = await abr.execute_action( + "get_reviews_google_play", + {"app_name": "WhatsApp", "max_pages": 1}, + live_context, + ) + + data = result.result.data + assert "reviews" in data + assert "product_id" in data + + +class TestGetReviewsGoogleMaps: + async def test_returns_reviews_chained_from_search(self, live_context): + # Get a real place_id from search + search_result = await abr.execute_action( + "search_places_google_maps", + {"query": "Starbucks Sydney", "num_results": 1}, + live_context, + ) + places = search_result.result.data["places"] + if not places: + pytest.skip("No places returned from search") + + place = places[0] + identifier = {"place_id": place["place_id"]} if place.get("place_id") else {"data_id": place["data_id"]} + + result = await abr.execute_action("get_reviews_google_maps", {**identifier, "max_pages": 1}, live_context) + + data = result.result.data + assert "reviews" in data + assert "total_reviews" in data + assert "business_name" in data + + async def test_response_structure(self, live_context): + search_result = await abr.execute_action( + "search_places_google_maps", + {"query": "McDonald's Melbourne", "num_results": 1}, + live_context, + ) + places = search_result.result.data["places"] + if not places: + pytest.skip("No places returned from search") + + place = places[0] + identifier = {"place_id": place["place_id"]} if place.get("place_id") else {"data_id": place["data_id"]} + + result = await abr.execute_action("get_reviews_google_maps", {**identifier, "max_pages": 1}, live_context) + + data = result.result.data + assert "reviews" in data + assert "average_rating" in data + assert "place_id" in data diff --git a/app-business-reviews/tests/test_app_business_reviews_unit.py b/app-business-reviews/tests/test_app_business_reviews_unit.py new file mode 100644 index 00000000..8b2deb1a --- /dev/null +++ b/app-business-reviews/tests/test_app_business_reviews_unit.py @@ -0,0 +1,739 @@ +import pytest +from unittest.mock import AsyncMock, MagicMock +from autohive_integrations_sdk import FetchResponse +from autohive_integrations_sdk.integration import ResultType + +from app_business_reviews import app_business_reviews + +pytestmark = pytest.mark.unit + +# ---- Shared sample data ---- + +SAMPLE_IOS_APP = { + "id": 310633997, + "title": "WhatsApp Messenger", + "bundle_id": "net.whatsapp.WhatsApp", + "developer": {"name": "WhatsApp Inc.", "id": 310633998}, + "rating": [{"rating": 4.7}], + "price": {"value": 0}, + "link": "https://apps.apple.com/us/app/whatsapp/id310633997", +} + +SAMPLE_IOS_REVIEW = { + "id": "rev_001", + "title": "Great app", + "text": "Works perfectly", + "rating": 5, + "author": {"name": "JohnDoe", "author_id": "a1"}, + "review_date": "2024-01-15", + "reviewed_version": "24.1", + "helpfulness_vote_information": "", +} + +SAMPLE_ANDROID_APP_SECTION = { + "items": [ + { + "product_id": "com.whatsapp", + "title": "WhatsApp Messenger", + "author": "WhatsApp LLC", + "rating": 4.3, + "price": "Free", + "thumbnail": "https://example.com/thumb.png", + "link": "https://play.google.com/store/apps/details?id=com.whatsapp", + } + ] +} + +SAMPLE_ANDROID_REVIEW = { + "id": "gp_rev_001", + "rating": 5, + "snippet": "Excellent app", + "user": {"name": "Jane", "avatar": "https://example.com/avatar.png"}, + "date": "2024-02-01", + "likes": 12, +} + +SAMPLE_MAPS_PLACE = { + "place_id": "ChIJN1t_tDeuEmsRUsoyG83frY4", + "data_id": "0xdata123", + "title": "Starbucks Reserve Roastery", + "address": "1124 Pike St, Seattle, WA", + "rating": 4.5, + "reviews": 1234, + "type": "Coffee shop", + "phone": "+1-206-123-4567", +} + +SAMPLE_MAPS_REVIEW = { + "rating": 5, + "snippet": "Best coffee ever", + "user": {"name": "Alice"}, + "date": "January 2024", + "likes": 3, +} + + +# ---- Fixtures ---- + + +@pytest.fixture +def mock_context(): + ctx = MagicMock(name="ExecutionContext") + ctx.fetch = AsyncMock(name="fetch") + # Source reads context.auth.get("credentials", {}).get("api_key", "") + ctx.auth = { + "credentials": {"api_key": "test_serpapi_key"}, # nosec B105 + } + return ctx + + +# ---- iOS App Store: Search Apps ---- + + +class TestSearchAppsIOS: + @pytest.mark.asyncio + async def test_happy_path_returns_apps(self, mock_context): + mock_context.fetch.return_value = FetchResponse( + status=200, + headers={}, + data={"organic_results": [SAMPLE_IOS_APP]}, + ) + + result = await app_business_reviews.execute_action("search_apps_ios", {"term": "WhatsApp"}, mock_context) + + assert result.result.data["total_results"] == 1 + assert result.result.data["apps"][0]["title"] == "WhatsApp Messenger" + + @pytest.mark.asyncio + async def test_request_url_and_method(self, mock_context): + mock_context.fetch.return_value = FetchResponse(status=200, headers={}, data={"organic_results": []}) + + await app_business_reviews.execute_action("search_apps_ios", {"term": "WhatsApp"}, mock_context) + + call_args = mock_context.fetch.call_args + assert call_args.args[0] == "https://serpapi.com/search" + assert call_args.kwargs["method"] == "GET" + + @pytest.mark.asyncio + async def test_request_params_include_term(self, mock_context): + mock_context.fetch.return_value = FetchResponse(status=200, headers={}, data={"organic_results": []}) + + await app_business_reviews.execute_action( + "search_apps_ios", {"term": "Instagram", "country": "uk"}, mock_context + ) + + params = mock_context.fetch.call_args.kwargs["params"] + assert params["term"] == "Instagram" + assert params["engine"] == "apple_app_store" + assert params["country"] == "uk" + # Regression: api_key must be forwarded from context.auth.credentials.api_key + assert params["api_key"] == "test_serpapi_key" # nosec B105 + + @pytest.mark.asyncio + async def test_empty_results(self, mock_context): + mock_context.fetch.return_value = FetchResponse(status=200, headers={}, data={"organic_results": []}) + + result = await app_business_reviews.execute_action("search_apps_ios", {"term": "NonExistent"}, mock_context) + + assert result.result.data["apps"] == [] + assert result.result.data["total_results"] == 0 + + @pytest.mark.asyncio + async def test_num_limit_applied(self, mock_context): + apps = [dict(SAMPLE_IOS_APP, id=i, title=f"App {i}") for i in range(5)] + mock_context.fetch.return_value = FetchResponse(status=200, headers={}, data={"organic_results": apps}) + + result = await app_business_reviews.execute_action("search_apps_ios", {"term": "App", "num": 3}, mock_context) + + assert result.result.data["total_results"] == 3 + + @pytest.mark.asyncio + async def test_developer_property_search(self, mock_context): + mock_context.fetch.return_value = FetchResponse(status=200, headers={}, data={"organic_results": []}) + + await app_business_reviews.execute_action( + "search_apps_ios", + {"term": "WhatsApp", "property": "developer"}, + mock_context, + ) + + params = mock_context.fetch.call_args.kwargs["params"] + assert params["property"] == "developer" + + +# ---- iOS App Store: Get Reviews ---- + + +class TestGetReviewsAppStore: + @pytest.mark.asyncio + async def test_happy_path_with_product_id(self, mock_context): + mock_context.fetch.return_value = FetchResponse( + status=200, + headers={}, + data={"reviews": [SAMPLE_IOS_REVIEW], "serpapi_pagination": {}}, + ) + + result = await app_business_reviews.execute_action( + "get_reviews_app_store", {"product_id": "310633997"}, mock_context + ) + + assert result.result.data["total_reviews"] == 1 + assert result.result.data["reviews"][0]["title"] == "Great app" + assert result.result.data["product_id"] == "310633997" + + @pytest.mark.asyncio + async def test_request_url_and_method(self, mock_context): + mock_context.fetch.return_value = FetchResponse( + status=200, headers={}, data={"reviews": [], "serpapi_pagination": {}} + ) + + await app_business_reviews.execute_action("get_reviews_app_store", {"product_id": "310633997"}, mock_context) + + call_args = mock_context.fetch.call_args + assert call_args.args[0] == "https://serpapi.com/search" + assert call_args.kwargs["method"] == "GET" + + @pytest.mark.asyncio + async def test_request_params_apple_reviews_engine(self, mock_context): + mock_context.fetch.return_value = FetchResponse( + status=200, headers={}, data={"reviews": [], "serpapi_pagination": {}} + ) + + await app_business_reviews.execute_action( + "get_reviews_app_store", + {"product_id": "310633997", "sort": "mostrecent"}, + mock_context, + ) + + params = mock_context.fetch.call_args.kwargs["params"] + assert params["engine"] == "apple_reviews" + assert params["product_id"] == "310633997" + assert params["sort"] == "mostrecent" + + @pytest.mark.asyncio + async def test_auto_resolve_app_name(self, mock_context): + mock_context.fetch.side_effect = [ + # Search response + FetchResponse( + status=200, + headers={}, + data={"organic_results": [SAMPLE_IOS_APP]}, + ), + # Reviews response + FetchResponse( + status=200, + headers={}, + data={"reviews": [SAMPLE_IOS_REVIEW], "serpapi_pagination": {}}, + ), + ] + + result = await app_business_reviews.execute_action( + "get_reviews_app_store", {"app_name": "WhatsApp"}, mock_context + ) + + assert result.result.data["app_name"] == "WhatsApp" + assert result.result.data["total_reviews"] == 1 + + @pytest.mark.asyncio + async def test_error_no_product_id_or_app_name(self, mock_context): + result = await app_business_reviews.execute_action("get_reviews_app_store", {}, mock_context) + + assert result.type == ResultType.ACTION_ERROR + assert "product_id" in result.result.message or "app_name" in result.result.message + + @pytest.mark.asyncio + async def test_error_app_name_not_found(self, mock_context): + mock_context.fetch.return_value = FetchResponse(status=200, headers={}, data={"organic_results": []}) + + result = await app_business_reviews.execute_action( + "get_reviews_app_store", {"app_name": "NonExistentApp"}, mock_context + ) + + assert result.type == ResultType.ACTION_ERROR + assert "NonExistentApp" in result.result.message + + @pytest.mark.asyncio + async def test_pagination_stops_when_no_next(self, mock_context): + mock_context.fetch.return_value = FetchResponse( + status=200, + headers={}, + data={ + "reviews": [SAMPLE_IOS_REVIEW], + "serpapi_pagination": {}, + }, # no "next" key + ) + + result = await app_business_reviews.execute_action( + "get_reviews_app_store", + {"product_id": "310633997", "max_pages": 5}, + mock_context, + ) + + # Should only make 1 call since there's no next page + assert mock_context.fetch.call_count == 1 + assert result.result.data["total_reviews"] == 1 + + @pytest.mark.asyncio + async def test_response_shape(self, mock_context): + mock_context.fetch.return_value = FetchResponse( + status=200, + headers={}, + data={"reviews": [SAMPLE_IOS_REVIEW], "serpapi_pagination": {}}, + ) + + result = await app_business_reviews.execute_action( + "get_reviews_app_store", {"product_id": "310633997"}, mock_context + ) + + data = result.result.data + assert "reviews" in data + assert "total_reviews" in data + assert "app_name" in data + assert "product_id" in data + + +# ---- Android: Search Apps ---- + + +class TestSearchAppsAndroid: + @pytest.mark.asyncio + async def test_happy_path_returns_apps(self, mock_context): + mock_context.fetch.return_value = FetchResponse( + status=200, + headers={}, + data={"organic_results": [SAMPLE_ANDROID_APP_SECTION]}, + ) + + result = await app_business_reviews.execute_action("search_apps_android", {"query": "WhatsApp"}, mock_context) + + assert result.result.data["total_results"] == 1 + assert result.result.data["apps"][0]["product_id"] == "com.whatsapp" + + @pytest.mark.asyncio + async def test_request_url_and_params(self, mock_context): + mock_context.fetch.return_value = FetchResponse(status=200, headers={}, data={"organic_results": []}) + + await app_business_reviews.execute_action("search_apps_android", {"query": "Spotify"}, mock_context) + + call_args = mock_context.fetch.call_args + assert call_args.args[0] == "https://serpapi.com/search" + params = call_args.kwargs["params"] + assert params["engine"] == "google_play" + assert params["q"] == "Spotify" + assert params["store"] == "apps" + + @pytest.mark.asyncio + async def test_limit_applied(self, mock_context): + items = [dict(product_id=f"com.app{i}", title=f"App {i}") for i in range(10)] + mock_context.fetch.return_value = FetchResponse( + status=200, + headers={}, + data={"organic_results": [{"items": items}]}, + ) + + result = await app_business_reviews.execute_action( + "search_apps_android", {"query": "App", "limit": 3}, mock_context + ) + + assert result.result.data["total_results"] == 3 + + @pytest.mark.asyncio + async def test_empty_results(self, mock_context): + mock_context.fetch.return_value = FetchResponse(status=200, headers={}, data={"organic_results": []}) + + result = await app_business_reviews.execute_action( + "search_apps_android", {"query": "NonExistent"}, mock_context + ) + + assert result.result.data["apps"] == [] + assert result.result.data["total_results"] == 0 + + +# ---- Android: Get Reviews Google Play ---- + + +class TestGetReviewsGooglePlay: + @pytest.mark.asyncio + async def test_happy_path_with_product_id(self, mock_context): + mock_context.fetch.return_value = FetchResponse( + status=200, + headers={}, + data={ + "reviews": [SAMPLE_ANDROID_REVIEW], + "product_info": {"title": "WhatsApp", "rating": 4.3}, + "serpapi_pagination": {}, + }, + ) + + result = await app_business_reviews.execute_action( + "get_reviews_google_play", {"product_id": "com.whatsapp"}, mock_context + ) + + assert result.result.data["total_reviews"] == 1 + assert result.result.data["product_id"] == "com.whatsapp" + assert result.result.data["app_name"] == "WhatsApp" + + @pytest.mark.asyncio + async def test_request_uses_google_play_product_engine(self, mock_context): + mock_context.fetch.return_value = FetchResponse( + status=200, + headers={}, + data={"reviews": [], "product_info": {}, "serpapi_pagination": {}}, + ) + + await app_business_reviews.execute_action( + "get_reviews_google_play", {"product_id": "com.whatsapp"}, mock_context + ) + + params = mock_context.fetch.call_args.kwargs["params"] + assert params["engine"] == "google_play_product" + assert params["product_id"] == "com.whatsapp" + assert params["all_reviews"] == "true" + + @pytest.mark.asyncio + async def test_auto_resolve_app_name(self, mock_context): + mock_context.fetch.side_effect = [ + FetchResponse( + status=200, + headers={}, + data={"organic_results": [SAMPLE_ANDROID_APP_SECTION]}, + ), + FetchResponse( + status=200, + headers={}, + data={ + "reviews": [SAMPLE_ANDROID_REVIEW], + "product_info": {"title": "WhatsApp", "rating": 4.3}, + "serpapi_pagination": {}, + }, + ), + ] + + result = await app_business_reviews.execute_action( + "get_reviews_google_play", {"app_name": "WhatsApp"}, mock_context + ) + + assert result.result.data["product_id"] == "com.whatsapp" + + @pytest.mark.asyncio + async def test_error_no_product_id_or_app_name(self, mock_context): + result = await app_business_reviews.execute_action("get_reviews_google_play", {}, mock_context) + + assert result.type == ResultType.ACTION_ERROR + + @pytest.mark.asyncio + async def test_error_app_name_not_found(self, mock_context): + mock_context.fetch.return_value = FetchResponse(status=200, headers={}, data={"organic_results": []}) + + result = await app_business_reviews.execute_action( + "get_reviews_google_play", {"app_name": "NonExistentApp"}, mock_context + ) + + assert result.type == ResultType.ACTION_ERROR + assert "NonExistentApp" in result.result.message + + @pytest.mark.asyncio + async def test_optional_filters_included_in_params(self, mock_context): + mock_context.fetch.return_value = FetchResponse( + status=200, + headers={}, + data={"reviews": [], "product_info": {}, "serpapi_pagination": {}}, + ) + + await app_business_reviews.execute_action( + "get_reviews_google_play", + { + "product_id": "com.instagram.android", + "rating": 5, + "platform": "phone", + "sort_by": 2, + }, + mock_context, + ) + + params = mock_context.fetch.call_args.kwargs["params"] + assert params["rating"] == 5 + assert params["platform"] == "phone" + assert params["sort_by"] == 2 + + @pytest.mark.asyncio + async def test_response_shape(self, mock_context): + mock_context.fetch.return_value = FetchResponse( + status=200, + headers={}, + data={ + "reviews": [SAMPLE_ANDROID_REVIEW], + "product_info": {"title": "App", "rating": 4.0}, + "serpapi_pagination": {}, + }, + ) + + result = await app_business_reviews.execute_action( + "get_reviews_google_play", {"product_id": "com.whatsapp"}, mock_context + ) + + data = result.result.data + assert "reviews" in data + assert "total_reviews" in data + assert "app_name" in data + assert "app_rating" in data + assert "product_id" in data + + +# ---- Google Maps: Search Places ---- + + +class TestSearchPlacesGoogleMaps: + @pytest.mark.asyncio + async def test_happy_path_returns_places(self, mock_context): + mock_context.fetch.return_value = FetchResponse( + status=200, + headers={}, + data={"local_results": [SAMPLE_MAPS_PLACE]}, + ) + + result = await app_business_reviews.execute_action( + "search_places_google_maps", {"query": "Starbucks"}, mock_context + ) + + assert result.result.data["total_results"] == 1 + assert result.result.data["places"][0]["title"] == "Starbucks Reserve Roastery" + + @pytest.mark.asyncio + async def test_request_url_and_engine(self, mock_context): + mock_context.fetch.return_value = FetchResponse(status=200, headers={}, data={"local_results": []}) + + await app_business_reviews.execute_action("search_places_google_maps", {"query": "Pizza"}, mock_context) + + call_args = mock_context.fetch.call_args + assert call_args.args[0] == "https://serpapi.com/search" + params = call_args.kwargs["params"] + assert params["engine"] == "google_maps" + assert params["type"] == "search" + + @pytest.mark.asyncio + async def test_location_appended_to_query(self, mock_context): + mock_context.fetch.return_value = FetchResponse(status=200, headers={}, data={"local_results": []}) + + await app_business_reviews.execute_action( + "search_places_google_maps", + {"query": "Pizza", "location": "New York, NY"}, + mock_context, + ) + + params = mock_context.fetch.call_args.kwargs["params"] + assert params["q"] == "Pizza in New York, NY" + + @pytest.mark.asyncio + async def test_num_results_limit(self, mock_context): + places = [dict(SAMPLE_MAPS_PLACE, place_id=f"place_{i}", title=f"Place {i}") for i in range(10)] + mock_context.fetch.return_value = FetchResponse(status=200, headers={}, data={"local_results": places}) + + result = await app_business_reviews.execute_action( + "search_places_google_maps", + {"query": "Coffee", "num_results": 3}, + mock_context, + ) + + assert result.result.data["total_results"] == 3 + + @pytest.mark.asyncio + async def test_empty_results(self, mock_context): + mock_context.fetch.return_value = FetchResponse(status=200, headers={}, data={"local_results": []}) + + result = await app_business_reviews.execute_action( + "search_places_google_maps", {"query": "NonExistent"}, mock_context + ) + + assert result.result.data["places"] == [] + + @pytest.mark.asyncio + async def test_response_shape(self, mock_context): + mock_context.fetch.return_value = FetchResponse( + status=200, headers={}, data={"local_results": [SAMPLE_MAPS_PLACE]} + ) + + result = await app_business_reviews.execute_action( + "search_places_google_maps", {"query": "Starbucks"}, mock_context + ) + + place = result.result.data["places"][0] + assert "place_id" in place + assert "data_id" in place + assert "title" in place + assert "address" in place + assert "rating" in place + + +# ---- Google Maps: Get Reviews ---- + + +class TestGetReviewsGoogleMaps: + @pytest.mark.asyncio + async def test_happy_path_with_place_id(self, mock_context): + mock_context.fetch.return_value = FetchResponse( + status=200, + headers={}, + data={ + "reviews": [SAMPLE_MAPS_REVIEW], + "place_info": { + "title": "Starbucks", + "rating": 4.5, + "place_id": "ChIJN1t_tDeuEmsRUsoyG83frY4", + }, + "serpapi_pagination": {}, + }, + ) + + result = await app_business_reviews.execute_action( + "get_reviews_google_maps", + {"place_id": "ChIJN1t_tDeuEmsRUsoyG83frY4"}, + mock_context, + ) + + assert result.result.data["total_reviews"] == 1 + assert result.result.data["average_rating"] == 4.5 + + @pytest.mark.asyncio + async def test_request_uses_google_maps_reviews_engine(self, mock_context): + mock_context.fetch.return_value = FetchResponse( + status=200, + headers={}, + data={"reviews": [], "place_info": {}, "serpapi_pagination": {}}, + ) + + await app_business_reviews.execute_action( + "get_reviews_google_maps", + {"place_id": "ChIJN1t_tDeuEmsRUsoyG83frY4"}, + mock_context, + ) + + params = mock_context.fetch.call_args.kwargs["params"] + assert params["engine"] == "google_maps_reviews" + assert params["place_id"] == "ChIJN1t_tDeuEmsRUsoyG83frY4" + + @pytest.mark.asyncio + async def test_uses_data_id_when_no_place_id(self, mock_context): + mock_context.fetch.return_value = FetchResponse( + status=200, + headers={}, + data={"reviews": [], "place_info": {}, "serpapi_pagination": {}}, + ) + + await app_business_reviews.execute_action("get_reviews_google_maps", {"data_id": "0xdata123"}, mock_context) + + params = mock_context.fetch.call_args.kwargs["params"] + assert params["data_id"] == "0xdata123" + assert "place_id" not in params + + @pytest.mark.asyncio + async def test_auto_resolve_by_query(self, mock_context): + mock_context.fetch.side_effect = [ + FetchResponse( + status=200, + headers={}, + data={"local_results": [SAMPLE_MAPS_PLACE]}, + ), + FetchResponse( + status=200, + headers={}, + data={ + "reviews": [SAMPLE_MAPS_REVIEW], + "place_info": { + "title": "Starbucks Reserve Roastery", + "rating": 4.5, + }, + "serpapi_pagination": {}, + }, + ), + ] + + result = await app_business_reviews.execute_action( + "get_reviews_google_maps", + {"query": "Starbucks Reserve Roastery", "location": "Seattle, WA"}, + mock_context, + ) + + assert result.result.data["total_reviews"] == 1 + assert result.result.data["business_name"] == "Starbucks Reserve Roastery" + + @pytest.mark.asyncio + async def test_error_no_identifiers(self, mock_context): + result = await app_business_reviews.execute_action("get_reviews_google_maps", {}, mock_context) + + assert result.type == ResultType.ACTION_ERROR + assert "place_id" in result.result.message or "query" in result.result.message + + @pytest.mark.asyncio + async def test_error_query_no_results(self, mock_context): + mock_context.fetch.return_value = FetchResponse(status=200, headers={}, data={"local_results": []}) + + result = await app_business_reviews.execute_action( + "get_reviews_google_maps", {"query": "NonExistentBusiness"}, mock_context + ) + + assert result.type == ResultType.ACTION_ERROR + assert "NonExistentBusiness" in result.result.message + + @pytest.mark.asyncio + async def test_sort_by_included_in_params(self, mock_context): + mock_context.fetch.return_value = FetchResponse( + status=200, + headers={}, + data={"reviews": [], "place_info": {}, "serpapi_pagination": {}}, + ) + + await app_business_reviews.execute_action( + "get_reviews_google_maps", + {"place_id": "ChIJN1t_tDeuEmsRUsoyG83frY4", "sort_by": "newestFirst"}, + mock_context, + ) + + params = mock_context.fetch.call_args.kwargs["params"] + assert params["sort_by"] == "newestFirst" + + @pytest.mark.asyncio + async def test_pagination_stops_when_no_next_token(self, mock_context): + mock_context.fetch.return_value = FetchResponse( + status=200, + headers={}, + data={ + "reviews": [SAMPLE_MAPS_REVIEW], + "place_info": {"title": "Starbucks", "rating": 4.5}, + "serpapi_pagination": {}, # no next_page_token + }, + ) + + result = await app_business_reviews.execute_action( + "get_reviews_google_maps", + {"place_id": "ChIJN1t_tDeuEmsRUsoyG83frY4", "max_pages": 5}, + mock_context, + ) + + assert mock_context.fetch.call_count == 1 + assert result.result.data["total_reviews"] == 1 + + @pytest.mark.asyncio + async def test_response_shape(self, mock_context): + mock_context.fetch.return_value = FetchResponse( + status=200, + headers={}, + data={ + "reviews": [SAMPLE_MAPS_REVIEW], + "place_info": {"title": "Cafe", "rating": 4.2}, + "serpapi_pagination": {}, + }, + ) + + result = await app_business_reviews.execute_action( + "get_reviews_google_maps", + {"place_id": "ChIJN1t_tDeuEmsRUsoyG83frY4"}, + mock_context, + ) + + data = result.result.data + assert "reviews" in data + assert "total_reviews" in data + assert "average_rating" in data + assert "business_name" in data + assert "place_id" in data