Upload test_system.py with huggingface_hub

test_system.py

@@ -0,0 +1,509 @@
+#!/usr/bin/env python3
+"""
+Comprehensive Test and Validation Script
+
+Tests all components of the GPU monitoring and fan control system
+to ensure proper functionality and integration.
+"""
+
+import sys
+import os
+import time
+import json
+import logging
+import subprocess
+import threading
+from pathlib import Path
+from typing import Dict, List, Any, Optional
+
+# Add the project directory to Python path
+sys.path.insert(0, str(Path(__file__).parent))
+
+from gpu_monitoring import GPUManager, GPUStatus
+from gpu_fan_controller import FanController, FanMode, ProfileType
+from alert_system import AlertManager, AlertThreshold
+from performance_optimizer import SystemOptimizer
+from web_interface import app as web_app
+
+logger = logging.getLogger(__name__)
+
+
+class SystemTester:
+    """Comprehensive system tester."""
+    
+    def __init__(self):
+        self.test_results = []
+        self.gpu_manager = None
+        self.fan_controller = None
+        self.alert_manager = None
+        self.optimizer = None
+        
+        # Setup logging
+        self.setup_logging()
+    
+    def setup_logging(self):
+        """Setup test logging."""
+        logging.basicConfig(
+            level=logging.INFO,
+            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+            handlers=[
+                logging.FileHandler('test_results.log'),
+                logging.StreamHandler(sys.stdout)
+            ]
+        )
+    
+    def log_test(self, test_name: str, success: bool, message: str = ""):
+        """Log test result."""
+        status = "PASS" if success else "FAIL"
+        logger.info(f"[{status}] {test_name}: {message}")
+        self.test_results.append({
+            "test": test_name,
+            "success": success,
+            "message": message,
+            "timestamp": time.time()
+        })
+    
+    def test_gpu_detection(self) -> bool:
+        """Test GPU detection and initialization."""
+        try:
+            self.gpu_manager = GPUManager()
+            success = self.gpu_manager.initialize()
+            
+            if success:
+                gpus = self.gpu_manager.get_gpu_list()
+                gpu_info = self.gpu_manager.get_gpu_info()
+                
+                self.log_test(
+                    "GPU Detection",
+                    True,
+                    f"Found {len(gpus)} GPU(s): {', '.join(gpus)}"
+                )
+                
+                # Test status collection
+                status = self.gpu_manager.get_status()
+                if status:
+                    self.log_test(
+                        "GPU Status Collection",
+                        True,
+                        f"Successfully collected status for {len(status)} GPU(s)"
+                    )
+                    return True
+                else:
+                    self.log_test("GPU Status Collection", False, "No status data collected")
+                    return False
+            else:
+                self.log_test("GPU Detection", False, "Failed to initialize GPU manager")
+                return False
+                
+        except Exception as e:
+            self.log_test("GPU Detection", False, f"Exception: {e}")
+            return False
+    
+    def test_fan_control(self) -> bool:
+        """Test fan control functionality."""
+        try:
+            self.fan_controller = FanController()
+            success = self.fan_controller.initialize()
+            
+            if success:
+                # Test profile management
+                profiles = self.fan_controller.get_profiles()
+                self.log_test(
+                    "Fan Profiles",
+                    True,
+                    f"Loaded {len(profiles)} profiles: {', '.join(profiles.keys())}"
+                )
+                
+                # Test profile switching
+                if "balanced" in profiles:
+                    success = self.fan_controller.set_profile("balanced")
+                    self.log_test("Profile Switching", success, "Switched to balanced profile")
+                
+                # Test manual control
+                self.fan_controller.set_manual_pwm(100)
+                time.sleep(1)  # Allow time for change
+                
+                status = self.fan_controller.get_status()
+                if status and status.manual_override:
+                    self.log_test("Manual Control", True, "Manual PWM control working")
+                else:
+                    self.log_test("Manual Control", False, "Manual control not working")
+                
+                return True
+            else:
+                self.log_test("Fan Control", False, "Failed to initialize fan controller")
+                return False
+                
+        except Exception as e:
+            self.log_test("Fan Control", False, f"Exception: {e}")
+            return False
+    
+    def test_alert_system(self) -> bool:
+        """Test alert system functionality."""
+        try:
+            self.alert_manager = AlertManager()
+            
+            # Test threshold management
+            threshold = AlertThreshold(
+                metric="temperature",
+                threshold=50.0,
+                operator=">=",
+                duration=5,
+                enabled=True,
+                cooldown=60
+            )
+            
+            self.alert_manager.add_threshold(threshold)
+            self.log_test("Alert Thresholds", True, "Successfully added alert threshold")
+            
+            # Test alert history
+            history = self.alert_manager.get_alert_history(1)
+            self.log_test("Alert History", True, f"Retrieved {len(history)} alerts from history")
+            
+            return True
+            
+        except Exception as e:
+            self.log_test("Alert System", False, f"Exception: {e}")
+            return False
+    
+    def test_performance_optimizer(self) -> bool:
+        """Test performance optimization system."""
+        try:
+            self.optimizer = SystemOptimizer()
+            
+            # Test profile management
+            profiles = self.optimizer.profiles
+            self.log_test(
+                "Optimization Profiles",
+                True,
+                f"Loaded {len(profiles)} profiles: {', '.join(profiles.keys())}"
+            )
+            
+            # Test profile application
+            if "balanced" in profiles:
+                success = self.optimizer.apply_profile("balanced")
+                self.log_test("Profile Application", success, "Applied balanced profile")
+            
+            # Test performance analytics
+            analytics = self.optimizer.get_performance_analytics(1)
+            if "system" in analytics:
+                self.log_test("Performance Analytics", True, "Successfully generated analytics")
+            else:
+                self.log_test("Performance Analytics", False, "Failed to generate analytics")
+            
+            return True
+            
+        except Exception as e:
+            self.log_test("Performance Optimizer", False, f"Exception: {e}")
+            return False
+    
+    def test_web_interface(self) -> bool:
+        """Test web interface components."""
+        try:
+            # Test API endpoints
+            with web_app.test_client() as client:
+                # Test status endpoint
+                response = client.get('/api/status')
+                if response.status_code == 200:
+                    self.log_test("Web API - Status", True, "Status endpoint working")
+                else:
+                    self.log_test("Web API - Status", False, f"Status endpoint failed: {response.status_code}")
+                
+                # Test GPU list endpoint
+                response = client.get('/api/gpus')
+                if response.status_code == 200:
+                    self.log_test("Web API - GPUs", True, "GPU list endpoint working")
+                else:
+                    self.log_test("Web API - GPUs", False, f"GPU list endpoint failed: {response.status_code}")
+                
+                # Test fan profiles endpoint
+                response = client.get('/api/fan/profiles')
+                if response.status_code == 200:
+                    self.log_test("Web API - Fan Profiles", True, "Fan profiles endpoint working")
+                else:
+                    self.log_test("Web API - Fan Profiles", False, f"Fan profiles endpoint failed: {response.status_code}")
+            
+            return True
+            
+        except Exception as e:
+            self.log_test("Web Interface", False, f"Exception: {e}")
+            return False
+    
+    def test_data_persistence(self) -> bool:
+        """Test data persistence and database functionality."""
+        try:
+            if not self.gpu_manager:
+                self.gpu_manager = GPUManager()
+                self.gpu_manager.initialize()
+            
+            # Test data collection and storage
+            status = self.gpu_manager.get_status()
+            if status:
+                # Test historical data retrieval
+                for gpu_name in self.gpu_manager.get_gpu_list():
+                    history = self.gpu_manager.get_historical_data(gpu_name, 1)
+                    self.log_test(
+                        f"Data Persistence - {gpu_name}",
+                        True,
+                        f"Retrieved {len(history)} historical records"
+                    )
+            
+            return True
+            
+        except Exception as e:
+            self.log_test("Data Persistence", False, f"Exception: {e}")
+            return False
+    
+    def test_system_integration(self) -> bool:
+        """Test system integration and component interaction."""
+        try:
+            # Test that all components can work together
+            components_working = []
+            
+            if self.gpu_manager and self.gpu_manager.get_gpu_list():
+                components_working.append("GPU Manager")
+            
+            if self.fan_controller and self.fan_controller.get_profiles():
+                components_working.append("Fan Controller")
+            
+            if self.alert_manager:
+                components_working.append("Alert Manager")
+            
+            if self.optimizer and self.optimizer.profiles:
+                components_working.append("Performance Optimizer")
+            
+            success = len(components_working) >= 3  # At least 3 components should work
+            self.log_test(
+                "System Integration",
+                success,
+                f"Working components: {', '.join(components_working)}"
+            )
+            
+            return success
+            
+        except Exception as e:
+            self.log_test("System Integration", False, f"Exception: {e}")
+            return False
+    
+    def test_configuration_files(self) -> bool:
+        """Test configuration file loading and validation."""
+        try:
+            config_files = [
+                "config/fan_profiles.json",
+                "config/monitoring.json",
+                "config/alerts.json",
+                "config/optimization.json"
+            ]
+            
+            valid_configs = 0
+            for config_file in config_files:
+                if Path(config_file).exists():
+                    try:
+                        with open(config_file, 'r') as f:
+                            json.load(f)
+                        valid_configs += 1
+                    except json.JSONDecodeError:
+                        self.log_test(f"Config Validation - {config_file}", False, "Invalid JSON")
+                else:
+                    self.log_test(f"Config File - {config_file}", False, "File not found")
+            
+            success = valid_configs == len(config_files)
+            self.log_test(
+                "Configuration Files",
+                success,
+                f"Valid configs: {valid_configs}/{len(config_files)}"
+            )
+            
+            return success
+            
+        except Exception as e:
+            self.log_test("Configuration Files", False, f"Exception: {e}")
+            return False
+    
+    def test_permissions(self) -> bool:
+        """Test system permissions for GPU access."""
+        try:
+            # Check if we can access GPU sysfs
+            gpu_paths = []
+            for card_path in Path("/sys/class/drm").glob("card*"):
+                device_path = card_path / "device"
+                if (device_path / "vendor").exists():
+                    with open(device_path / "vendor", 'r') as f:
+                        if f.read().strip() == "0x1002":  # AMD
+                            gpu_paths.append(str(device_path))
+            
+            if gpu_paths:
+                # Check hwmon access
+                hwmon_accessible = 0
+                for device_path in gpu_paths:
+                    hwmon_path = Path(device_path) / "hwmon"
+                    if hwmon_path.exists():
+                        hwmons = list(hwmon_path.glob("*"))
+                        if hwmons:
+                            # Test read access
+                            temp_file = hwmons[0] / "temp1_input"
+                            if temp_file.exists():
+                                try:
+                                    with open(temp_file, 'r') as f:
+                                        f.read()
+                                    hwmon_accessible += 1
+                                except PermissionError:
+                                    pass
+                
+                success = hwmon_accessible > 0
+                self.log_test(
+                    "Permissions",
+                    success,
+                    f"Accessible hwmon devices: {hwmon_accessible}/{len(gpu_paths)}"
+                )
+                return success
+            else:
+                self.log_test("Permissions", False, "No AMD GPUs detected")
+                return False
+                
+        except Exception as e:
+            self.log_test("Permissions", False, f"Exception: {e}")
+            return False
+    
+    def run_all_tests(self) -> Dict[str, Any]:
+        """Run all tests and return comprehensive results."""
+        logger.info("=" * 60)
+        logger.info("GPU Monitoring System - Comprehensive Test Suite")
+        logger.info("=" * 60)
+        
+        # Run individual tests
+        tests = [
+            ("Configuration Files", self.test_configuration_files),
+            ("Permissions", self.test_permissions),
+            ("GPU Detection", self.test_gpu_detection),
+            ("Fan Control", self.test_fan_control),
+            ("Alert System", self.test_alert_system),
+            ("Performance Optimizer", self.test_performance_optimizer),
+            ("Web Interface", self.test_web_interface),
+            ("Data Persistence", self.test_data_persistence),
+            ("System Integration", self.test_system_integration),
+        ]
+        
+        for test_name, test_func in tests:
+            logger.info(f"\nRunning: {test_name}")
+            logger.info("-" * 40)
+            try:
+                test_func()
+            except Exception as e:
+                self.log_test(test_name, False, f"Test framework error: {e}")
+        
+        # Calculate results
+        total_tests = len(self.test_results)
+        passed_tests = sum(1 for result in self.test_results if result['success'])
+        failed_tests = total_tests - passed_tests
+        
+        # Generate summary
+        summary = {
+            "total_tests": total_tests,
+            "passed": passed_tests,
+            "failed": failed_tests,
+            "success_rate": (passed_tests / total_tests * 100) if total_tests > 0 else 0,
+            "test_results": self.test_results,
+            "timestamp": time.time(),
+            "system_info": self.get_system_info()
+        }
+        
+        # Save results
+        self.save_test_results(summary)
+        
+        # Print summary
+        logger.info("\n" + "=" * 60)
+        logger.info("TEST SUMMARY")
+        logger.info("=" * 60)
+        logger.info(f"Total Tests: {total_tests}")
+        logger.info(f"Passed: {passed_tests}")
+        logger.info(f"Failed: {failed_tests}")
+        logger.info(f"Success Rate: {summary['success_rate']:.1f}%")
+        
+        if failed_tests > 0:
+            logger.info("\nFailed Tests:")
+            for result in self.test_results:
+                if not result['success']:
+                    logger.info(f"  - {result['test']}: {result['message']}")
+        
+        logger.info("=" * 60)
+        
+        return summary
+    
+    def get_system_info(self) -> Dict[str, Any]:
+        """Get system information for test context."""
+        try:
+            import psutil
+            
+            return {
+                "platform": sys.platform,
+                "python_version": sys.version,
+                "cpu_count": psutil.cpu_count(),
+                "memory_total": psutil.virtual_memory().total // (1024**3),  # GB
+                "disk_total": psutil.disk_usage('/').total // (1024**3),  # GB
+                "uptime": time.time() - psutil.boot_time()
+            }
+        except:
+            return {
+                "platform": sys.platform,
+                "python_version": sys.version
+            }
+    
+    def save_test_results(self, summary: Dict[str, Any]):
+        """Save test results to file."""
+        try:
+            # Save detailed results
+            with open('test_results_detailed.json', 'w') as f:
+                json.dump(summary, f, indent=2, default=str)
+            
+            # Save summary
+            with open('test_summary.txt', 'w') as f:
+                f.write(f"GPU Monitoring System Test Results\n")
+                f.write(f"====================================\n")
+                f.write(f"Timestamp: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
+                f.write(f"Total Tests: {summary['total_tests']}\n")
+                f.write(f"Passed: {summary['passed']}\n")
+                f.write(f"Failed: {summary['failed']}\n")
+                f.write(f"Success Rate: {summary['success_rate']:.1f}%\n\n")
+                
+                f.write("Detailed Results:\n")
+                f.write("================\n")
+                for result in summary['test_results']:
+                    status = "PASS" if result['success'] else "FAIL"
+                    f.write(f"{status}: {result['test']} - {result['message']}\n")
+            
+            logger.info("Test results saved to test_results_detailed.json and test_summary.txt")
+            
+        except Exception as e:
+            logger.error(f"Failed to save test results: {e}")
+
+
+def main():
+    """Main test execution."""
+    print("GPU Monitoring System - Comprehensive Test Suite")
+    print("=" * 60)
+    print("This test suite will validate all components of the")
+    print("GPU monitoring and fan control system.")
+    print()
+    
+    # Check if running as root for full functionality
+    if os.geteuid() != 0:
+        print("WARNING: Not running as root. Some tests may fail due to permission issues.")
+        print("For complete testing, run: sudo python3 test_system.py")
+        print()
+    
+    # Run tests
+    tester = SystemTester()
+    results = tester.run_all_tests()
+    
+    # Exit with appropriate code
+    if results['failed'] > 0:
+        print(f"\nSome tests failed. Please review the results and fix any issues.")
+        sys.exit(1)
+    else:
+        print(f"\nAll tests passed! The system is ready for use.")
+        sys.exit(0)
+
+
+if __name__ == "__main__":
+    main()