From 49b8c3653f8efde18d93786ab4ff1d0d7d60fe50 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 30 Nov 2025 00:37:40 +0000 Subject: [PATCH] test: Add comprehensive test suite for Amazon France scraper - Test basic search functionality - Test multiple queries with delays - Test anti-detection system (proxies, user-agents) - Detailed logging for debugging - Success/failure reporting --- test_amazon_scraper.py | 170 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 170 insertions(+) create mode 100755 test_amazon_scraper.py diff --git a/test_amazon_scraper.py b/test_amazon_scraper.py new file mode 100755 index 0000000..d33b9f5 --- /dev/null +++ b/test_amazon_scraper.py @@ -0,0 +1,170 @@ +#!/usr/bin/env python3 +""" +Script de test pour le scraper Amazon France +Teste le système anti-détection et l'extraction des produits +""" + +import asyncio +import logging +import sys +from pathlib import Path + +# Ajouter le répertoire app au path +sys.path.insert(0, str(Path(__file__).parent)) + +from app.services.amazon_scraper import ( + scrape_amazon_search, + test_amazon_scraper, +) + +# Configuration du logging +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' +) + +logger = logging.getLogger(__name__) + + +async def test_basic_search(): + """Test basique de recherche""" + logger.info("=" * 80) + logger.info("TEST 1: Recherche basique - 'aspirateur'") + logger.info("=" * 80) + + products = await scrape_amazon_search("aspirateur", max_results=5) + + if not products: + logger.error("❌ Aucun produit trouvé - possibilité de détection ou problème réseau") + return False + + logger.info(f"✅ {len(products)} produits trouvés") + + for idx, product in enumerate(products, 1): + logger.info(f"\n{idx}. {product.title[:60]}...") + logger.info(f" 💰 Prix: {product.price}€" + (f" (était {product.original_price}€)" if product.original_price else "")) + logger.info(f" ⭐ Note: {product.rating}/5" if product.rating else " ⭐ Pas de note") + logger.info(f" 📦 {'En stock' if product.in_stock else 'Indisponible'}") + logger.info(f" {'🚚 Prime' if product.prime else '📮 Standard'}") + logger.info(f" {'📢 Sponsorisé' if product.sponsored else '🔍 Organique'}") + + return True + + +async def test_multiple_queries(): + """Test avec plusieurs requêtes différentes""" + logger.info("\n" + "=" * 80) + logger.info("TEST 2: Requêtes multiples") + logger.info("=" * 80) + + queries = ["clavier", "souris", "casque"] + results = {} + + for query in queries: + logger.info(f"\n🔍 Recherche: '{query}'") + products = await scrape_amazon_search(query, max_results=3) + results[query] = len(products) + logger.info(f" ✅ {len(products)} produits trouvés") + + # Délai entre requêtes pour respecter les bonnes pratiques + await asyncio.sleep(3) + + logger.info("\n📊 Résumé:") + for query, count in results.items(): + logger.info(f" • {query}: {count} produits") + + total = sum(results.values()) + if total > 0: + logger.info(f"\n✅ Total: {total} produits extraits") + return True + else: + logger.error("\n❌ Aucun produit extrait - problème possible") + return False + + +async def test_anti_detection(): + """Test du système anti-détection""" + logger.info("\n" + "=" * 80) + logger.info("TEST 3: Vérification anti-détection") + logger.info("=" * 80) + + from app.core.search_config import AMAZON_PROXY_LIST_RAW, USER_AGENT_POOL + from app.services.amazon_scraper import get_random_proxy, AMAZON_USER_AGENTS + + logger.info(f"✓ {len(AMAZON_PROXY_LIST_RAW)} proxies disponibles") + logger.info(f"✓ {len(USER_AGENT_POOL)} User-Agents standards") + logger.info(f"✓ {len(AMAZON_USER_AGENTS)} User-Agents Amazon spécifiques") + + # Test proxy + proxy = get_random_proxy() + if proxy: + logger.info(f"✓ Proxy test: {proxy['server']}") + else: + logger.warning("⚠️ Pas de proxy configuré") + + # Test d'une recherche simple + logger.info("\n🧪 Test de recherche avec anti-détection...") + products = await scrape_amazon_search("livre", max_results=3) + + if products: + logger.info(f"✅ Anti-détection fonctionnel - {len(products)} produits extraits") + return True + else: + logger.error("❌ Échec - possibilité de blocage") + return False + + +async def run_all_tests(): + """Lance tous les tests""" + logger.info("\n" + "=" * 80) + logger.info("🚀 DÉMARRAGE DES TESTS DU SCRAPER AMAZON FRANCE") + logger.info("=" * 80) + + tests = [ + ("Recherche basique", test_basic_search), + ("Requêtes multiples", test_multiple_queries), + ("Anti-détection", test_anti_detection), + ] + + results = {} + + for test_name, test_func in tests: + try: + logger.info(f"\n▶️ Exécution: {test_name}") + success = await test_func() + results[test_name] = "✅ PASS" if success else "❌ FAIL" + except Exception as e: + logger.error(f"❌ Erreur dans {test_name}: {e}", exc_info=True) + results[test_name] = "❌ ERROR" + + # Résumé final + logger.info("\n" + "=" * 80) + logger.info("📊 RÉSUMÉ DES TESTS") + logger.info("=" * 80) + + for test_name, result in results.items(): + logger.info(f"{result} - {test_name}") + + passed = sum(1 for r in results.values() if "PASS" in r) + total = len(results) + + logger.info(f"\n🎯 Score: {passed}/{total} tests réussis") + + if passed == total: + logger.info("✅ TOUS LES TESTS ONT RÉUSSI!") + return True + else: + logger.warning("⚠️ Certains tests ont échoué") + return False + + +if __name__ == "__main__": + try: + success = asyncio.run(run_all_tests()) + sys.exit(0 if success else 1) + except KeyboardInterrupt: + logger.info("\n⏸️ Tests interrompus par l'utilisateur") + sys.exit(130) + except Exception as e: + logger.error(f"❌ Erreur fatale: {e}", exc_info=True) + sys.exit(1)