Spaces:
Sleeping
Sleeping
| """ | |
| Data Enrichment Testing Suite | |
| This module provides comprehensive testing for the data_enrichment.py module. | |
| It validates all 14 API integrations, caching mechanisms, error handling, | |
| and batch processing capabilities. | |
| Usage: | |
| python test_enrichment.py | |
| Or run specific test: | |
| python -m unittest test_enrichment.TestDataEnrichmentManager.test_linkedin_enrichment | |
| Requirements: | |
| - All RAPIDAPI_KEY_* environment variables must be set | |
| - Internet connection required for API calls | |
| - Test user: 'test_user' (creates test cache directory) | |
| Test Coverage: | |
| - Initialization and configuration | |
| - Individual API endpoint testing (14 APIs) | |
| - Full contact enrichment workflow | |
| - Batch processing capabilities | |
| - Caching mechanisms and TTL | |
| - Error handling and fallback logic | |
| - Data privacy and per-user isolation | |
| - Statistics and monitoring | |
| Author: StrategyStack Team | |
| Version: 2.0.0 | |
| Last Updated: November 6, 2025 | |
| """ | |
| import unittest | |
| import os | |
| import sys | |
| import json | |
| import logging | |
| import time | |
| from datetime import datetime, timedelta | |
| # Add parent directory to path to import data_enrichment | |
| sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) | |
| from data_enrichment import DataEnrichmentManager | |
| # Configure logging for tests | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' | |
| ) | |
| logger = logging.getLogger(__name__) | |
| class TestDataEnrichmentManager(unittest.TestCase): | |
| """ | |
| Comprehensive test suite for DataEnrichmentManager. | |
| This test class validates all functionality of the data enrichment module | |
| including API integrations, caching, error handling, and batch processing. | |
| """ | |
| def setUpClass(cls): | |
| """ | |
| Set up test environment once before all tests. | |
| Creates a test user instance and validates that all required | |
| API keys are configured in the environment. | |
| """ | |
| logger.info("="*70) | |
| logger.info("Starting Data Enrichment Test Suite") | |
| logger.info("="*70) | |
| # Test user for all tests | |
| cls.test_username = "test_user" | |
| # Initialize the enrichment manager | |
| cls.manager = DataEnrichmentManager(cls.test_username) | |
| # Test data samples | |
| cls.test_linkedin_url = "https://www.linkedin.com/in/williamhgates/" | |
| cls.test_company_url = "https://www.linkedin.com/company/microsoft/" | |
| cls.test_email = "test@example.com" | |
| cls.test_domain = "microsoft.com" | |
| cls.test_twitter_handle = "BillGates" | |
| cls.test_business_query = "Microsoft Redmond WA" | |
| logger.info(f"Test manager initialized for user: {cls.test_username}") | |
| logger.info(f"Cache directory: {cls.manager.cache_dir}") | |
| def tearDownClass(cls): | |
| """Clean up after all tests complete.""" | |
| logger.info("="*70) | |
| logger.info("Data Enrichment Test Suite Completed") | |
| logger.info("="*70) | |
| def test_01_initialization(self): | |
| """ | |
| Test proper initialization of DataEnrichmentManager. | |
| Validates: | |
| - Cache directory creation | |
| - Per-user data isolation | |
| - API key configuration | |
| - Stats initialization | |
| """ | |
| logger.info("\nTest 1: Initialization and Configuration") | |
| logger.info("-" * 50) | |
| # Check cache directory exists | |
| self.assertTrue( | |
| os.path.exists(self.manager.cache_dir), | |
| "Cache directory should be created" | |
| ) | |
| # Verify per-user isolation | |
| self.assertIn( | |
| self.test_username, | |
| self.manager.cache_dir, | |
| "Cache directory should include username for data isolation" | |
| ) | |
| # Check stats initialization | |
| stats = self.manager.get_enrichment_stats() | |
| self.assertIsNotNone(stats, "Stats should be initialized") | |
| self.assertIn('total_api_calls', stats) | |
| self.assertIn('cache_hits', stats) | |
| self.assertIn('cache_misses', stats) | |
| logger.info(f"✓ Cache directory verified: {self.manager.cache_dir}") | |
| logger.info(f"✓ Per-user isolation confirmed") | |
| logger.info(f"✓ Stats initialized: {stats}") | |
| def test_02_linkedin_profile_enrichment(self): | |
| """ | |
| Test LinkedIn profile enrichment with fallback providers. | |
| Tests all 4 LinkedIn profile API providers: | |
| - Fresh LinkedIn Profile Data | |
| - RockAPI LinkedIn | |
| - LiScraper | |
| - LinkedIn Profile Scraper | |
| """ | |
| logger.info("\nTest 2: LinkedIn Profile Enrichment") | |
| logger.info("-" * 50) | |
| result = self.manager.enrich_linkedin_profile(self.test_linkedin_url) | |
| if result: | |
| logger.info(f"✓ LinkedIn profile enrichment successful") | |
| logger.info(f" Profile data keys: {list(result.keys())[:5]}...") | |
| self.assertIsInstance(result, dict, "Result should be a dictionary") | |
| else: | |
| logger.warning("✗ LinkedIn enrichment failed (may be API limit or network issue)") | |
| self.assertIsNone(result, "Failed enrichment should return None") | |
| def test_03_company_enrichment(self): | |
| """ | |
| Test LinkedIn company profile enrichment. | |
| Validates company data extraction from LinkedIn company pages. | |
| """ | |
| logger.info("\nTest 3: LinkedIn Company Enrichment") | |
| logger.info("-" * 50) | |
| result = self.manager.enrich_company(self.test_company_url) | |
| if result: | |
| logger.info(f"✓ Company enrichment successful") | |
| logger.info(f" Company data keys: {list(result.keys())[:5]}...") | |
| self.assertIsInstance(result, dict) | |
| else: | |
| logger.warning("✗ Company enrichment failed") | |
| def test_04_apollo_enrichment(self): | |
| """ | |
| Test Apollo.io person enrichment. | |
| Validates contact enrichment using email or domain. | |
| """ | |
| logger.info("\nTest 4: Apollo.io Person Enrichment") | |
| logger.info("-" * 50) | |
| result = self.manager.enrich_apollo_person(self.test_email) | |
| if result: | |
| logger.info(f"✓ Apollo enrichment successful") | |
| logger.info(f" Person data keys: {list(result.keys())[:5]}...") | |
| self.assertIsInstance(result, dict) | |
| else: | |
| logger.warning("✗ Apollo enrichment failed") | |
| def test_05_twitter_enrichment(self): | |
| """ | |
| Test Twitter user data scraping. | |
| Validates Twitter profile data extraction. | |
| """ | |
| logger.info("\nTest 5: Twitter User Scraping") | |
| logger.info("-" * 50) | |
| result = self.manager.scrape_twitter_user(self.test_twitter_handle) | |
| if result: | |
| logger.info(f"✓ Twitter scraping successful") | |
| logger.info(f" Twitter data keys: {list(result.keys())[:5]}...") | |
| self.assertIsInstance(result, dict) | |
| else: | |
| logger.warning("✗ Twitter scraping failed") | |
| def test_06_facebook_enrichment(self): | |
| """ | |
| Test Facebook data scraping. | |
| Validates Facebook profile/page data extraction. | |
| """ | |
| logger.info("\nTest 6: Facebook Data Scraping") | |
| logger.info("-" * 50) | |
| # Note: Facebook scraping requires valid page/profile URL | |
| # This test may fail without proper test data | |
| result = self.manager.scrape_facebook_data({"url": "https://www.facebook.com/Microsoft"}) | |
| if result: | |
| logger.info(f"✓ Facebook scraping successful") | |
| self.assertIsInstance(result, dict) | |
| else: | |
| logger.warning("✗ Facebook scraping failed") | |
| def test_07_google_search(self): | |
| """ | |
| Test Google Search API integration. | |
| Validates web search functionality. | |
| """ | |
| logger.info("\nTest 7: Google Search") | |
| logger.info("-" * 50) | |
| query = "Microsoft CEO" | |
| result = self.manager.search_google(query) | |
| if result: | |
| logger.info(f"✓ Google search successful") | |
| logger.info(f" Search results count: {len(result) if isinstance(result, list) else 'N/A'}") | |
| else: | |
| logger.warning("✗ Google search failed") | |
| def test_08_mediafy_enrichment(self): | |
| """ | |
| Test Mediafy API integration. | |
| Validates media data extraction. | |
| """ | |
| logger.info("\nTest 8: Mediafy Enrichment") | |
| logger.info("-" * 50) | |
| result = self.manager.enrich_mediafy(self.test_domain) | |
| if result: | |
| logger.info(f"✓ Mediafy enrichment successful") | |
| self.assertIsInstance(result, dict) | |
| else: | |
| logger.warning("✗ Mediafy enrichment failed") | |
| def test_09_local_business_scraping(self): | |
| """ | |
| Test LetScrape Local Business Data API. | |
| Validates local business information extraction. | |
| """ | |
| logger.info("\nTest 9: Local Business Scraping") | |
| logger.info("-" * 50) | |
| result = self.manager.scrape_local_business(self.test_business_query) | |
| if result: | |
| logger.info(f"✓ Local business scraping successful") | |
| logger.info(f" Business data keys: {list(result.keys())[:5]}...") | |
| self.assertIsInstance(result, dict) | |
| else: | |
| logger.warning("✗ Local business scraping failed") | |
| def test_10_website_contacts_scraping(self): | |
| """ | |
| Test LetScrape Website Contacts Scraper API. | |
| Validates contact information extraction from websites. | |
| """ | |
| logger.info("\nTest 10: Website Contacts Scraping") | |
| logger.info("-" * 50) | |
| result = self.manager.scrape_website_contacts(self.test_domain) | |
| if result: | |
| logger.info(f"✓ Website contacts scraping successful") | |
| logger.info(f" Contact data keys: {list(result.keys())[:5]}...") | |
| self.assertIsInstance(result, dict) | |
| else: | |
| logger.warning("✗ Website contacts scraping failed") | |
| def test_11_full_contact_enrichment(self): | |
| """ | |
| Test comprehensive full contact enrichment workflow. | |
| This test validates the complete enrichment pipeline that: | |
| - Enriches LinkedIn profile | |
| - Enriches company data | |
| - Enriches with Apollo | |
| - Scrapes social media | |
| - Extracts website contacts | |
| - Aggregates all data sources | |
| """ | |
| logger.info("\nTest 11: Full Contact Enrichment Workflow") | |
| logger.info("-" * 50) | |
| contact_data = { | |
| "email": self.test_email, | |
| "linkedin_url": self.test_linkedin_url, | |
| "company_url": self.test_company_url, | |
| "twitter_handle": self.test_twitter_handle, | |
| "domain": self.test_domain | |
| } | |
| result = self.manager.enrich_contact_full(contact_data) | |
| self.assertIsNotNone(result, "Full enrichment should return data") | |
| self.assertIsInstance(result, dict) | |
| logger.info(f"✓ Full enrichment completed") | |
| logger.info(f" Total data sources: {len(result)}") | |
| logger.info(f" Data keys: {list(result.keys())}") | |
| def test_12_batch_enrichment(self): | |
| """ | |
| Test batch processing capabilities. | |
| Validates processing multiple contacts efficiently with caching. | |
| """ | |
| logger.info("\nTest 12: Batch Enrichment Processing") | |
| logger.info("-" * 50) | |
| contacts = [ | |
| {"email": "test1@example.com", "linkedin_url": self.test_linkedin_url}, | |
| {"email": "test2@example.com", "domain": self.test_domain}, | |
| {"email": "test3@example.com", "twitter_handle": self.test_twitter_handle} | |
| ] | |
| results = self.manager.enrich_contacts_batch(contacts) | |
| self.assertIsInstance(results, list, "Batch results should be a list") | |
| self.assertEqual(len(results), len(contacts), "Should return result for each contact") | |
| logger.info(f"✓ Batch processing completed") | |
| logger.info(f" Processed {len(contacts)} contacts") | |
| logger.info(f" Successful enrichments: {sum(1 for r in results if r)}") | |
| def test_13_caching_mechanism(self): | |
| """ | |
| Test caching functionality and cache hits. | |
| Validates: | |
| - Data is cached after first API call | |
| - Subsequent calls use cache (faster) | |
| - Cache stats are updated correctly | |
| """ | |
| logger.info("\nTest 13: Caching Mechanism") | |
| logger.info("-" * 50) | |
| # Clear stats | |
| self.manager.stats = { | |
| 'total_api_calls': 0, | |
| 'cache_hits': 0, | |
| 'cache_misses': 0, | |
| 'total_enrichments': 0, | |
| 'successful_enrichments': 0, | |
| 'failed_enrichments': 0 | |
| } | |
| # First call - should be cache miss | |
| test_data = {"email": "cache_test@example.com", "linkedin_url": self.test_linkedin_url} | |
| result1 = self.manager.enrich_contact_full(test_data) | |
| stats_after_first = self.manager.get_enrichment_stats() | |
| initial_cache_misses = stats_after_first['cache_misses'] | |
| # Second call - should use cache | |
| result2 = self.manager.enrich_contact_full(test_data) | |
| stats_after_second = self.manager.get_enrichment_stats() | |
| logger.info(f"✓ First call stats: {stats_after_first}") | |
| logger.info(f"✓ Second call stats: {stats_after_second}") | |
| logger.info(f" Cache hit improvement: {stats_after_second['cache_hits'] > stats_after_first['cache_hits']}") | |
| # Verify caching worked | |
| self.assertGreater( | |
| stats_after_second['cache_hits'], | |
| stats_after_first['cache_hits'], | |
| "Second call should increase cache hits" | |
| ) | |
| def test_14_cache_invalidation(self): | |
| """ | |
| Test cache invalidation functionality. | |
| Validates: | |
| - Cache can be invalidated for specific contact | |
| - Cache can be cleared entirely | |
| - TTL-based expiration works correctly | |
| """ | |
| logger.info("\nTest 14: Cache Invalidation") | |
| logger.info("-" * 50) | |
| test_key = "test_invalidation@example.com" | |
| # Create cached data | |
| self.manager._cache_result("test_api", test_key, {"test": "data"}) | |
| # Verify it's cached | |
| cached = self.manager._get_cached_result("test_api", test_key) | |
| self.assertIsNotNone(cached, "Data should be cached") | |
| # Invalidate cache for this key | |
| self.manager.invalidate_cache(test_key) | |
| # Verify it's gone | |
| cached_after = self.manager._get_cached_result("test_api", test_key) | |
| self.assertIsNone(cached_after, "Cache should be invalidated") | |
| logger.info(f"✓ Cache invalidation successful") | |
| def test_15_error_handling(self): | |
| """ | |
| Test error handling and graceful degradation. | |
| Validates: | |
| - Invalid inputs return None | |
| - API failures are caught and logged | |
| - Fallback providers are tried | |
| - Stats track failures correctly | |
| """ | |
| logger.info("\nTest 15: Error Handling") | |
| logger.info("-" * 50) | |
| # Test with invalid data | |
| invalid_url = "not_a_valid_url" | |
| result = self.manager.enrich_linkedin_profile(invalid_url) | |
| self.assertIsNone(result, "Invalid input should return None") | |
| logger.info(f"✓ Error handling verified") | |
| logger.info(f" Invalid input handled gracefully") | |
| def test_16_statistics_tracking(self): | |
| """ | |
| Test statistics tracking functionality. | |
| Validates: | |
| - API call stats are tracked | |
| - Success/failure rates are calculated | |
| - Cache hit ratios are recorded | |
| """ | |
| logger.info("\nTest 16: Statistics Tracking") | |
| logger.info("-" * 50) | |
| stats = self.manager.get_enrichment_stats() | |
| self.assertIsNotNone(stats) | |
| self.assertIn('total_api_calls', stats) | |
| self.assertIn('cache_hits', stats) | |
| self.assertIn('cache_misses', stats) | |
| self.assertIn('total_enrichments', stats) | |
| self.assertIn('successful_enrichments', stats) | |
| self.assertIn('failed_enrichments', stats) | |
| # Calculate cache hit rate | |
| total_cache_ops = stats['cache_hits'] + stats['cache_misses'] | |
| if total_cache_ops > 0: | |
| cache_hit_rate = (stats['cache_hits'] / total_cache_ops) * 100 | |
| logger.info(f"✓ Cache hit rate: {cache_hit_rate:.2f}%") | |
| # Calculate success rate | |
| if stats['total_enrichments'] > 0: | |
| success_rate = (stats['successful_enrichments'] / stats['total_enrichments']) * 100 | |
| logger.info(f"✓ Enrichment success rate: {success_rate:.2f}%") | |
| logger.info(f"✓ Statistics tracking verified") | |
| logger.info(f" Current stats: {stats}") | |
| def test_17_data_privacy_isolation(self): | |
| """ | |
| Test per-user data isolation. | |
| Validates: | |
| - Each user has separate cache directory | |
| - User data cannot access other user data | |
| - Cache paths include username | |
| """ | |
| logger.info("\nTest 17: Data Privacy and Per-User Isolation") | |
| logger.info("-" * 50) | |
| # Create second user manager | |
| user2_manager = DataEnrichmentManager("user2_test") | |
| # Verify different cache directories | |
| self.assertNotEqual( | |
| self.manager.cache_dir, | |
| user2_manager.cache_dir, | |
| "Different users should have different cache directories" | |
| ) | |
| self.assertIn(self.test_username, self.manager.cache_dir) | |
| self.assertIn("user2_test", user2_manager.cache_dir) | |
| logger.info(f"✓ Per-user isolation verified") | |
| logger.info(f" User 1 cache: {self.manager.cache_dir}") | |
| logger.info(f" User 2 cache: {user2_manager.cache_dir}") | |
| if __name__ == "__main__": | |
| # Run all tests | |
| logger.info("\n" + "="*70) | |
| logger.info("STARTING DATA ENRICHMENT TEST SUITE") | |
| logger.info("="*70 + "\n") | |
| # Create test suite | |
| suite = unittest.TestLoader().loadTestsFromTestCase(TestDataEnrichmentManager) | |
| # Run tests with verbose output | |
| runner = unittest.TextTestRunner(verbosity=2) | |
| result = runner.run(suite) | |
| # Print summary | |
| logger.info("\n" + "="*70) | |
| logger.info("TEST SUMMARY") | |
| logger.info("="*70) | |
| logger.info(f"Tests run: {result.testsRun}") | |
| logger.info(f"Successes: {result.testsRun - len(result.failures) - len(result.errors)}") | |
| logger.info(f"Failures: {len(result.failures)}") | |
| logger.info(f"Errors: {len(result.errors)}") | |
| logger.info("="*70 + "\n") | |
| # Exit with appropriate code | |
| sys.exit(0 if result.wasSuccessful() else 1) |