"""
Comprehensive test suite for Digikala MCP Server
This test suite covers all tools and their parameters to ensure
proper functionality across different scenarios.
"""
import sys
import time
from main import (
get_autocomplete_suggestions,
search_products,
extract_products_from_search,
get_products_paginated,
get_product_details,
get_product_recommendations,
search_text_lenz,
tooman_to_rial,
rial_to_tooman
)
def print_section(title):
"""Print a section header"""
print("\n" + "=" * 80)
print(title)
print("=" * 80)
def test_autocomplete_suggestions():
"""Test get_autocomplete_suggestions tool with various queries"""
print_section("TEST SUITE: get_autocomplete_suggestions")
test_cases = [
("iphone", "English product query"),
("گوشی موبایل", "Persian product query"),
("laptop asus", "English with brand"),
("a", "Single character query"),
]
for query, description in test_cases:
print(f"\nTest: {description} - Query: '{query}'")
try:
result = get_autocomplete_suggestions(query)
assert result.get('status') == 200, f"Expected status 200, got {result.get('status')}"
data = result.get('data', {})
auto_complete = data.get('auto_complete', [])
categories = data.get('categories', [])
print(f"✓ Status: 200, Suggestions: {len(auto_complete)}, Categories: {len(categories)}")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("✅ All autocomplete tests passed!")
def test_search_products_basic():
"""Test basic search_products functionality"""
print_section("TEST SUITE: search_products - Basic Functionality")
test_cases = [
("laptop", "English query"),
("گوشی موبایل", "Persian query"),
]
for query, description in test_cases:
print(f"\nTest: {description} - Query: '{query}'")
try:
result = search_products(query)
assert result.get('status') == 200
products = extract_products_from_search(result)
print(f"✓ Status: 200, Products: {len(products)}")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("✅ All basic search tests passed!")
def test_search_products_pagination():
"""Test search_products pagination parameters"""
print_section("TEST SUITE: search_products - Pagination")
test_cases = [
{"query": "laptop", "page": 1, "rows": 15, "desc": "Page 1, 15 items"},
{"query": "laptop", "page": 2, "rows": 20, "desc": "Page 2, 20 items"},
]
for test_case in test_cases:
desc = test_case.pop("desc")
print(f"\nTest: {desc}")
try:
result = search_products(**test_case)
products = extract_products_from_search(result)
print(f"✓ Status: 200, Products: {len(products)}, Expected rows: {test_case.get('rows', 15)}")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("✅ All pagination tests passed!")
def test_search_products_sorting():
"""Test search_products sorting options"""
print_section("TEST SUITE: search_products - Sorting")
sort_options = [
(1, "Relevance"),
(2, "Price: Low to High"),
(3, "Price: High to Low"),
]
for sort_value, description in sort_options:
print(f"\nTest: Sort by {description} (value: {sort_value})")
try:
result = search_products("هدفون", sort=sort_value, rows=15)
products = extract_products_from_search(result)
print(f"✓ Products: {len(products)}")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("✅ All sorting tests passed!")
def test_currency_conversion():
"""Test currency conversion utilities"""
print_section("TEST SUITE: Currency Conversion")
print("\nTest: Tooman to Rial conversion")
assert tooman_to_rial(1000) == 10000
assert tooman_to_rial(5000) == 50000
print("✓ Tooman to Rial working")
print("\nTest: Rial to Tooman conversion")
assert rial_to_tooman(10000) == 1000
assert rial_to_tooman(50000) == 5000
print("✓ Rial to Tooman working")
print("\nTest: Round-trip conversion")
assert rial_to_tooman(tooman_to_rial(12345)) == 12345
print("✓ Round-trip conversion working")
print("✅ All currency conversion tests passed!")
def test_search_products_price_filters():
"""Test search_products price range filters (using Tooman)"""
print_section("TEST SUITE: search_products - Price Filters")
test_cases = [
{
"query": "گوشی موبایل",
"price_min_tooman": 500000, # 5M Rials
"price_max_tooman": 1500000, # 15M Rials
"desc": "Mobile phones 500K-1.5M Toomans"
},
{
"query": "کتاب",
"price_max_tooman": 50000, # 500K Rials
"desc": "Books under 50K Toomans"
},
]
for test_case in test_cases:
desc = test_case.pop("desc")
print(f"\nTest: {desc}")
try:
result = search_products(**test_case)
products = extract_products_from_search(result)
print(f"✓ Products: {len(products)}")
if products:
sample_prices = [f"{p['selling_price_tooman']:,}" for p in products[:2]]
print(f" Sample prices (Tooman): {sample_prices}")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("✅ All price filter tests passed!")
def test_search_products_discount_filters():
"""Test search_products discount range filters"""
print_section("TEST SUITE: search_products - Discount Filters")
print("\nTest: Products with 10-30% discount")
try:
result = search_products(query="لباس", discount_min=10, discount_max=30)
products = extract_products_from_search(result)
print(f"✓ Products: {len(products)}")
if products:
discounts = [p['discount_percent'] for p in products[:3]]
print(f" Sample discounts: {discounts}%")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("✅ All discount filter tests passed!")
def test_search_products_stock_filters():
"""Test search_products stock availability filters"""
print_section("TEST SUITE: search_products - Stock Filters")
test_cases = [
{
"query": "کتاب",
"has_selling_stock": 1,
"desc": "Books with selling stock"
},
{
"query": "لپ تاپ",
"has_selling_stock": 1,
"has_ready_to_shipment": 1,
"desc": "Laptops available and ready to ship"
},
]
for test_case in test_cases:
desc = test_case.pop("desc")
print(f"\nTest: {desc}")
try:
result = search_products(**test_case)
products = extract_products_from_search(result)
print(f"✓ Products: {len(products)}")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("✅ All stock filter tests passed!")
def test_search_products_digiplus_filter():
"""Test search_products Digiplus filter"""
print_section("TEST SUITE: search_products - Digiplus Filter")
print("\nTest: Digiplus mobile phones")
try:
result = search_products(query="گوشی موبایل", digiplus=1)
products = extract_products_from_search(result)
print(f"✓ Products: {len(products)}")
if products:
jet_eligible = sum(1 for p in products if p['is_digiplus_jet_eligible'])
print(f" Jet-eligible: {jet_eligible}/{len(products)}")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("✅ All Digiplus filter tests passed!")
def test_search_products_combined_filters():
"""Test search_products with multiple filters combined (using Tooman)"""
print_section("TEST SUITE: search_products - Combined Filters")
test_cases = [
{
"query": "گوشی موبایل",
"price_min_tooman": 1000000, # 10M Rials
"price_max_tooman": 3000000, # 30M Rials
"has_selling_stock": 1,
"sort": 2,
"desc": "Available phones 1M-3M Toomans, sorted by price"
},
{
"query": "کتاب",
"price_min_tooman": 10000, # 100K Rials
"price_max_tooman": 50000, # 500K Rials
"has_selling_stock": 1,
"rows": 20,
"desc": "Available books 10K-50K Toomans, 20 per page"
},
]
for test_case in test_cases:
desc = test_case.pop("desc")
print(f"\nTest: {desc}")
try:
result = search_products(**test_case)
products = extract_products_from_search(result)
print(f"✓ Products: {len(products)}")
if products:
print(f" First: {products[0]['title_fa'][:40]}...")
print(f" Price: {products[0]['selling_price_tooman']:,} Toomans")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("✅ All combined filter tests passed!")
def test_extract_products_from_search():
"""Test extract_products_from_search tool with currency conversion"""
print_section("TEST SUITE: extract_products_from_search")
print("\nTest: Extract products with both currencies")
try:
search_result = search_products("laptop")
products = extract_products_from_search(search_result)
print(f"✓ Products extracted: {len(products)}")
if products:
product = products[0]
print(f" Validating product structure:")
print(f" ✓ ID: {product['id']}")
print(f" ✓ Title: {product['title_fa'][:40]}...")
print(f" ✓ Price (Tooman): {product['selling_price_tooman']:,}")
print(f" ✓ Price (Rial): {product['selling_price_rial']:,}")
print(f" ✓ Rating: {product['rating_rate']}/100 ({product['rating_count']} reviews)")
print(f" ✓ Brand: {product['brand']}")
# Validate currency conversion
assert product['selling_price_tooman'] == product['selling_price_rial'] // 10, \
"Tooman should equal Rial / 10"
print(f" ✓ Currency conversion validated")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("✅ All extraction tests passed!")
def test_get_products_paginated():
"""Test get_products_paginated tool with Tooman currency"""
print_section("TEST SUITE: get_products_paginated")
print("\n--- Test: Basic Pagination (20 items per page) ---")
try:
result = get_products_paginated("laptop", page=1)
assert result['items_per_page'] == 20, "Items per page should always be 20"
assert 'products' in result, "Result should contain products list"
print(f"✓ Page: {result['page']}")
print(f"✓ Items per page: {result['items_per_page']}")
print(f"✓ Products in page: {result['total_products_in_page']}")
print(f"✓ Has more: {result['has_more']}")
# Verify currency fields
if result['products']:
first = result['products'][0]
assert 'selling_price_tooman' in first, "Should have Tooman price"
assert 'selling_price_rial' in first, "Should have Rial price"
print(f" Sample: {first['selling_price_tooman']:,} Toomans")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("\n--- Test: Paginated Search with Filters (Tooman) ---")
try:
result = get_products_paginated(
query="لپ تاپ",
page=1,
price_min_tooman=2000000, # 20M Rials
sort=2
)
print(f"✓ Filtered results: {result['total_products_in_page']} products")
print(f"✓ Has more: {result['has_more']}")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("✅ All pagination tests passed!")
def test_edge_cases():
"""Test edge cases and error scenarios"""
print_section("TEST SUITE: Edge Cases")
print("\n--- Test: Single character query ---")
try:
result = get_autocomplete_suggestions("a")
print(f"✓ Single char handled: status {result.get('status')}")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("\n--- Test: Very long query ---")
try:
long_query = "لپ تاپ گیمینگ با پردازنده قوی و کارت گرافیک مناسب"
result = search_products(long_query)
products = extract_products_from_search(result)
print(f"✓ Long query handled: {len(products)} products found")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("✅ All edge case tests passed!")
def test_get_product_details():
"""Test get_product_details tool"""
print_section("TEST SUITE: get_product_details")
# Use a known product ID from search results
print("\n--- Test: Get product details by ID ---")
try:
# First get a product ID from search
search_result = search_products("laptop", rows=15)
products = extract_products_from_search(search_result)
if not products:
print("⚠️ No products found to test product details")
return
test_product_id = products[0]['id']
print(f"Testing with product ID: {test_product_id}")
result = get_product_details(test_product_id)
assert result.get('status') == 200, f"Expected status 200, got {result.get('status')}"
product_data = result.get('data', {}).get('product', {})
print(f"✓ Status: 200")
print(f"✓ Product ID: {product_data.get('id')}")
print(f"✓ Title: {product_data.get('title_fa', '')[:50]}...")
print(f"✓ Status: {product_data.get('status')}")
# Validate structure
assert 'id' in product_data, "Product data should have 'id'"
assert 'title_fa' in product_data, "Product data should have 'title_fa'"
assert 'default_variant' in product_data, "Product data should have 'default_variant'"
assert 'category' in product_data, "Product data should have 'category'"
assert 'brand' in product_data, "Product data should have 'brand'"
# Check variant details
default_variant = product_data.get('default_variant', {})
assert 'price' in default_variant, "Variant should have 'price'"
assert 'seller' in default_variant, "Variant should have 'seller'"
price = default_variant.get('price', {})
print(f"✓ Price: {price.get('selling_price', 0):,} Rials")
print(f"✓ Discount: {price.get('discount_percent', 0)}%")
# Check rating
rating = product_data.get('rating', {})
print(f"✓ Rating: {rating.get('rate', 0)}/100 ({rating.get('count', 0)} reviews)")
# Check specifications
specs = product_data.get('specifications', [])
print(f"✓ Specifications groups: {len(specs)}")
# Check images
images = product_data.get('images', {})
main_image = images.get('main', {})
print(f"✓ Has main image: {len(main_image.get('url', [])) > 0}")
# Check variants
variants = product_data.get('variants', [])
print(f"✓ Total variants: {len(variants)}")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("\n--- Test: Multiple product IDs ---")
try:
# Test with different products
search_result = search_products("گوشی موبایل", rows=15)
products = extract_products_from_search(search_result)
test_count = min(3, len(products))
for i in range(test_count):
product_id = products[i]['id']
result = get_product_details(product_id)
product_data = result.get('data', {}).get('product', {})
print(f"Product {i+1}: {product_data.get('title_fa', '')[:40]}... - Status: {result.get('status')}")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("✅ All product details tests passed!")
def test_get_product_recommendations():
"""Test get_product_recommendations tool"""
print_section("TEST SUITE: get_product_recommendations")
print("\n--- Test: Get default recommendations ---")
try:
# Get a product ID from search
search_result = search_products("laptop", rows=15)
products = extract_products_from_search(search_result)
if not products:
print("⚠️ No products found to test recommendations")
return
test_product_id = products[0]['id']
print(f"Testing with product ID: {test_product_id}")
result = get_product_recommendations(test_product_id)
assert result.get('status') == 200, f"Expected status 200, got {result.get('status')}"
data = result.get('data', {})
meta = data.get('meta', {})
print(f"✓ Status: 200")
print(f"✓ Current offset: {meta.get('current')}")
# Check available offsets
offsets = meta.get('offsets', [])
print(f"✓ Available recommendation tabs: {len(offsets)}")
for offset_info in offsets[:3]:
print(f" - {offset_info.get('type')}: {offset_info.get('title')}")
# Check recommendation data
recommendation_data = data.get('data', {})
recommended_products = recommendation_data.get('products', [])
print(f"✓ Recommended products: {len(recommended_products)}")
if recommended_products:
first_rec = recommended_products[0]
print(f" First recommendation: {first_rec.get('title_fa', '')[:50]}...")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("\n--- Test: Get specific offset recommendations ---")
try:
# Get available offsets first
result = get_product_recommendations(test_product_id)
offsets = result.get('data', {}).get('meta', {}).get('offsets', [])
if len(offsets) > 1:
# Test with second offset
second_offset = offsets[1].get('offset')
offset_type = offsets[1].get('type')
print(f"Testing offset {second_offset}: {offset_type}")
result = get_product_recommendations(test_product_id, offset=second_offset)
recommendation_data = result.get('data', {}).get('data', {})
products = recommendation_data.get('products', [])
print(f"✓ Products for offset {second_offset}: {len(products)}")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("\n--- Test: Multiple products recommendations ---")
try:
search_result = search_products("گوشی", rows=15)
products = extract_products_from_search(search_result)
test_count = min(2, len(products))
for i in range(test_count):
product_id = products[i]['id']
result = get_product_recommendations(product_id)
meta = result.get('data', {}).get('meta', {})
offsets = meta.get('offsets', [])
print(f"Product {i+1} recommendations: {len(offsets)} tabs available")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("✅ All recommendations tests passed!")
def test_search_text_lenz():
"""Test search_text_lenz tool"""
print_section("TEST SUITE: search_text_lenz")
print("\n--- Test: Basic Text-Lenz search ---")
try:
result = search_text_lenz("laptop for gaming")
assert result.get('status') == 200, f"Expected status 200, got {result.get('status')}"
data = result.get('data', {})
print(f"✓ Status: 200")
print(f"✓ Search method: {data.get('search_method')}")
print(f"✓ Text-Lenz eligible: {data.get('is_text_lenz_eligible')}")
print(f"✓ Eligibility reason: {data.get('text_lenz_eligibility')}")
# Check pagination
pager = data.get('pager', {})
print(f"✓ Current page: {pager.get('current_page')}")
print(f"✓ Total items: {pager.get('total_items')}")
print(f"✓ Total pages: {pager.get('total_pages')}")
# Check products
products = data.get('products', [])
print(f"✓ Products returned: {len(products)}")
if products:
first_product = products[0]
print(f" First product: {first_product.get('title_fa', '')[:50]}...")
# Check related searches
related_words = data.get('related_search_words', [])
print(f"✓ Related searches: {len(related_words)}")
if related_words:
print(f" Samples: {related_words[:3]}")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("\n--- Test: Natural language queries ---")
try:
natural_queries = [
"هدفون با قابلیت حذف نویز",
"laptop with good battery life",
"best smartphone for photography"
]
for query in natural_queries:
result = search_text_lenz(query)
data = result.get('data', {})
products = data.get('products', [])
eligible = data.get('is_text_lenz_eligible')
print(f"Query: '{query[:30]}...' - Products: {len(products)}, Eligible: {eligible}")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("\n--- Test: Text-Lenz with filters ---")
try:
result = search_text_lenz(
"wireless headphones",
has_ready_to_shipment=1
)
data = result.get('data', {})
products = data.get('products', [])
print(f"✓ Products with ready-to-shipment filter: {len(products)}")
# Check filters in response
filters = data.get('filters', {})
has_ready = filters.get('has_ready_to_shipment', {})
print(f"✓ Ready-to-shipment count: {has_ready.get('count', 0)}")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("\n--- Test: Text-Lenz pagination ---")
try:
# Test multiple pages
for page in [1, 2]:
result = search_text_lenz("کتاب", page=page)
data = result.get('data', {})
pager = data.get('pager', {})
products = data.get('products', [])
print(f"Page {page}: {len(products)} products, Total items: {pager.get('total_items')}")
except Exception as e:
print(f"✗ Error: {e}")
raise
print("✅ All Text-Lenz tests passed!")
def run_all_tests():
"""Run all test suites with timeout protection"""
print_section("DIGIKALA MCP SERVER - COMPREHENSIVE TEST SUITE")
start_time = time.time()
max_runtime = 300 # 5 minutes max
test_suites = [
("Currency Conversion", test_currency_conversion),
("Autocomplete", test_autocomplete_suggestions),
("Basic Search", test_search_products_basic),
("Pagination", test_search_products_pagination),
("Sorting", test_search_products_sorting),
("Price Filters", test_search_products_price_filters),
("Discount Filters", test_search_products_discount_filters),
("Stock Filters", test_search_products_stock_filters),
("Digiplus Filter", test_search_products_digiplus_filter),
("Combined Filters", test_search_products_combined_filters),
("Product Extraction", test_extract_products_from_search),
("Paginated Retrieval", test_get_products_paginated),
("Product Details", test_get_product_details),
("Product Recommendations", test_get_product_recommendations),
("Text-Lenz Search", test_search_text_lenz),
("Edge Cases", test_edge_cases),
]
passed = 0
failed = 0
try:
for suite_name, test_func in test_suites:
elapsed = time.time() - start_time
if elapsed > max_runtime:
print(f"\n⚠️ Test timeout reached ({max_runtime}s). Stopping.")
break
try:
test_func()
passed += 1
except Exception as e:
print(f"\n❌ {suite_name} FAILED: {e}")
failed += 1
print_section("TEST SUMMARY")
print(f"\n✅ Passed: {passed}/{len(test_suites)}")
if failed > 0:
print(f"❌ Failed: {failed}/{len(test_suites)}")
elapsed = time.time() - start_time
print(f"\n⏱️ Total runtime: {elapsed:.2f} seconds")
if failed == 0 and passed == len(test_suites):
print("\n🎉 ALL TESTS PASSED SUCCESSFULLY! 🎉")
return 0
else:
print(f"\n⚠️ Some tests did not complete")
return 1
except KeyboardInterrupt:
print("\n\n⚠️ Tests interrupted by user")
return 1
except Exception as e:
print(f"\n\n❌ Test suite failed with error: {e}")
return 1
if __name__ == "__main__":
exit_code = run_all_tests()
sys.exit(exit_code)
