""" Test 5: Modular Application Structure Purpose: Test multi-file application structure in Container Runtime """ import streamlit as st import sys import os from pathlib import Path # Page configuration st.set_page_config( page_title="Spark POC - Modular Test", page_icon="🏗️", layout="wide" ) st.title("🏗️ Spark POC - Modular Structure Test") st.write("Testing multi-file application architecture in Container Runtime") # Test 5.1: Module Import Testing st.header("Test 5.1: Module Import Testing") col1, col2 = st.columns(2) with col1: st.subheader("Utils Module Imports") # Test importing utils modules import_results = {} try: import database import_results['database'] = "✅ Success" st.success("✅ database module imported successfully") except ImportError as e: import_results['database'] = f"❌ Failed: {str(e)}" st.error(f"❌ database import failed: {str(e)}") try: import data_processor import_results['data_processor'] = "✅ Success" st.success("✅ data_processor module imported successfully") except ImportError as e: import_results['data_processor'] = f"❌ Failed: {str(e)}" st.error(f"❌ data_processor import failed: {str(e)}") with col2: st.subheader("File System Analysis") # Show current working directory and file structure current_dir = os.getcwd() st.write(f"**Current Directory**: {current_dir}") # Check if utils directory exists utils_path = os.path.join(current_dir, 'utils') if os.path.exists(utils_path): st.success("✅ utils/ directory found") # List files in utils directory try: utils_files = os.listdir(utils_path) st.write("**Utils Directory Contents**:") for file in utils_files: if file.endswith('.py'): st.write(f"- {file}") except Exception as e: st.warning(f"Could not list utils directory: {e}") else: st.error("❌ utils/ directory not found") # Show Python path st.write("**Python Path (first 3 entries)**:") for i, path in enumerate(sys.path[:3]): st.write(f"{i+1}. {path}") # Test 5.2: Function Testing from Modules st.header("Test 5.2: Cross-Module Functionality") if 'database' in import_results and 'Success' in import_results['database']: try: from database import get_sample_songs, get_config, test_connection col1, col2 = st.columns(2) with col1: st.subheader("Database Utils Functions") if st.button("Test Database Config"): config = get_config() st.json(config) if st.button("Test Connection"): conn_result = test_connection() if "✅" in conn_result: st.success(conn_result) else: st.warning(conn_result) with col2: st.subheader("Sample Data Functions") if st.button("Load Sample Songs"): songs_df = get_sample_songs() st.write(f"Loaded {len(songs_df)} sample songs") st.dataframe(songs_df, use_container_width=True) # Store in session state for other tests st.session_state.sample_songs = songs_df except Exception as e: st.error(f"Error testing database functions: {str(e)}") # Test 5.3: Data Processing Module if 'data_processor' in import_results and 'Success' in import_results['data_processor']: try: from data_processor import process_video_data, calculate_analytics, generate_insights st.subheader("Data Processing Functions") col1, col2 = st.columns(2) with col1: if st.button("Test Video Data Processing"): # Create mock video data mock_videos = [ { 'video_id': 'test_001', 'creator': 'test_creator_1', 'views': 10000, 'likes': 500, 'shares': 50, 'hearts': 450 }, { 'video_id': 'test_002', 'creator': 'test_creator_2', 'views': 5000, 'likes': 200, 'shares': 25, 'hearts': 180 } ] processed_df = process_video_data(mock_videos) st.write(f"Processed {len(processed_df)} video records") st.dataframe(processed_df, use_container_width=True) # Store for analytics test st.session_state.processed_videos = processed_df with col2: if hasattr(st.session_state, 'processed_videos') and st.button("Test Analytics Calculation"): analytics = calculate_analytics(st.session_state.processed_videos) st.write("**Analytics Results:**") st.json(analytics) # Test insights generation if st.session_state.processed_videos is not None: insights = generate_insights(analytics, st.session_state.processed_videos) st.write("**Generated Insights:**") for insight in insights: st.write(f"- {insight}") except Exception as e: st.error(f"Error testing data processing functions: {str(e)}") # Test 5.4: Integration Test st.header("Test 5.4: Module Integration Test") if all('Success' in result for result in import_results.values()): if st.button("Run Full Integration Test"): try: # Simulate full workflow from database import get_sample_songs, format_tiktok_url from data_processor import process_video_data, calculate_analytics # Step 1: Get song data songs_df = get_sample_songs() st.write(f"✅ Step 1: Loaded {len(songs_df)} songs from database module") # Step 2: Generate TikTok URL for first song if len(songs_df) > 0: first_song = songs_df.iloc[0] tiktok_url = format_tiktok_url(first_song['track'], first_song['tiktok_track_id']) st.write(f"✅ Step 2: Generated TikTok URL: {tiktok_url}") # Step 3: Mock video processing mock_videos = [ {'video_id': f'vid_{i}', 'creator': f'creator_{i}', 'views': 1000 + i*100, 'likes': 50 + i*10, 'shares': 5 + i, 'hearts': 45 + i*8} for i in range(5) ] processed_videos = process_video_data(mock_videos) st.write(f"✅ Step 3: Processed {len(processed_videos)} video records") # Step 4: Calculate analytics analytics = calculate_analytics(processed_videos) st.write(f"✅ Step 4: Generated analytics with {analytics['total_videos']} videos and {analytics['total_views']} total views") st.success("🎉 Full integration test successful!") except Exception as e: st.error(f"Integration test failed: {str(e)}") # Test Results Summary st.header("🔍 Modular Structure Test Results") test_results = { "Module Imports": "✅ Working" if all('Success' in result for result in import_results.values()) else "❌ Failed", "File Structure": "✅ Working" if os.path.exists(os.path.join(os.getcwd(), 'utils')) else "❌ Failed", "Cross-Module Functions": "✅ Working" if 'utils.database' in import_results and 'Success' in import_results['utils.database'] else "⚠️ Limited", "Data Processing": "✅ Working" if 'utils.data_processor' in import_results and 'Success' in import_results['utils.data_processor'] else "⚠️ Limited", "Module Integration": "✅ Ready" if all('Success' in result for result in import_results.values()) else "⚠️ Needs attention" } for test, status in test_results.items(): st.write(f"**{test}**: {status}") # Environment Information st.header("📊 Container Runtime Module Environment") col1, col2 = st.columns(2) with col1: st.subheader("File System Details") # Current working directory contents try: current_files = [f for f in os.listdir('.') if f.endswith('.py')] st.write("**Python files in current directory:**") for file in current_files[:10]: # Limit to first 10 st.write(f"- {file}") if len(current_files) > 10: st.write(f"... and {len(current_files) - 10} more files") except Exception as e: st.write(f"Could not list directory: {e}") with col2: st.subheader("Module Loading Details") st.write("**Import Results Summary:**") for module, result in import_results.items(): status_emoji = "✅" if "Success" in result else "❌" st.write(f"{status_emoji} **{module}**: {result}") # Notes section st.header("📝 Modular Structure Notes") st.text_area( "Record observations about module structure and cross-file imports:", placeholder="Enter observations about import behavior, file organization, performance...", height=100 ) # Container Runtime Specific Notes st.header("🐳 Container Runtime Module Insights") st.info(""" 💡 **Key Findings for POC Development:** **Module Structure**: Container Runtime supports multi-file Python applications with proper import structure. **File Organization**: Utils modules can be organized in subdirectories and imported using standard Python import syntax. **Cross-Module Communication**: Functions and classes can be shared between modules, enabling modular POC architecture. **Integration Ready**: The environment supports the planned POC architecture with separate modules for database access, data processing, and API integration. """)