""" Integration tests for configuration save/rollback flows. """ import unittest import tempfile import shutil import json from pathlib import Path from src.config_manager_atomic import AtomicConfigManager, SaveResultStatus from src.config_manager import ConfigManager class TestConfigFlowsIntegration(unittest.TestCase): """Integration tests for configuration flows.""" def setUp(self): """Set up test fixtures.""" self.temp_dir = Path(tempfile.mkdtemp()) self.config_path = self.temp_dir / "config.json" self.secrets_path = self.temp_dir / "secrets.json" self.backup_dir = self.temp_dir / "backups" # Create initial config initial_config = { "plugin1": {"enabled": True, "display_duration": 30}, "plugin2": {"enabled": False, "display_duration": 15} } with open(self.config_path, 'w') as f: json.dump(initial_config, f) # Initialize atomic config manager self.atomic_manager = AtomicConfigManager( config_path=str(self.config_path), secrets_path=str(self.secrets_path), backup_dir=str(self.backup_dir), max_backups=5 ) # Initialize regular config manager self.config_manager = ConfigManager() # Override paths for testing self.config_manager.config_path = self.config_path self.config_manager.secrets_path = self.secrets_path def tearDown(self): """Clean up test fixtures.""" shutil.rmtree(self.temp_dir) def test_save_and_rollback_flow(self): """Test saving config and rolling back.""" # Load initial config initial_config = self.config_manager.load_config() self.assertIn("plugin1", initial_config) # Make changes new_config = initial_config.copy() new_config["plugin1"]["display_duration"] = 60 new_config["plugin3"] = {"enabled": True, "display_duration": 20} # Save with atomic manager result = self.atomic_manager.save_config_atomic(new_config, create_backup=True) self.assertEqual(result.status, SaveResultStatus.SUCCESS) self.assertIsNotNone(result.backup_path) # Verify config was saved saved_config = self.config_manager.load_config() self.assertEqual(saved_config["plugin1"]["display_duration"], 60) self.assertIn("plugin3", saved_config) # Rollback - extract version from backup path or use most recent # The backup_path is a full path, but rollback_config expects a version string # So we'll use None to get the most recent backup rollback_success = self.atomic_manager.rollback_config(backup_version=None) self.assertTrue(rollback_success) # Verify config was rolled back rolled_back_config = self.config_manager.load_config() self.assertEqual(rolled_back_config["plugin1"]["display_duration"], 30) self.assertNotIn("plugin3", rolled_back_config) def test_backup_rotation(self): """Test that backup rotation works correctly.""" max_backups = 3 # Create multiple backups for i in range(5): config = {"test": f"value_{i}"} result = self.atomic_manager.save_config_atomic(config, create_backup=True) self.assertEqual(result.status, SaveResultStatus.SUCCESS) # List backups backups = self.atomic_manager.list_backups() # Verify only max_backups are kept self.assertLessEqual(len(backups), max_backups) def test_validation_failure_triggers_rollback(self): """Test that validation failure triggers automatic rollback.""" # Create invalid config (this would fail validation in real scenario) # For this test, we'll simulate by making save fail after write initial_config = self.config_manager.load_config() # Try to save (in real scenario, validation would fail) # Here we'll just verify the atomic save mechanism works new_config = initial_config.copy() new_config["plugin1"]["display_duration"] = 60 result = self.atomic_manager.save_config_atomic(new_config, create_backup=True) # If validation fails, the atomic save should rollback automatically # (This would be handled by the validation step in the atomic save process) self.assertEqual(result.status, SaveResultStatus.SUCCESS) def test_multiple_config_changes(self): """Test multiple sequential config changes.""" config = self.config_manager.load_config() # Make first change config["plugin1"]["display_duration"] = 45 result1 = self.atomic_manager.save_config_atomic(config, create_backup=True) self.assertEqual(result1.status, SaveResultStatus.SUCCESS) # Make second change config = self.config_manager.load_config() config["plugin2"]["display_duration"] = 20 result2 = self.atomic_manager.save_config_atomic(config, create_backup=True) self.assertEqual(result2.status, SaveResultStatus.SUCCESS) # Verify both changes persisted final_config = self.config_manager.load_config() self.assertEqual(final_config["plugin1"]["display_duration"], 45) self.assertEqual(final_config["plugin2"]["display_duration"], 20) # Rollback to first change - get the backup version from the backup path # Extract version from backup path (format: config.json.backup.YYYYMMDD_HHMMSS) import os backup_filename = os.path.basename(result1.backup_path) # Extract timestamp part if '.backup.' in backup_filename: version = backup_filename.split('.backup.')[-1] rollback_success = self.atomic_manager.rollback_config(backup_version=version) else: # Fallback: use most recent backup rollback_success = self.atomic_manager.rollback_config(backup_version=None) self.assertTrue(rollback_success) # Verify rollback rolled_back_config = self.config_manager.load_config() self.assertEqual(rolled_back_config["plugin1"]["display_duration"], 45) self.assertEqual(rolled_back_config["plugin2"]["display_duration"], 15) # Original value if __name__ == '__main__': unittest.main()