299 lines
11 KiB
Python
299 lines
11 KiB
Python
"""
|
|
Unit tests for the configuration manager.
|
|
"""
|
|
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from config_manager import ConfigurationManager
|
|
from database import SQLiteDatabase
|
|
from models import QueueConfig, QueueState
|
|
|
|
|
|
class TestConfigurationManager(unittest.TestCase):
|
|
"""Test cases for ConfigurationManager class."""
|
|
|
|
def setUp(self):
|
|
"""Set up test database and configuration manager."""
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.db_path = Path(self.temp_dir) / "test_config.db"
|
|
self.db = SQLiteDatabase(self.db_path)
|
|
self.assertTrue(self.db.initialize())
|
|
self.config_manager = ConfigurationManager(self.db)
|
|
|
|
def tearDown(self):
|
|
"""Clean up test database."""
|
|
self.db.close()
|
|
if self.db_path.exists():
|
|
self.db_path.unlink()
|
|
|
|
def test_get_default_config(self):
|
|
"""Test getting default configuration."""
|
|
config = self.config_manager.get_config()
|
|
|
|
self.assertEqual(config.queue_state, QueueState.RUNNING)
|
|
self.assertEqual(config.max_concurrent_workflows, 1)
|
|
self.assertFalse(config.auto_archive_completed)
|
|
self.assertEqual(config.auto_archive_days, 30)
|
|
|
|
def test_update_config(self):
|
|
"""Test updating configuration."""
|
|
new_config = QueueConfig(
|
|
queue_state=QueueState.PAUSED,
|
|
max_concurrent_workflows=3,
|
|
auto_archive_completed=True,
|
|
auto_archive_days=7
|
|
)
|
|
|
|
self.assertTrue(self.config_manager.update_config(new_config))
|
|
|
|
# Verify configuration was updated
|
|
retrieved_config = self.config_manager.get_config()
|
|
self.assertEqual(retrieved_config.queue_state, QueueState.PAUSED)
|
|
self.assertEqual(retrieved_config.max_concurrent_workflows, 3)
|
|
self.assertTrue(retrieved_config.auto_archive_completed)
|
|
self.assertEqual(retrieved_config.auto_archive_days, 7)
|
|
|
|
def test_queue_state_operations(self):
|
|
"""Test queue state management."""
|
|
# Initially running
|
|
self.assertEqual(self.config_manager.get_queue_state(), QueueState.RUNNING)
|
|
self.assertFalse(self.config_manager.is_queue_paused())
|
|
|
|
# Pause queue
|
|
self.assertTrue(self.config_manager.pause_queue())
|
|
self.assertEqual(self.config_manager.get_queue_state(), QueueState.PAUSED)
|
|
self.assertTrue(self.config_manager.is_queue_paused())
|
|
|
|
# Resume queue
|
|
self.assertTrue(self.config_manager.resume_queue())
|
|
self.assertEqual(self.config_manager.get_queue_state(), QueueState.RUNNING)
|
|
self.assertFalse(self.config_manager.is_queue_paused())
|
|
|
|
def test_max_concurrent_workflows(self):
|
|
"""Test max concurrent workflows setting."""
|
|
# Default value
|
|
self.assertEqual(self.config_manager.get_max_concurrent_workflows(), 1)
|
|
|
|
# Set new value
|
|
self.assertTrue(self.config_manager.set_max_concurrent_workflows(5))
|
|
self.assertEqual(self.config_manager.get_max_concurrent_workflows(), 5)
|
|
|
|
# Test invalid value
|
|
self.assertFalse(self.config_manager.set_max_concurrent_workflows(0))
|
|
self.assertFalse(self.config_manager.set_max_concurrent_workflows(-1))
|
|
# Value should remain unchanged
|
|
self.assertEqual(self.config_manager.get_max_concurrent_workflows(), 5)
|
|
|
|
def test_auto_archive_settings(self):
|
|
"""Test auto-archive settings."""
|
|
# Default values
|
|
enabled, days = self.config_manager.get_auto_archive_settings()
|
|
self.assertFalse(enabled)
|
|
self.assertEqual(days, 30)
|
|
|
|
# Set new values
|
|
self.assertTrue(self.config_manager.set_auto_archive_settings(True, 14))
|
|
enabled, days = self.config_manager.get_auto_archive_settings()
|
|
self.assertTrue(enabled)
|
|
self.assertEqual(days, 14)
|
|
|
|
# Test invalid days value
|
|
self.assertFalse(self.config_manager.set_auto_archive_settings(True, 0))
|
|
self.assertFalse(self.config_manager.set_auto_archive_settings(True, -1))
|
|
# Values should remain unchanged
|
|
enabled, days = self.config_manager.get_auto_archive_settings()
|
|
self.assertTrue(enabled)
|
|
self.assertEqual(days, 14)
|
|
|
|
def test_user_preferences(self):
|
|
"""Test user preferences management."""
|
|
# Initially empty
|
|
prefs = self.config_manager.get_user_preferences()
|
|
self.assertEqual(prefs, {})
|
|
|
|
# Set preferences
|
|
test_prefs = {
|
|
"theme": "dark",
|
|
"auto_refresh": True,
|
|
"refresh_interval": 5000,
|
|
"columns": ["name", "status", "created_at"]
|
|
}
|
|
self.assertTrue(self.config_manager.set_user_preferences(test_prefs))
|
|
|
|
# Verify preferences
|
|
retrieved_prefs = self.config_manager.get_user_preferences()
|
|
self.assertEqual(retrieved_prefs, test_prefs)
|
|
|
|
def test_individual_user_preference(self):
|
|
"""Test individual user preference operations."""
|
|
# Test getting non-existent preference with default
|
|
value = self.config_manager.get_user_preference("theme", "light")
|
|
self.assertEqual(value, "light")
|
|
|
|
# Set individual preference
|
|
self.assertTrue(self.config_manager.set_user_preference("theme", "dark"))
|
|
value = self.config_manager.get_user_preference("theme")
|
|
self.assertEqual(value, "dark")
|
|
|
|
# Set another preference
|
|
self.assertTrue(self.config_manager.set_user_preference("auto_refresh", True))
|
|
|
|
# Verify both preferences exist
|
|
prefs = self.config_manager.get_user_preferences()
|
|
self.assertEqual(prefs["theme"], "dark")
|
|
self.assertTrue(prefs["auto_refresh"])
|
|
|
|
def test_reset_to_defaults(self):
|
|
"""Test resetting configuration to defaults."""
|
|
# Modify configuration
|
|
self.config_manager.pause_queue()
|
|
self.config_manager.set_max_concurrent_workflows(10)
|
|
self.config_manager.set_auto_archive_settings(True, 7)
|
|
|
|
# Reset to defaults
|
|
self.assertTrue(self.config_manager.reset_to_defaults())
|
|
|
|
# Verify defaults
|
|
config = self.config_manager.get_config()
|
|
self.assertEqual(config.queue_state, QueueState.RUNNING)
|
|
self.assertEqual(config.max_concurrent_workflows, 1)
|
|
self.assertFalse(config.auto_archive_completed)
|
|
self.assertEqual(config.auto_archive_days, 30)
|
|
|
|
def test_export_config(self):
|
|
"""Test configuration export."""
|
|
# Set some configuration
|
|
self.config_manager.pause_queue()
|
|
self.config_manager.set_max_concurrent_workflows(3)
|
|
self.config_manager.set_user_preference("theme", "dark")
|
|
|
|
# Export configuration
|
|
exported = self.config_manager.export_config()
|
|
|
|
# Verify export structure
|
|
self.assertIn("queue_config", exported)
|
|
self.assertIn("user_preferences", exported)
|
|
self.assertIn("version", exported)
|
|
|
|
# Verify queue config
|
|
queue_config = exported["queue_config"]
|
|
self.assertEqual(queue_config["queue_state"], "paused")
|
|
self.assertEqual(queue_config["max_concurrent_workflows"], 3)
|
|
|
|
# Verify user preferences
|
|
user_prefs = exported["user_preferences"]
|
|
self.assertEqual(user_prefs["theme"], "dark")
|
|
|
|
def test_import_config(self):
|
|
"""Test configuration import."""
|
|
# Prepare import data
|
|
import_data = {
|
|
"queue_config": {
|
|
"queue_state": "paused",
|
|
"max_concurrent_workflows": 5,
|
|
"auto_archive_completed": True,
|
|
"auto_archive_days": 14
|
|
},
|
|
"user_preferences": {
|
|
"theme": "dark",
|
|
"auto_refresh": False
|
|
},
|
|
"version": "1.0"
|
|
}
|
|
|
|
# Import configuration
|
|
self.assertTrue(self.config_manager.import_config(import_data))
|
|
|
|
# Verify imported configuration
|
|
config = self.config_manager.get_config()
|
|
self.assertEqual(config.queue_state, QueueState.PAUSED)
|
|
self.assertEqual(config.max_concurrent_workflows, 5)
|
|
self.assertTrue(config.auto_archive_completed)
|
|
self.assertEqual(config.auto_archive_days, 14)
|
|
|
|
# Verify imported preferences
|
|
prefs = self.config_manager.get_user_preferences()
|
|
self.assertEqual(prefs["theme"], "dark")
|
|
self.assertFalse(prefs["auto_refresh"])
|
|
|
|
def test_config_persistence(self):
|
|
"""Test that configuration persists across manager instances."""
|
|
# Set configuration
|
|
self.config_manager.pause_queue()
|
|
self.config_manager.set_max_concurrent_workflows(7)
|
|
self.config_manager.set_user_preference("theme", "dark")
|
|
|
|
# Create new configuration manager with same database
|
|
new_config_manager = ConfigurationManager(self.db)
|
|
|
|
# Verify configuration persisted
|
|
config = new_config_manager.get_config()
|
|
self.assertEqual(config.queue_state, QueueState.PAUSED)
|
|
self.assertEqual(config.max_concurrent_workflows, 7)
|
|
|
|
prefs = new_config_manager.get_user_preferences()
|
|
self.assertEqual(prefs["theme"], "dark")
|
|
|
|
def test_cache_invalidation(self):
|
|
"""Test configuration cache invalidation."""
|
|
# Load initial config
|
|
config1 = self.config_manager.get_config()
|
|
self.assertEqual(config1.queue_state, QueueState.RUNNING)
|
|
|
|
# Modify database directly (bypassing cache)
|
|
self.db.set_config("queue_state", "paused")
|
|
|
|
# Config should still be cached
|
|
config2 = self.config_manager.get_config()
|
|
self.assertEqual(config2.queue_state, QueueState.RUNNING)
|
|
|
|
# Invalidate cache
|
|
self.config_manager.invalidate_cache()
|
|
|
|
# Now should get updated config
|
|
config3 = self.config_manager.get_config()
|
|
self.assertEqual(config3.queue_state, QueueState.PAUSED)
|
|
|
|
def test_invalid_json_handling(self):
|
|
"""Test handling of invalid JSON in preferences."""
|
|
# Set invalid JSON directly in database
|
|
self.db.set_config("user_preferences", "invalid json {")
|
|
|
|
# Should return empty dict and not crash
|
|
prefs = self.config_manager.get_user_preferences()
|
|
self.assertEqual(prefs, {})
|
|
|
|
def test_partial_import(self):
|
|
"""Test importing partial configuration."""
|
|
# Import only queue config
|
|
import_data = {
|
|
"queue_config": {
|
|
"queue_state": "paused",
|
|
"max_concurrent_workflows": 2
|
|
}
|
|
}
|
|
|
|
self.assertTrue(self.config_manager.import_config(import_data))
|
|
|
|
config = self.config_manager.get_config()
|
|
self.assertEqual(config.queue_state, QueueState.PAUSED)
|
|
self.assertEqual(config.max_concurrent_workflows, 2)
|
|
|
|
# Import only user preferences
|
|
import_data = {
|
|
"user_preferences": {
|
|
"theme": "light"
|
|
}
|
|
}
|
|
|
|
self.assertTrue(self.config_manager.import_config(import_data))
|
|
|
|
prefs = self.config_manager.get_user_preferences()
|
|
self.assertEqual(prefs["theme"], "light")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |