DJ-Goanna-Coding/oppo-node

1"""2Comprehensive tests for self_healing_worker.py3 4Tests cover:5- SelfHealingWorker initialization6- Script health checking7- Python script validation8- Bash script validation9- Auto-repair functionality10- Backup creation11- Health reporting12"""13import pytest14import json15import ast16from pathlib import Path17from unittest.mock import Mock, patch, MagicMock18import sys19 20# Add parent directory to path21sys.path.insert(0, str(Path(__file__).parent.parent))22 23from workers.self_healing_worker import SelfHealingWorker, ScriptHealth24 25 26class TestScriptHealth:27    """Test ScriptHealth class"""28 29    def test_script_health_init(self, temp_dir):30        """Test ScriptHealth initialization"""31        test_path = temp_dir / "test.py"32 33        health = ScriptHealth(test_path)34 35        assert health.path == test_path36        assert health.syntax_valid == False37        assert health.imports_valid == False38        assert health.executable == False39        assert health.last_run_success is None40        assert isinstance(health.errors, list)41        assert isinstance(health.warnings, list)42        assert health.last_check is not None43 44 45class TestSelfHealingWorkerInit:46    """Test SelfHealingWorker initialization"""47 48    def test_init_default_values(self):49        """Test that SelfHealingWorker initializes correctly"""50        worker = SelfHealingWorker()51 52        assert worker.base_path is not None53        assert worker.scripts_path is not None54        assert worker.services_path is not None55        assert worker.data_path is not None56        assert isinstance(worker.stats, dict)57 58    def test_init_creates_directories(self, temp_dir, monkeypatch):59        """Test that initialization creates necessary directories"""60        monkeypatch.chdir(temp_dir)61 62        worker = SelfHealingWorker()63 64        assert worker.monitoring_path.exists()65 66    def test_init_stats_structure(self):67        """Test that stats dict has correct structure"""68        worker = SelfHealingWorker()69 70        assert "total_scripts" in worker.stats71        assert "healthy_scripts" in worker.stats72        assert "repaired_scripts" in worker.stats73        assert "failed_repairs" in worker.stats74        assert "scan_time" in worker.stats75 76 77class TestSelfHealingWorkerPythonScriptCheck:78    """Test Python script health checking"""79 80    def test_check_python_script_valid(self, temp_dir):81        """Test checking a valid Python script"""82        worker = SelfHealingWorker()83 84        script_file = temp_dir / "valid.py"85        script_file.write_text("""#!/usr/bin/env python386import os87from pathlib import Path88 89def hello():90    return "Hello"91""")92 93        health = worker.check_python_script(script_file)94 95        assert health.syntax_valid == True96        assert health.imports_valid == True97 98    def test_check_python_script_syntax_error(self, temp_dir):99        """Test checking Python script with syntax error"""100        worker = SelfHealingWorker()101 102        script_file = temp_dir / "invalid.py"103        script_file.write_text("""104def broken(:105    pass106""")107 108        health = worker.check_python_script(script_file)109 110        assert health.syntax_valid == False111        assert len(health.errors) > 0112 113    def test_check_python_script_missing_imports(self, temp_dir):114        """Test checking script with potentially missing imports"""115        worker = SelfHealingWorker()116 117        script_file = temp_dir / "test.py"118        script_file.write_text("""119def test():120    pass121""")122 123        health = worker.check_python_script(script_file)124 125        # Should still be valid even without imports126        assert health.syntax_valid == True127 128    def test_check_python_script_not_executable(self, temp_dir):129        """Test checking non-executable script"""130        worker = SelfHealingWorker()131 132        script_file = temp_dir / "test.py"133        script_file.write_text("print('hello')")134 135        health = worker.check_python_script(script_file)136 137        # Should have warning about not being executable138        assert health.executable == False139 140    def test_check_python_script_nonexistent(self, temp_dir):141        """Test checking non-existent script"""142        worker = SelfHealingWorker()143 144        script_file = temp_dir / "nonexistent.py"145 146        health = worker.check_python_script(script_file)147 148        assert len(health.errors) > 0149 150 151class TestSelfHealingWorkerBashScriptCheck:152    """Test Bash script health checking"""153 154    def test_check_bash_script_valid(self, temp_dir):155        """Test checking a valid Bash script"""156        worker = SelfHealingWorker()157 158        script_file = temp_dir / "valid.sh"159        script_file.write_text("""#!/bin/bash160echo "Hello, World!"161""")162 163        health = worker.check_bash_script(script_file)164 165        assert health.syntax_valid == True166 167    def test_check_bash_script_syntax_error(self, temp_dir):168        """Test checking Bash script with syntax error"""169        worker = SelfHealingWorker()170 171        script_file = temp_dir / "invalid.sh"172        script_file.write_text("""#!/bin/bash173if [ true174echo "missing fi"175""")176 177        health = worker.check_bash_script(script_file)178 179        assert health.syntax_valid == False180        assert len(health.errors) > 0181 182 183class TestSelfHealingWorkerImportCheck:184    """Test import validation"""185 186    def test_check_imports_valid(self):187        """Test checking valid imports"""188        worker = SelfHealingWorker()189 190        content = """191import os192import sys193from pathlib import Path194"""195 196        result = worker.check_imports(content)197 198        assert result == True199 200    def test_check_imports_invalid_syntax(self):201        """Test checking imports with invalid syntax"""202        worker = SelfHealingWorker()203 204        content = """205import os206def broken(:207"""208 209        result = worker.check_imports(content)210 211        assert result == False212 213 214class TestSelfHealingWorkerAutoRepair:215    """Test auto-repair functionality"""216 217    def test_auto_repair_adds_python_shebang(self, temp_dir):218        """Test that auto-repair adds missing Python shebang"""219        worker = SelfHealingWorker()220        worker.data_path = temp_dir / "data"221        worker.data_path.mkdir()222 223        script_file = temp_dir / "test.py"224        script_file.write_text("""225print("Hello")226""")227 228        health = ScriptHealth(script_file)229        health.syntax_valid = True230 231        result = worker.auto_repair_script(script_file, health)232 233        content = script_file.read_text()234        assert content.startswith("#!/usr/bin/env python3")235 236    def test_auto_repair_adds_bash_shebang(self, temp_dir):237        """Test that auto-repair adds missing Bash shebang"""238        worker = SelfHealingWorker()239        worker.data_path = temp_dir / "data"240        worker.data_path.mkdir()241 242        script_file = temp_dir / "test.sh"243        script_file.write_text("""244echo "Hello"245""")246 247        health = ScriptHealth(script_file)248 249        result = worker.auto_repair_script(script_file, health)250 251        content = script_file.read_text()252        assert content.startswith("#!/bin/bash")253 254    def test_auto_repair_makes_executable(self, temp_dir):255        """Test that auto-repair makes script executable"""256        worker = SelfHealingWorker()257        worker.data_path = temp_dir / "data"258        worker.data_path.mkdir()259 260        script_file = temp_dir / "test.py"261        script_file.write_text("#!/usr/bin/env python3\nprint('hello')")262 263        import os264        os.chmod(script_file, 0o644)  # Not executable265 266        health = ScriptHealth(script_file)267        health.executable = False268 269        worker.auto_repair_script(script_file, health)270 271        # Check if file is now executable272        assert os.access(script_file, os.X_OK)273 274    def test_auto_repair_adds_pathlib_import(self, temp_dir):275        """Test that auto-repair adds missing pathlib import"""276        worker = SelfHealingWorker()277        worker.data_path = temp_dir / "data"278        worker.data_path.mkdir()279 280        script_file = temp_dir / "test.py"281        script_file.write_text("""#!/usr/bin/env python3282import os283 284def test():285    p = Path("test")286    return p287""")288 289        health = ScriptHealth(script_file)290 291        worker.auto_repair_script(script_file, health)292 293        content = script_file.read_text()294        assert "from pathlib import Path" in content295 296    def test_auto_repair_creates_backup(self, temp_dir):297        """Test that auto-repair creates a backup"""298        worker = SelfHealingWorker()299        worker.data_path = temp_dir / "data"300        worker.data_path.mkdir()301 302        script_file = temp_dir / "test.py"303        script_file.write_text("print('hello')")304 305        health = ScriptHealth(script_file)306 307        worker.auto_repair_script(script_file, health)308 309        backup_dir = temp_dir / "data" / "backups" / "scripts"310        assert backup_dir.exists()311        backups = list(backup_dir.glob("test.py.*.bak"))312        assert len(backups) > 0313 314 315class TestSelfHealingWorkerBackup:316    """Test backup functionality"""317 318    def test_backup_script(self, temp_dir):319        """Test creating script backup"""320        worker = SelfHealingWorker()321        worker.data_path = temp_dir / "data"322 323        script_file = temp_dir / "test.py"324        script_file.write_text("print('hello')")325 326        backup_path = worker.backup_script(script_file)327 328        assert backup_path is not None329        assert backup_path.exists()330        assert "test.py" in backup_path.name331        assert ".bak" in backup_path.name332 333    def test_backup_script_preserves_content(self, temp_dir):334        """Test that backup preserves original content"""335        worker = SelfHealingWorker()336        worker.data_path = temp_dir / "data"337 338        script_file = temp_dir / "test.py"339        original_content = "print('original')"340        script_file.write_text(original_content)341 342        backup_path = worker.backup_script(script_file)343 344        assert backup_path.read_text() == original_content345 346 347class TestSelfHealingWorkerScanning:348    """Test script scanning functionality"""349 350    def test_scan_all_scripts(self, temp_dir, monkeypatch):351        """Test scanning all scripts"""352        monkeypatch.chdir(temp_dir)353        worker = SelfHealingWorker()354 355        # Create test scripts356        scripts_dir = temp_dir / "scripts"357        scripts_dir.mkdir()358        (scripts_dir / "test1.py").write_text("#!/usr/bin/env python3\nprint('test1')")359        (scripts_dir / "test2.py").write_text("#!/usr/bin/env python3\nprint('test2')")360 361        worker.scripts_path = scripts_dir362 363        health_map = worker.scan_all_scripts()364 365        assert len(health_map) >= 2366        assert worker.stats["total_scripts"] >= 2367 368    def test_scan_all_scripts_empty_directory(self, temp_dir, monkeypatch):369        """Test scanning empty directory"""370        monkeypatch.chdir(temp_dir)371        worker = SelfHealingWorker()372 373        scripts_dir = temp_dir / "scripts"374        scripts_dir.mkdir()375 376        worker.scripts_path = scripts_dir377 378        health_map = worker.scan_all_scripts()379 380        assert len(health_map) == 0381 382 383class TestSelfHealingWorkerReporting:384    """Test health reporting functionality"""385 386    def test_generate_health_report(self, temp_dir):387        """Test generating health report"""388        worker = SelfHealingWorker()389 390        script_path = temp_dir / "test.py"391        health = ScriptHealth(script_path)392        health.syntax_valid = True393        health.imports_valid = True394 395        worker.stats["total_scripts"] = 1396        worker.stats["healthy_scripts"] = 1397 398        health_map = {"test.py": health}399 400        report = worker.generate_health_report(health_map)401 402        assert "timestamp" in report403        assert "summary" in report404        assert "scripts" in report405        assert report["summary"]["total_scripts"] == 1406        assert report["summary"]["healthy_scripts"] == 1407 408    def test_save_health_report(self, temp_dir):409        """Test saving health report to file"""410        worker = SelfHealingWorker()411        worker.monitoring_path = temp_dir412        worker.health_report_path = temp_dir / "health_report.json"413 414        report = {415            "timestamp": "2026-04-14",416            "summary": {"total_scripts": 5},417            "scripts": {}418        }419 420        worker.save_health_report(report)421 422        assert worker.health_report_path.exists()423 424        with open(worker.health_report_path, 'r') as f:425            loaded_report = json.load(f)426 427        assert loaded_report["summary"]["total_scripts"] == 5428 429 430class TestSelfHealingWorkerIntegration:431    """Integration tests for SelfHealingWorker"""432 433    def test_run_full_heal(self, temp_dir, monkeypatch):434        """Test full healing workflow"""435        monkeypatch.chdir(temp_dir)436        worker = SelfHealingWorker()437 438        # Create test scripts439        scripts_dir = temp_dir / "scripts"440        scripts_dir.mkdir()441        (scripts_dir / "valid.py").write_text("#!/usr/bin/env python3\nprint('valid')")442        (scripts_dir / "needs_shebang.py").write_text("print('no shebang')")443 444        worker.scripts_path = scripts_dir445        worker.monitoring_path = temp_dir446        worker.health_report_path = temp_dir / "health.json"447        worker.data_path = temp_dir / "data"448 449        report = worker.run_full_heal(auto_repair=True)450 451        assert "summary" in report452        assert report["summary"]["total_scripts"] >= 1453        assert worker.health_report_path.exists()454 455    def test_run_full_heal_no_repair(self, temp_dir, monkeypatch):456        """Test full scan without auto-repair"""457        monkeypatch.chdir(temp_dir)458        worker = SelfHealingWorker()459 460        scripts_dir = temp_dir / "scripts"461        scripts_dir.mkdir()462        (scripts_dir / "test.py").write_text("print('test')")463 464        worker.scripts_path = scripts_dir465        worker.monitoring_path = temp_dir466        worker.health_report_path = temp_dir / "health.json"467 468        report = worker.run_full_heal(auto_repair=False)469 470        assert worker.stats["repaired_scripts"] == 0471