#!/usr/bin/env python3 """ Test script to verify standardized API patterns for feature extractors. This ensures all extractors follow the documented API correctly. """ import pandas as pd import numpy as np from pathlib import Path # Import all feature extractors from user_selection import UserSelector from reaction_features import ReactionFeatureExtractor from content_features import ContentFeatureExtractor from bot_features import BotFeatureExtractor from engagement_features import EngagementFeatureCreator def test_user_selector(): """Test UserSelector API - should return DataFrame directly""" print("Testing UserSelector...") # Create dummy data total_clip_df = pd.DataFrame({ 'user_id': [1, 2, 3, 1, 2], 'id': range(5) }) discord_info_df = pd.DataFrame({ 'user_id': [1, 2, 3], 'subscription_status': ['active', 'active', 'past_due'] }) # Test API selector = UserSelector() result = selector.create_initial_features(total_clip_df, discord_info_df, verbose=False) # Verify return type assert isinstance(result, pd.DataFrame), "UserSelector should return DataFrame" assert 'user_id' in result.columns, "Should have user_id column" assert 'subscription_tier' in result.columns, "Should have subscription_tier column" print("✅ UserSelector API correct: returns DataFrame") return True def test_reaction_extractor(): """Test ReactionFeatureExtractor API - should return tuple""" print("\nTesting ReactionFeatureExtractor...") # Create dummy data reaction_df = pd.DataFrame({ 'user_id': [1, 1, 2, 2, 3], 'reaction_type': ['L', 'D', 'L', 'L', 'D'], 'clip_id': range(5) }) user_ids = pd.Series([1, 2, 3]) # Test API extractor = ReactionFeatureExtractor() result = extractor.extract_features(reaction_df, user_ids, verbose=False) # Verify return type assert isinstance(result, tuple), "ReactionFeatureExtractor should return tuple" assert len(result) == 2, "Should return tuple of length 2" assert isinstance(result[0], pd.DataFrame), "First element should be DataFrame" assert isinstance(result[1], dict), "Second element should be dict" print("✅ ReactionFeatureExtractor API correct: returns (DataFrame, dict)") return True def test_content_extractor(): """Test ContentFeatureExtractor API - should return tuple""" print("\nTesting ContentFeatureExtractor...") # Create dummy data total_clip_df = pd.DataFrame({ 'user_id': [1, 1, 2, 2, 3], 'created_at': pd.date_range('2025-01-01', periods=5), 'model_name': ['v3', 'v4p5', 'v3', 'v4', 'v4p5'] }) user_ids = pd.Series([1, 2, 3]) # Test API extractor = ContentFeatureExtractor() result = extractor.extract_features(total_clip_df, user_ids, verbose=False) # Verify return type assert isinstance(result, tuple), "ContentFeatureExtractor should return tuple" assert len(result) == 2, "Should return tuple of length 2" assert isinstance(result[0], pd.DataFrame), "First element should be DataFrame" assert isinstance(result[1], dict), "Second element should be dict" print("✅ ContentFeatureExtractor API correct: returns (DataFrame, dict)") return True def test_bot_extractor(): """Test BotFeatureExtractor API - should return tuple""" print("\nTesting BotFeatureExtractor...") # Create dummy data bots_action_df = pd.DataFrame({ 'clip_id': [1, 2, 3], 'download_audio_count': [1, 0, 2], 'share_count': [0, 1, 1] }) total_clip_df = pd.DataFrame({ 'id': [1, 2, 3, 4], 'user_id': [1, 1, 2, 3] }) user_ids = pd.Series([1, 2, 3]) features_df = pd.DataFrame({ 'user_id': [1, 2, 3], 'total_clips_created': [2, 1, 1] }) # Test API extractor = BotFeatureExtractor() result = extractor.extract_features( bots_action_df, total_clip_df, user_ids, features_df, verbose=False ) # Verify return type assert isinstance(result, tuple), "BotFeatureExtractor should return tuple" assert len(result) == 2, "Should return tuple of length 2" assert isinstance(result[0], pd.DataFrame), "First element should be DataFrame" assert isinstance(result[1], dict), "Second element should be dict" print("✅ BotFeatureExtractor API correct: returns (DataFrame, dict)") return True def test_engagement_creator(): """Test EngagementFeatureCreator API - should return tuple with updated DataFrame""" print("\nTesting EngagementFeatureCreator...") # Create dummy features DataFrame with all required columns features_df = pd.DataFrame({ 'user_id': [1, 2, 3], 'total_clips_created': [10, 5, 20], 'reaction_frequency': [0.5, 0.2, 0.8], 'total_bot_actions': [5, 2, 10], 'subscription_tier': [2, 2, 1], 'is_recent_creator': [1, 1, 0], 'clip_creation_rate': [0.5, 0.3, 1.0], # Add more columns that might be referenced 'share_count': [2, 1, 5], 'total_downloads': [10, 5, 20], 'model_diversity': [2.5, 1.5, 3.0], 'public_clip_ratio': [0.5, 0.3, 0.8], 'advanced_model_ratio': [0.1, 0.0, 0.5], 'v4p5_ratio': [0.0, 0.0, 0.4], 'daily_generation_rate': [2.0, 1.0, 5.0], 'days_creating': [30, 30, 30], 'api_usage_tier': [1, 0, 2], 'community_interaction_score': [10, 5, 20] }) # Store original column count original_cols = len(features_df.columns) # Test API creator = EngagementFeatureCreator() result = creator.create_engagement_features(features_df, verbose=False) # Verify return type assert isinstance(result, tuple), "EngagementFeatureCreator should return tuple" assert len(result) == 2, "Should return tuple of length 2" assert isinstance(result[0], pd.DataFrame), "First element should be DataFrame" assert isinstance(result[1], dict), "Second element should be dict" # Verify DataFrame is updated with new features updated_df = result[0] assert 'engagement_score' in updated_df.columns, "Should add engagement_score" assert 'user_segment' in updated_df.columns, "Should add user_segment" assert 'user_segment_encoded' in updated_df.columns, "Should add user_segment_encoded" assert 'activity_diversity' in updated_df.columns, "Should add activity_diversity" assert len(updated_df.columns) > original_cols, f"Should add new columns: had {original_cols}, now {len(updated_df.columns)}" print("✅ EngagementFeatureCreator API correct: returns (updated_DataFrame, dict)") return True def main(): """Run all API tests""" print("🧪 TESTING STANDARDIZED API PATTERNS") print("=" * 50) tests = [ test_user_selector, test_reaction_extractor, test_content_extractor, test_bot_extractor, test_engagement_creator ] results = [] for test in tests: try: results.append(test()) except Exception as e: print(f"❌ {test.__name__} failed: {e}") results.append(False) print("\n" + "=" * 50) if all(results): print("✅ ALL TESTS PASSED! API is correctly standardized.") else: print("❌ Some tests failed. Check the API implementation.") return all(results) if __name__ == "__main__": main()