""" test_heatmap.py — Integration tests for the Heatmap Audit Generator. Tests each component (scraper, analyzer, PDF generator) with real businesses. Run: python -m scripts.heatmap_audit.test_heatmap """ import asyncio import os import sys import time import json import logging from pathlib import Path # Add project root to path PROJECT_ROOT = str(Path(__file__).resolve().parents[2]) if PROJECT_ROOT not in sys.path: sys.path.insert(0, PROJECT_ROOT) import aiohttp from scripts.heatmap_audit.scraper import ( scrape_all, fetch_pagespeed, fetch_brave_search, fetch_website_basics, ) from scripts.heatmap_audit.analyzer import ( analyze, calculate_scores, calculate_overall_score, generate_recommendations, ) from scripts.heatmap_audit.pdf_generator import generate_pdf # Setup logging logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S", ) logger = logging.getLogger(__name__) # Test businesses TEST_BUSINESSES = [ { "url": "https://www.jimsmowing.com.au", "name": "Jim's Mowing", "location": "Australia", }, { "url": "https://www.kennards.com.au", "name": "Kennards Hire", "location": "Australia", }, { "url": "https://www.bunnings.com.au", "name": "Bunnings", "location": "Australia", }, ] OUTPUT_DIR = os.path.join(PROJECT_ROOT, "data", "heatmap_reports", "test") def _result_tag(passed: bool) -> str: return "PASS" if passed else "FAIL" class TestRunner: def __init__(self): self.passed = 0 self.failed = 0 self.results = [] def check(self, name: str, condition: bool, detail: str = ""): tag = _result_tag(condition) if condition: self.passed += 1 else: self.failed += 1 self.results.append((name, condition, detail)) print(f" [{tag}] {name}{f' — {detail}' if detail else ''}") def summary(self): total = self.passed + self.failed print(f"\n{'='*60}") print(f" RESULTS: {self.passed}/{total} passed, {self.failed} failed") print(f"{'='*60}") return self.failed == 0 async def test_scraper_pagespeed(session, runner): """Test PageSpeed Insights API with a real URL.""" print("\n--- Test: PageSpeed Insights ---") url = "https://www.jimsmowing.com.au" result = await fetch_pagespeed(session, url) runner.check( "PageSpeed returns data", result is not None, f"Got {len(result) if result else 0} fields", ) if result: runner.check( "Mobile score present", "mobile_performance_score" in result, f"Score: {result.get('mobile_performance_score', 'N/A')}", ) runner.check( "Desktop score present", "desktop_performance_score" in result, f"Score: {result.get('desktop_performance_score', 'N/A')}", ) runner.check( "Mobile FCP present", "mobile_fcp_ms" in result, f"FCP: {result.get('mobile_fcp_ms', 'N/A')}ms", ) runner.check( "Scores are valid range", 0 <= result.get("mobile_performance_score", -1) <= 100, ) async def test_scraper_brave(session, runner): """Test Brave Search API.""" print("\n--- Test: Brave Search ---") result = await fetch_brave_search(session, "Jim's Mowing", "Australia") runner.check( "Brave Search returns data", result is not None, f"Got {len(result) if result else 0} fields", ) if result: runner.check( "Search result count present", "search_result_count" in result, f"Results: {result.get('search_result_count', 0)}", ) runner.check( "Social profiles detected", "social_profiles" in result, f"Found: {result.get('social_profiles', [])}", ) runner.check( "Has search results", result.get("search_result_count", 0) > 0, ) async def test_scraper_website(session, runner): """Test direct website scraping.""" print("\n--- Test: Website Basics ---") result = await fetch_website_basics(session, "https://www.jimsmowing.com.au") runner.check( "Website scrape returns data", result is not None, f"Got {len(result) if result else 0} fields", ) if result: runner.check( "Site is reachable", result.get("is_reachable", False), ) runner.check( "SSL detected", "has_ssl" in result, f"SSL: {result.get('has_ssl')}", ) runner.check( "Title tag found", result.get("has_title", False), f"Title: {result.get('title_tag', '')[:50]}", ) runner.check( "Meta description check", "has_meta_description" in result, f"Has meta: {result.get('has_meta_description')}", ) runner.check( "H1 count tracked", "h1_count" in result, f"H1s: {result.get('h1_count', 0)}", ) async def test_full_scrape(session, runner): """Test scrape_all for all test businesses.""" print("\n--- Test: Full Scrape (all businesses) ---") for biz in TEST_BUSINESSES: start = time.time() result = await scrape_all( url=biz["url"], business_name=biz["name"], location=biz["location"], session=session, ) elapsed = time.time() - start runner.check( f"Scrape {biz['name']}", result is not None, f"{elapsed:.1f}s", ) if result: runner.check( f" {biz['name']} has pagespeed data", bool(result.get("pagespeed")), ) runner.check( f" {biz['name']} has brave data", bool(result.get("brave")), ) runner.check( f" {biz['name']} has website data", bool(result.get("website")), ) def test_analyzer(scraped_data_list, runner): """Test analyzer with scraped data.""" print("\n--- Test: Analyzer ---") for scraped in scraped_data_list: name = scraped.get("business_name", "Unknown") scores, recommendations = analyze(scraped) runner.check( f"Analyze {name} — scores generated", bool(scores), f"{len(scores)} categories", ) overall = scores.get("overall", {}).get("score", -1) runner.check( f" {name} overall score valid", 0 <= overall <= 100, f"Score: {overall}", ) runner.check( f" {name} has recommendations", len(recommendations) > 0, f"{len(recommendations)} recommendations", ) # Check all categories present expected_cats = ["website_speed", "mobile_experience", "online_presence", "seo_basics", "ai_readiness"] for cat in expected_cats: runner.check( f" {name} has {cat}", cat in scores, f"Score: {scores.get(cat, {}).get('score', 'N/A')}", ) # Print score summary print(f" Scores for {name}:") for cat in expected_cats: s = scores.get(cat, {}).get("score", 0) print(f" {cat}: {s}/100") print(f" OVERALL: {overall}/100") def test_pdf_generator(scraped_data_list, runner): """Test PDF generation for all businesses.""" print("\n--- Test: PDF Generator ---") os.makedirs(OUTPUT_DIR, exist_ok=True) for scraped in scraped_data_list: name = scraped.get("business_name", "Unknown") scores, recommendations = analyze(scraped) safe_name = name.lower().replace("'", "").replace(" ", "_") pdf_path = os.path.join(OUTPUT_DIR, f"{safe_name}_ai_score.pdf") try: result_path = generate_pdf( business_name=name, scores=scores, recommendations=recommendations, output_path=pdf_path, phone_cta="+61 7 3130 4377", website_url=scraped.get("url", ""), ) runner.check( f"PDF generated for {name}", os.path.exists(result_path), f"Path: {result_path}", ) if os.path.exists(result_path): size = os.path.getsize(result_path) runner.check( f" {name} PDF size valid", size > 1024, f"Size: {size:,} bytes", ) except Exception as e: runner.check(f"PDF generation for {name}", False, str(e)) def test_analyzer_edge_cases(runner): """Test analyzer with edge case data.""" print("\n--- Test: Analyzer Edge Cases ---") # Empty data empty = { "business_name": "Empty Corp", "url": "", "pagespeed": {}, "brave": {}, "website": {}, } scores, recs = analyze(empty) runner.check( "Empty data produces valid scores", "overall" in scores, f"Overall: {scores.get('overall', {}).get('score', 'N/A')}", ) # Unreachable site unreachable = { "business_name": "Dead Site Ltd", "url": "https://definitely-not-a-real-site-abc123.com", "pagespeed": {}, "brave": {"search_result_count": 5, "social_profile_count": 0, "has_google_business": False, "has_directory_listings": False, "review_mentions": 0}, "website": {"is_reachable": False, "has_ssl": False, "status_code": 0}, } scores2, recs2 = analyze(unreachable) runner.check( "Unreachable site scores low", scores2.get("overall", {}).get("score", 100) < 50, f"Overall: {scores2.get('overall', {}).get('score', 'N/A')}", ) runner.check( "Unreachable site has high AI readiness", scores2.get("ai_readiness", {}).get("score", 0) >= 80, f"AI Readiness: {scores2.get('ai_readiness', {}).get('score', 'N/A')}", ) # Perfect site perfect = { "business_name": "Perfect Inc", "url": "https://perfect.com", "pagespeed": { "mobile_performance_score": 95, "desktop_performance_score": 99, "mobile_seo_score": 100, "mobile_accessibility_score": 98, "mobile_best_practices_score": 100, "mobile_fcp_ms": 800, "mobile_lcp_ms": 1200, "mobile_speed_index_ms": 1000, "mobile_tti_ms": 1500, "mobile_cls": 0.02, "has_mobile_viewport": True, }, "brave": { "search_result_count": 5000, "social_profile_count": 5, "social_profiles": ["Facebook", "Instagram", "LinkedIn", "YouTube", "Twitter"], "has_google_business": True, "has_directory_listings": True, "review_mentions": 5, }, "website": { "is_reachable": True, "has_ssl": True, "has_mobile_viewport": True, "has_title": True, "title_tag": "Perfect Inc — The Best Company", "has_meta_description": True, "meta_description": "We are the best company with amazing services for everyone.", "h1_count": 1, "has_schema_markup": True, "has_og_tags": True, "total_images": 10, "images_without_alt": 0, "has_chat_widget": True, "has_booking_system": True, "has_phone_visible": True, "page_social_links": ["facebook", "instagram"], "word_count": 1500, }, } scores3, recs3 = analyze(perfect) runner.check( "Perfect site scores high", scores3.get("overall", {}).get("score", 0) >= 70, f"Overall: {scores3.get('overall', {}).get('score', 'N/A')}", ) runner.check( "Perfect site has low AI readiness", scores3.get("ai_readiness", {}).get("score", 100) < 50, f"AI Readiness: {scores3.get('ai_readiness', {}).get('score', 'N/A')}", ) async def main(): print("=" * 60) print(" HEATMAP AUDIT GENERATOR — TEST SUITE") print(" Testing with 3 real Australian businesses") print("=" * 60) runner = TestRunner() total_start = time.time() # Create session connector = aiohttp.TCPConnector(ssl=False, limit=20) async with aiohttp.ClientSession(connector=connector) as session: # Test individual scrapers await test_scraper_pagespeed(session, runner) await test_scraper_brave(session, runner) await test_scraper_website(session, runner) # Full scrape of all test businesses scraped_data_list = [] for biz in TEST_BUSINESSES: data = await scrape_all( url=biz["url"], business_name=biz["name"], location=biz["location"], session=session, ) scraped_data_list.append(data) # Test analyzer test_analyzer(scraped_data_list, runner) # Test edge cases test_analyzer_edge_cases(runner) # Test PDF generation test_pdf_generator(scraped_data_list, runner) total_elapsed = time.time() - total_start print(f"\nTotal test time: {total_elapsed:.1f}s") all_passed = runner.summary() # Write test results JSON results_path = os.path.join(OUTPUT_DIR, "test_results.json") os.makedirs(OUTPUT_DIR, exist_ok=True) with open(results_path, "w") as f: json.dump({ "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S"), "total_tests": runner.passed + runner.failed, "passed": runner.passed, "failed": runner.failed, "elapsed_seconds": round(total_elapsed, 1), "tests": [ {"name": name, "passed": passed, "detail": detail} for name, passed, detail in runner.results ], }, f, indent=2) print(f"\nTest results saved to: {results_path}") print(f"Test PDFs saved to: {OUTPUT_DIR}") return 0 if all_passed else 1 if __name__ == "__main__": exit_code = asyncio.run(main()) sys.exit(exit_code)