Spaces:
Sleeping
Sleeping
| """ | |
| Comprehensive tests for self_healing_worker.py | |
| Tests cover: | |
| - SelfHealingWorker initialization | |
| - Script health checking | |
| - Python script validation | |
| - Bash script validation | |
| - Auto-repair functionality | |
| - Backup creation | |
| - Health reporting | |
| """ | |
| import pytest | |
| import json | |
| import ast | |
| from pathlib import Path | |
| from unittest.mock import Mock, patch, MagicMock | |
| import sys | |
| # Add parent directory to path | |
| sys.path.insert(0, str(Path(__file__).parent.parent)) | |
| from workers.self_healing_worker import SelfHealingWorker, ScriptHealth | |
| class TestScriptHealth: | |
| """Test ScriptHealth class""" | |
| def test_script_health_init(self, temp_dir): | |
| """Test ScriptHealth initialization""" | |
| test_path = temp_dir / "test.py" | |
| health = ScriptHealth(test_path) | |
| assert health.path == test_path | |
| assert health.syntax_valid == False | |
| assert health.imports_valid == False | |
| assert health.executable == False | |
| assert health.last_run_success is None | |
| assert isinstance(health.errors, list) | |
| assert isinstance(health.warnings, list) | |
| assert health.last_check is not None | |
| class TestSelfHealingWorkerInit: | |
| """Test SelfHealingWorker initialization""" | |
| def test_init_default_values(self): | |
| """Test that SelfHealingWorker initializes correctly""" | |
| worker = SelfHealingWorker() | |
| assert worker.base_path is not None | |
| assert worker.scripts_path is not None | |
| assert worker.services_path is not None | |
| assert worker.data_path is not None | |
| assert isinstance(worker.stats, dict) | |
| def test_init_creates_directories(self, temp_dir, monkeypatch): | |
| """Test that initialization creates necessary directories""" | |
| monkeypatch.chdir(temp_dir) | |
| worker = SelfHealingWorker() | |
| assert worker.monitoring_path.exists() | |
| def test_init_stats_structure(self): | |
| """Test that stats dict has correct structure""" | |
| worker = SelfHealingWorker() | |
| assert "total_scripts" in worker.stats | |
| assert "healthy_scripts" in worker.stats | |
| assert "repaired_scripts" in worker.stats | |
| assert "failed_repairs" in worker.stats | |
| assert "scan_time" in worker.stats | |
| class TestSelfHealingWorkerPythonScriptCheck: | |
| """Test Python script health checking""" | |
| def test_check_python_script_valid(self, temp_dir): | |
| """Test checking a valid Python script""" | |
| worker = SelfHealingWorker() | |
| script_file = temp_dir / "valid.py" | |
| script_file.write_text("""#!/usr/bin/env python3 | |
| import os | |
| from pathlib import Path | |
| def hello(): | |
| return "Hello" | |
| """) | |
| health = worker.check_python_script(script_file) | |
| assert health.syntax_valid == True | |
| assert health.imports_valid == True | |
| def test_check_python_script_syntax_error(self, temp_dir): | |
| """Test checking Python script with syntax error""" | |
| worker = SelfHealingWorker() | |
| script_file = temp_dir / "invalid.py" | |
| script_file.write_text(""" | |
| def broken(: | |
| pass | |
| """) | |
| health = worker.check_python_script(script_file) | |
| assert health.syntax_valid == False | |
| assert len(health.errors) > 0 | |
| def test_check_python_script_missing_imports(self, temp_dir): | |
| """Test checking script with potentially missing imports""" | |
| worker = SelfHealingWorker() | |
| script_file = temp_dir / "test.py" | |
| script_file.write_text(""" | |
| def test(): | |
| pass | |
| """) | |
| health = worker.check_python_script(script_file) | |
| # Should still be valid even without imports | |
| assert health.syntax_valid == True | |
| def test_check_python_script_not_executable(self, temp_dir): | |
| """Test checking non-executable script""" | |
| worker = SelfHealingWorker() | |
| script_file = temp_dir / "test.py" | |
| script_file.write_text("print('hello')") | |
| health = worker.check_python_script(script_file) | |
| # Should have warning about not being executable | |
| assert health.executable == False | |
| def test_check_python_script_nonexistent(self, temp_dir): | |
| """Test checking non-existent script""" | |
| worker = SelfHealingWorker() | |
| script_file = temp_dir / "nonexistent.py" | |
| health = worker.check_python_script(script_file) | |
| assert len(health.errors) > 0 | |
| class TestSelfHealingWorkerBashScriptCheck: | |
| """Test Bash script health checking""" | |
| def test_check_bash_script_valid(self, temp_dir): | |
| """Test checking a valid Bash script""" | |
| worker = SelfHealingWorker() | |
| script_file = temp_dir / "valid.sh" | |
| script_file.write_text("""#!/bin/bash | |
| echo "Hello, World!" | |
| """) | |
| health = worker.check_bash_script(script_file) | |
| assert health.syntax_valid == True | |
| def test_check_bash_script_syntax_error(self, temp_dir): | |
| """Test checking Bash script with syntax error""" | |
| worker = SelfHealingWorker() | |
| script_file = temp_dir / "invalid.sh" | |
| script_file.write_text("""#!/bin/bash | |
| if [ true | |
| echo "missing fi" | |
| """) | |
| health = worker.check_bash_script(script_file) | |
| assert health.syntax_valid == False | |
| assert len(health.errors) > 0 | |
| class TestSelfHealingWorkerImportCheck: | |
| """Test import validation""" | |
| def test_check_imports_valid(self): | |
| """Test checking valid imports""" | |
| worker = SelfHealingWorker() | |
| content = """ | |
| import os | |
| import sys | |
| from pathlib import Path | |
| """ | |
| result = worker.check_imports(content) | |
| assert result == True | |
| def test_check_imports_invalid_syntax(self): | |
| """Test checking imports with invalid syntax""" | |
| worker = SelfHealingWorker() | |
| content = """ | |
| import os | |
| def broken(: | |
| """ | |
| result = worker.check_imports(content) | |
| assert result == False | |
| class TestSelfHealingWorkerAutoRepair: | |
| """Test auto-repair functionality""" | |
| def test_auto_repair_adds_python_shebang(self, temp_dir): | |
| """Test that auto-repair adds missing Python shebang""" | |
| worker = SelfHealingWorker() | |
| worker.data_path = temp_dir / "data" | |
| worker.data_path.mkdir() | |
| script_file = temp_dir / "test.py" | |
| script_file.write_text(""" | |
| print("Hello") | |
| """) | |
| health = ScriptHealth(script_file) | |
| health.syntax_valid = True | |
| result = worker.auto_repair_script(script_file, health) | |
| content = script_file.read_text() | |
| assert content.startswith("#!/usr/bin/env python3") | |
| def test_auto_repair_adds_bash_shebang(self, temp_dir): | |
| """Test that auto-repair adds missing Bash shebang""" | |
| worker = SelfHealingWorker() | |
| worker.data_path = temp_dir / "data" | |
| worker.data_path.mkdir() | |
| script_file = temp_dir / "test.sh" | |
| script_file.write_text(""" | |
| echo "Hello" | |
| """) | |
| health = ScriptHealth(script_file) | |
| result = worker.auto_repair_script(script_file, health) | |
| content = script_file.read_text() | |
| assert content.startswith("#!/bin/bash") | |
| def test_auto_repair_makes_executable(self, temp_dir): | |
| """Test that auto-repair makes script executable""" | |
| worker = SelfHealingWorker() | |
| worker.data_path = temp_dir / "data" | |
| worker.data_path.mkdir() | |
| script_file = temp_dir / "test.py" | |
| script_file.write_text("#!/usr/bin/env python3\nprint('hello')") | |
| import os | |
| os.chmod(script_file, 0o644) # Not executable | |
| health = ScriptHealth(script_file) | |
| health.executable = False | |
| worker.auto_repair_script(script_file, health) | |
| # Check if file is now executable | |
| assert os.access(script_file, os.X_OK) | |
| def test_auto_repair_adds_pathlib_import(self, temp_dir): | |
| """Test that auto-repair adds missing pathlib import""" | |
| worker = SelfHealingWorker() | |
| worker.data_path = temp_dir / "data" | |
| worker.data_path.mkdir() | |
| script_file = temp_dir / "test.py" | |
| script_file.write_text("""#!/usr/bin/env python3 | |
| import os | |
| def test(): | |
| p = Path("test") | |
| return p | |
| """) | |
| health = ScriptHealth(script_file) | |
| worker.auto_repair_script(script_file, health) | |
| content = script_file.read_text() | |
| assert "from pathlib import Path" in content | |
| def test_auto_repair_creates_backup(self, temp_dir): | |
| """Test that auto-repair creates a backup""" | |
| worker = SelfHealingWorker() | |
| worker.data_path = temp_dir / "data" | |
| worker.data_path.mkdir() | |
| script_file = temp_dir / "test.py" | |
| script_file.write_text("print('hello')") | |
| health = ScriptHealth(script_file) | |
| worker.auto_repair_script(script_file, health) | |
| backup_dir = temp_dir / "data" / "backups" / "scripts" | |
| assert backup_dir.exists() | |
| backups = list(backup_dir.glob("test.py.*.bak")) | |
| assert len(backups) > 0 | |
| class TestSelfHealingWorkerBackup: | |
| """Test backup functionality""" | |
| def test_backup_script(self, temp_dir): | |
| """Test creating script backup""" | |
| worker = SelfHealingWorker() | |
| worker.data_path = temp_dir / "data" | |
| script_file = temp_dir / "test.py" | |
| script_file.write_text("print('hello')") | |
| backup_path = worker.backup_script(script_file) | |
| assert backup_path is not None | |
| assert backup_path.exists() | |
| assert "test.py" in backup_path.name | |
| assert ".bak" in backup_path.name | |
| def test_backup_script_preserves_content(self, temp_dir): | |
| """Test that backup preserves original content""" | |
| worker = SelfHealingWorker() | |
| worker.data_path = temp_dir / "data" | |
| script_file = temp_dir / "test.py" | |
| original_content = "print('original')" | |
| script_file.write_text(original_content) | |
| backup_path = worker.backup_script(script_file) | |
| assert backup_path.read_text() == original_content | |
| class TestSelfHealingWorkerScanning: | |
| """Test script scanning functionality""" | |
| def test_scan_all_scripts(self, temp_dir, monkeypatch): | |
| """Test scanning all scripts""" | |
| monkeypatch.chdir(temp_dir) | |
| worker = SelfHealingWorker() | |
| # Create test scripts | |
| scripts_dir = temp_dir / "scripts" | |
| scripts_dir.mkdir() | |
| (scripts_dir / "test1.py").write_text("#!/usr/bin/env python3\nprint('test1')") | |
| (scripts_dir / "test2.py").write_text("#!/usr/bin/env python3\nprint('test2')") | |
| worker.scripts_path = scripts_dir | |
| health_map = worker.scan_all_scripts() | |
| assert len(health_map) >= 2 | |
| assert worker.stats["total_scripts"] >= 2 | |
| def test_scan_all_scripts_empty_directory(self, temp_dir, monkeypatch): | |
| """Test scanning empty directory""" | |
| monkeypatch.chdir(temp_dir) | |
| worker = SelfHealingWorker() | |
| scripts_dir = temp_dir / "scripts" | |
| scripts_dir.mkdir() | |
| worker.scripts_path = scripts_dir | |
| health_map = worker.scan_all_scripts() | |
| assert len(health_map) == 0 | |
| class TestSelfHealingWorkerReporting: | |
| """Test health reporting functionality""" | |
| def test_generate_health_report(self, temp_dir): | |
| """Test generating health report""" | |
| worker = SelfHealingWorker() | |
| script_path = temp_dir / "test.py" | |
| health = ScriptHealth(script_path) | |
| health.syntax_valid = True | |
| health.imports_valid = True | |
| worker.stats["total_scripts"] = 1 | |
| worker.stats["healthy_scripts"] = 1 | |
| health_map = {"test.py": health} | |
| report = worker.generate_health_report(health_map) | |
| assert "timestamp" in report | |
| assert "summary" in report | |
| assert "scripts" in report | |
| assert report["summary"]["total_scripts"] == 1 | |
| assert report["summary"]["healthy_scripts"] == 1 | |
| def test_save_health_report(self, temp_dir): | |
| """Test saving health report to file""" | |
| worker = SelfHealingWorker() | |
| worker.monitoring_path = temp_dir | |
| worker.health_report_path = temp_dir / "health_report.json" | |
| report = { | |
| "timestamp": "2026-04-14", | |
| "summary": {"total_scripts": 5}, | |
| "scripts": {} | |
| } | |
| worker.save_health_report(report) | |
| assert worker.health_report_path.exists() | |
| with open(worker.health_report_path, 'r') as f: | |
| loaded_report = json.load(f) | |
| assert loaded_report["summary"]["total_scripts"] == 5 | |
| class TestSelfHealingWorkerIntegration: | |
| """Integration tests for SelfHealingWorker""" | |
| def test_run_full_heal(self, temp_dir, monkeypatch): | |
| """Test full healing workflow""" | |
| monkeypatch.chdir(temp_dir) | |
| worker = SelfHealingWorker() | |
| # Create test scripts | |
| scripts_dir = temp_dir / "scripts" | |
| scripts_dir.mkdir() | |
| (scripts_dir / "valid.py").write_text("#!/usr/bin/env python3\nprint('valid')") | |
| (scripts_dir / "needs_shebang.py").write_text("print('no shebang')") | |
| worker.scripts_path = scripts_dir | |
| worker.monitoring_path = temp_dir | |
| worker.health_report_path = temp_dir / "health.json" | |
| worker.data_path = temp_dir / "data" | |
| report = worker.run_full_heal(auto_repair=True) | |
| assert "summary" in report | |
| assert report["summary"]["total_scripts"] >= 1 | |
| assert worker.health_report_path.exists() | |
| def test_run_full_heal_no_repair(self, temp_dir, monkeypatch): | |
| """Test full scan without auto-repair""" | |
| monkeypatch.chdir(temp_dir) | |
| worker = SelfHealingWorker() | |
| scripts_dir = temp_dir / "scripts" | |
| scripts_dir.mkdir() | |
| (scripts_dir / "test.py").write_text("print('test')") | |
| worker.scripts_path = scripts_dir | |
| worker.monitoring_path = temp_dir | |
| worker.health_report_path = temp_dir / "health.json" | |
| report = worker.run_full_heal(auto_repair=False) | |
| assert worker.stats["repaired_scripts"] == 0 | |