Spaces:
Sleeping
Sleeping
| """ | |
| Comprehensive tests for worker_watchdog.py | |
| Tests cover: | |
| - WorkerWatchdog initialization | |
| - File hash calculation | |
| - Change detection | |
| - Self-healing trigger | |
| - Workflow health checking | |
| - State management | |
| """ | |
| import pytest | |
| import json | |
| import hashlib | |
| 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.worker_watchdog import WorkerWatchdog | |
| class TestWorkerWatchdogInit: | |
| """Test WorkerWatchdog initialization""" | |
| def test_init_default_values(self): | |
| """Test that WorkerWatchdog initializes with correct defaults""" | |
| watchdog = WorkerWatchdog() | |
| assert watchdog.check_interval == 300 | |
| assert watchdog.running == False | |
| assert isinstance(watchdog.file_hashes, dict) | |
| assert isinstance(watchdog.template_hashes, dict) | |
| assert isinstance(watchdog.stats, dict) | |
| def test_init_custom_interval(self): | |
| """Test initialization with custom check interval""" | |
| watchdog = WorkerWatchdog(check_interval=600) | |
| assert watchdog.check_interval == 600 | |
| def test_init_creates_monitoring_directory(self, temp_dir, monkeypatch): | |
| """Test that initialization creates monitoring directory""" | |
| monkeypatch.chdir(temp_dir) | |
| watchdog = WorkerWatchdog() | |
| assert watchdog.monitoring_path.exists() | |
| def test_init_stats_structure(self): | |
| """Test that stats dict has correct structure""" | |
| watchdog = WorkerWatchdog() | |
| assert "total_checks" in watchdog.stats | |
| assert "issues_detected" in watchdog.stats | |
| assert "auto_repairs_triggered" in watchdog.stats | |
| assert "successful_repairs" in watchdog.stats | |
| assert "start_time" in watchdog.stats | |
| assert "last_check" in watchdog.stats | |
| class TestWorkerWatchdogFileHashing: | |
| """Test file hashing functionality""" | |
| def test_calculate_file_hash(self, temp_dir): | |
| """Test that file hash is calculated correctly""" | |
| watchdog = WorkerWatchdog() | |
| test_file = temp_dir / "test.txt" | |
| test_content = b"test content" | |
| test_file.write_bytes(test_content) | |
| hash_result = watchdog.calculate_file_hash(test_file) | |
| # Verify it's a valid SHA256 hash | |
| assert len(hash_result) == 64 | |
| assert all(c in '0123456789abcdef' for c in hash_result) | |
| # Verify hash is correct | |
| expected_hash = hashlib.sha256(test_content).hexdigest() | |
| assert hash_result == expected_hash | |
| def test_calculate_file_hash_nonexistent(self, temp_dir): | |
| """Test hash calculation for non-existent file""" | |
| watchdog = WorkerWatchdog() | |
| nonexistent_file = temp_dir / "nonexistent.txt" | |
| hash_result = watchdog.calculate_file_hash(nonexistent_file) | |
| assert hash_result == "" | |
| def test_calculate_file_hash_empty_file(self, temp_dir): | |
| """Test hash calculation for empty file""" | |
| watchdog = WorkerWatchdog() | |
| empty_file = temp_dir / "empty.txt" | |
| empty_file.write_bytes(b"") | |
| hash_result = watchdog.calculate_file_hash(empty_file) | |
| expected_hash = hashlib.sha256(b"").hexdigest() | |
| assert hash_result == expected_hash | |
| def test_scan_file_hashes(self, temp_dir, monkeypatch): | |
| """Test scanning directory for file hashes""" | |
| monkeypatch.chdir(temp_dir) | |
| watchdog = WorkerWatchdog() | |
| # Create test Python files | |
| scripts_dir = temp_dir / "scripts" | |
| scripts_dir.mkdir() | |
| (scripts_dir / "test1.py").write_text("content1") | |
| (scripts_dir / "test2.py").write_text("content2") | |
| (scripts_dir / "test.txt").write_text("not python") | |
| hashes = watchdog.scan_file_hashes(scripts_dir, "*.py") | |
| assert len(hashes) == 2 | |
| assert any("test1.py" in key for key in hashes.keys()) | |
| assert any("test2.py" in key for key in hashes.keys()) | |
| def test_scan_file_hashes_empty_directory(self, temp_dir): | |
| """Test scanning empty directory""" | |
| watchdog = WorkerWatchdog() | |
| empty_dir = temp_dir / "empty" | |
| empty_dir.mkdir() | |
| hashes = watchdog.scan_file_hashes(empty_dir, "*.py") | |
| assert len(hashes) == 0 | |
| def test_scan_file_hashes_nonexistent_directory(self, temp_dir): | |
| """Test scanning non-existent directory""" | |
| watchdog = WorkerWatchdog() | |
| nonexistent_dir = temp_dir / "nonexistent" | |
| hashes = watchdog.scan_file_hashes(nonexistent_dir, "*.py") | |
| assert len(hashes) == 0 | |
| class TestWorkerWatchdogChangeDetection: | |
| """Test change detection functionality""" | |
| def test_detect_changes_new_file(self, temp_dir, monkeypatch): | |
| """Test detection of new files""" | |
| monkeypatch.chdir(temp_dir) | |
| watchdog = WorkerWatchdog() | |
| # Initial scan | |
| scripts_dir = temp_dir / "scripts" | |
| scripts_dir.mkdir() | |
| watchdog.scripts_path = scripts_dir | |
| watchdog.file_hashes = watchdog.scan_file_hashes(scripts_dir, "*.py") | |
| # Add new file | |
| (scripts_dir / "new_file.py").write_text("new content") | |
| changes = watchdog.detect_changes() | |
| assert len(changes["new_files"]) == 1 | |
| assert any("new_file.py" in f for f in changes["new_files"]) | |
| def test_detect_changes_modified_file(self, temp_dir, monkeypatch): | |
| """Test detection of modified files""" | |
| monkeypatch.chdir(temp_dir) | |
| watchdog = WorkerWatchdog() | |
| scripts_dir = temp_dir / "scripts" | |
| scripts_dir.mkdir() | |
| test_file = scripts_dir / "test.py" | |
| test_file.write_text("original content") | |
| watchdog.scripts_path = scripts_dir | |
| watchdog.file_hashes = watchdog.scan_file_hashes(scripts_dir, "*.py") | |
| # Modify file | |
| test_file.write_text("modified content") | |
| changes = watchdog.detect_changes() | |
| assert len(changes["modified_files"]) == 1 | |
| assert any("test.py" in f for f in changes["modified_files"]) | |
| def test_detect_changes_deleted_file(self, temp_dir, monkeypatch): | |
| """Test detection of deleted files""" | |
| monkeypatch.chdir(temp_dir) | |
| watchdog = WorkerWatchdog() | |
| scripts_dir = temp_dir / "scripts" | |
| scripts_dir.mkdir() | |
| test_file = scripts_dir / "test.py" | |
| test_file.write_text("content") | |
| watchdog.scripts_path = scripts_dir | |
| watchdog.file_hashes = watchdog.scan_file_hashes(scripts_dir, "*.py") | |
| # Delete file | |
| test_file.unlink() | |
| changes = watchdog.detect_changes() | |
| assert len(changes["deleted_files"]) == 1 | |
| assert any("test.py" in f for f in changes["deleted_files"]) | |
| def test_detect_changes_no_changes(self, temp_dir, monkeypatch): | |
| """Test when no changes detected""" | |
| monkeypatch.chdir(temp_dir) | |
| watchdog = WorkerWatchdog() | |
| scripts_dir = temp_dir / "scripts" | |
| scripts_dir.mkdir() | |
| (scripts_dir / "test.py").write_text("content") | |
| watchdog.scripts_path = scripts_dir | |
| watchdog.file_hashes = watchdog.scan_file_hashes(scripts_dir, "*.py") | |
| changes = watchdog.detect_changes() | |
| assert len(changes["new_files"]) == 0 | |
| assert len(changes["modified_files"]) == 0 | |
| assert len(changes["deleted_files"]) == 0 | |
| class TestWorkerWatchdogSelfHealing: | |
| """Test self-healing trigger functionality""" | |
| def test_trigger_self_healing_success(self, temp_dir, monkeypatch): | |
| """Test successful self-healing trigger""" | |
| monkeypatch.chdir(temp_dir) | |
| watchdog = WorkerWatchdog() | |
| # Create mock healing script | |
| workers_dir = temp_dir / "workers" | |
| workers_dir.mkdir() | |
| healing_script = workers_dir / "self_healing_worker.py" | |
| healing_script.write_text("#!/usr/bin/env python3\nprint('healing')") | |
| watchdog.base_path = temp_dir | |
| with patch('subprocess.run') as mock_run: | |
| mock_run.return_value = Mock(returncode=0, stdout="success", stderr="") | |
| result = watchdog.trigger_self_healing() | |
| assert result == True | |
| assert watchdog.stats["successful_repairs"] == 1 | |
| def test_trigger_self_healing_failure(self, temp_dir, monkeypatch): | |
| """Test failed self-healing trigger""" | |
| monkeypatch.chdir(temp_dir) | |
| watchdog = WorkerWatchdog() | |
| workers_dir = temp_dir / "workers" | |
| workers_dir.mkdir() | |
| healing_script = workers_dir / "self_healing_worker.py" | |
| healing_script.write_text("#!/usr/bin/env python3\nprint('healing')") | |
| watchdog.base_path = temp_dir | |
| with patch('subprocess.run') as mock_run: | |
| mock_run.return_value = Mock(returncode=1, stdout="", stderr="error") | |
| result = watchdog.trigger_self_healing() | |
| assert result == False | |
| def test_trigger_self_healing_script_not_found(self, temp_dir, monkeypatch): | |
| """Test self-healing when script not found""" | |
| monkeypatch.chdir(temp_dir) | |
| watchdog = WorkerWatchdog() | |
| watchdog.base_path = temp_dir | |
| result = watchdog.trigger_self_healing() | |
| assert result == False | |
| def test_trigger_self_healing_timeout(self, temp_dir, monkeypatch): | |
| """Test self-healing timeout""" | |
| monkeypatch.chdir(temp_dir) | |
| watchdog = WorkerWatchdog() | |
| workers_dir = temp_dir / "workers" | |
| workers_dir.mkdir() | |
| healing_script = workers_dir / "self_healing_worker.py" | |
| healing_script.write_text("#!/usr/bin/env python3\nprint('healing')") | |
| watchdog.base_path = temp_dir | |
| with patch('subprocess.run') as mock_run: | |
| from subprocess import TimeoutExpired | |
| mock_run.side_effect = TimeoutExpired("cmd", 300) | |
| result = watchdog.trigger_self_healing() | |
| assert result == False | |
| class TestWorkerWatchdogStateManagement: | |
| """Test state save/load functionality""" | |
| def test_save_state(self, temp_dir, monkeypatch): | |
| """Test saving watchdog state""" | |
| monkeypatch.chdir(temp_dir) | |
| watchdog = WorkerWatchdog() | |
| watchdog.monitoring_path = temp_dir | |
| watchdog.watchdog_state_file = temp_dir / "watchdog_state.json" | |
| watchdog.stats["total_checks"] = 5 | |
| watchdog.file_hashes = {"test.py": "hash123"} | |
| watchdog.save_state() | |
| assert watchdog.watchdog_state_file.exists() | |
| with open(watchdog.watchdog_state_file, 'r') as f: | |
| state = json.load(f) | |
| assert state["stats"]["total_checks"] == 5 | |
| assert state["file_count"] == 1 | |
| def test_load_state(self, temp_dir, monkeypatch): | |
| """Test loading watchdog state""" | |
| monkeypatch.chdir(temp_dir) | |
| watchdog = WorkerWatchdog() | |
| watchdog.monitoring_path = temp_dir | |
| watchdog.watchdog_state_file = temp_dir / "watchdog_state.json" | |
| # Create state file | |
| state = { | |
| "stats": {"total_checks": 10}, | |
| "file_count": 5, | |
| "last_update": "2026-04-14" | |
| } | |
| with open(watchdog.watchdog_state_file, 'w') as f: | |
| json.dump(state, f) | |
| watchdog.load_state() | |
| assert watchdog.stats["total_checks"] == 10 | |
| def test_load_state_file_not_exists(self, temp_dir, monkeypatch): | |
| """Test loading state when file doesn't exist""" | |
| monkeypatch.chdir(temp_dir) | |
| watchdog = WorkerWatchdog() | |
| watchdog.monitoring_path = temp_dir | |
| watchdog.watchdog_state_file = temp_dir / "nonexistent.json" | |
| # Should not raise error | |
| watchdog.load_state() | |
| class TestWorkerWatchdogHealthCheck: | |
| """Test health check functionality""" | |
| def test_perform_health_check(self, temp_dir, monkeypatch): | |
| """Test performing a health check""" | |
| monkeypatch.chdir(temp_dir) | |
| watchdog = WorkerWatchdog() | |
| scripts_dir = temp_dir / "scripts" | |
| scripts_dir.mkdir() | |
| watchdog.scripts_path = scripts_dir | |
| watchdog.monitoring_path = temp_dir | |
| with patch.object(watchdog, 'trigger_self_healing') as mock_heal: | |
| watchdog.perform_health_check() | |
| assert watchdog.stats["total_checks"] == 1 | |
| assert watchdog.stats["last_check"] is not None | |
| def test_run_once(self, temp_dir, monkeypatch): | |
| """Test running health check once""" | |
| monkeypatch.chdir(temp_dir) | |
| watchdog = WorkerWatchdog() | |
| scripts_dir = temp_dir / "scripts" | |
| scripts_dir.mkdir() | |
| watchdog.scripts_path = scripts_dir | |
| watchdog.monitoring_path = temp_dir | |
| with patch.object(watchdog, 'trigger_self_healing'): | |
| watchdog.run_once() | |
| assert watchdog.stats["total_checks"] == 1 | |