""" Property-based tests for Configuration Manager Tests universal correctness properties for the unified configuration system. Uses real functionality without mocks - tests actual file system operations, environment variable handling, and configuration validation. """ import pytest import os import tempfile import shutil from pathlib import Path from hypothesis import given, strategies as st, assume, settings from hypothesis.stateful import RuleBasedStateMachine, rule, initialize, invariant from dataclasses import asdict from typing import Dict, Any from core.config import ( ConfigurationManager, SystemConfig, ValidationError, get_configuration_manager, get_config ) class TestConfigurationProperties: """Property tests for Configuration Manager using real functionality""" def setup_method(self): """Setup real temporary directories and clean environment""" self.temp_dir = Path(tempfile.mkdtemp()) self.original_env = {} # Store original environment variables we'll modify env_vars_to_backup = [ "ENVIRONMENT", "API_PORT", "DASHBOARD_PORT", "WORKER_THREADS", "HEALTH_CHECK_INTERVAL", "SECRET_KEY", "ENCRYPTION_PASSWORD", "BASE_PATH", "DATA_PATH", "LOGS_PATH" ] for var in env_vars_to_backup: self.original_env[var] = os.environ.get(var) def teardown_method(self): """Restore original environment and cleanup real directories""" # Restore original environment for key, value in self.original_env.items(): if value is None: os.environ.pop(key, None) else: os.environ[key] = value # Cleanup real temporary directory if self.temp_dir.exists(): shutil.rmtree(self.temp_dir) @given( environment=st.sampled_from(["development", "staging", "production"]), api_port=st.integers(min_value=1024, max_value=65535), dashboard_port=st.integers(min_value=1024, max_value=65535), worker_threads=st.integers(min_value=1, max_value=32), health_check_interval=st.integers(min_value=1, max_value=300) ) @settings(max_examples=50) def test_property_1_configuration_consistency( self, environment, api_port, dashboard_port, worker_threads, health_check_interval ): """ **Feature: rpa-system-unification, Property 1: Configuration Consistency** For any system startup, all components should use identical configuration values for shared settings. **Validates: Requirements 1.1, 1.2** """ assume(api_port != dashboard_port) # Ports must be different # Set up real environment variables env_vars = { "ENVIRONMENT": environment, "API_PORT": str(api_port), "DASHBOARD_PORT": str(dashboard_port), "WORKER_THREADS": str(worker_threads), "HEALTH_CHECK_INTERVAL": str(health_check_interval), "BASE_PATH": str(self.temp_dir) } # For production environment, add required security keys if environment == "production": env_vars["SECRET_KEY"] = "test_production_secret_key_12345" env_vars["ENCRYPTION_PASSWORD"] = "test_production_encryption_password_12345" try: # Set test environment variables for key, value in env_vars.items(): os.environ[key] = value # Create multiple configuration manager instances (real objects) config_manager_1 = ConfigurationManager() config_manager_2 = ConfigurationManager() # Load configuration from both managers (real file system operations) config_1 = config_manager_1.load_config() config_2 = config_manager_2.load_config() # Property: All configuration managers should return identical values assert config_1.environment == config_2.environment == environment assert config_1.api_port == config_2.api_port == api_port assert config_1.dashboard_port == config_2.dashboard_port == dashboard_port assert config_1.worker_threads == config_2.worker_threads == worker_threads assert config_1.health_check_interval == config_2.health_check_interval == health_check_interval # Property: Configuration should be consistent across multiple loads config_1_reload = config_manager_1.reload_config() assert asdict(config_1) == asdict(config_1_reload) # Verify real directories were created assert config_1.data_path.exists() assert config_1.logs_path.exists() assert config_1.sessions_path.exists() # Test real file system permissions test_file = config_1.data_path / "test_write.txt" test_file.write_text("test") assert test_file.read_text() == "test" test_file.unlink() finally: # Environment cleanup handled by teardown_method pass @given( secret_key=st.text(min_size=10, max_size=100), encryption_password=st.text(min_size=10, max_size=100), invalid_port=st.integers(max_value=1023) | st.integers(min_value=65536), invalid_threads=st.integers(max_value=0), invalid_interval=st.integers(max_value=0) ) @settings(max_examples=30) def test_property_10_configuration_validation_completeness( self, secret_key, encryption_password, invalid_port, invalid_threads, invalid_interval ): """ **Feature: rpa-system-unification, Property 10: Configuration Validation Completeness** For any invalid configuration, the system should detect and report all validation errors before attempting to start services. **Validates: Requirements 1.4, 1.5** """ # Use real temporary directory for testing test_base_path = self.temp_dir / "validation_test" test_base_path.mkdir(exist_ok=True) config_manager = ConfigurationManager() # Test production environment validation with real SystemConfig prod_config = SystemConfig( base_path=test_base_path, environment="production", secret_key="dev_secret_key_not_for_production", # Invalid for production encryption_password="dev_default_key_not_for_production", # Invalid for production debug=True # Should warn in production ) # Real validation using actual validation logic validation_errors = config_manager.validate_config(prod_config) # Property: Should detect all production validation errors error_fields = {error.field for error in validation_errors if error.severity == "error"} assert "secret_key" in error_fields assert "encryption_password" in error_fields warning_fields = {error.field for error in validation_errors if error.severity == "warning"} assert "debug" in warning_fields # Test invalid port configuration with real SystemConfig invalid_port_config = SystemConfig( base_path=test_base_path, api_port=invalid_port, dashboard_port=5001 ) port_errors = config_manager.validate_config(invalid_port_config) port_error_fields = {error.field for error in port_errors if error.severity == "error"} assert "api_port" in port_error_fields # Test same ports configuration same_ports_config = SystemConfig( base_path=test_base_path, api_port=8000, dashboard_port=8000 # Same as API port ) same_port_errors = config_manager.validate_config(same_ports_config) same_port_error_fields = {error.field for error in same_port_errors if error.severity == "error"} assert "ports" in same_port_error_fields # Test invalid worker threads invalid_threads_config = SystemConfig( base_path=test_base_path, worker_threads=invalid_threads ) thread_errors = config_manager.validate_config(invalid_threads_config) thread_error_fields = {error.field for error in thread_errors if error.severity == "error"} assert "worker_threads" in thread_error_fields # Test invalid health check interval invalid_interval_config = SystemConfig( base_path=test_base_path, health_check_interval=invalid_interval ) interval_errors = config_manager.validate_config(invalid_interval_config) interval_error_fields = {error.field for error in interval_errors if error.severity == "error"} assert "health_check_interval" in interval_error_fields @given( base_path=st.text(min_size=1, max_size=50).filter( lambda x: not any(c in x for c in ['/', '\\', ':', '*', '?', '"', '<', '>', '|']) ) ) @settings(max_examples=20) def test_configuration_path_resolution(self, base_path): """Test that all paths are correctly resolved relative to base_path using real file system""" # Create real base directory base_path_obj = self.temp_dir / base_path base_path_obj.mkdir(exist_ok=True) # Create real SystemConfig with actual path resolution config = SystemConfig(base_path=base_path_obj) # Property: All paths should be absolute and accessible assert config.data_path.is_absolute() assert config.logs_path.is_absolute() assert config.sessions_path.is_absolute() assert config.workflows_path.is_absolute() # Test real directory creation config.ensure_directories() # Verify directories actually exist on file system assert config.data_path.exists() assert config.logs_path.exists() assert config.sessions_path.exists() assert config.workflows_path.exists() # Test real file operations in created directories test_file = config.data_path / "test.txt" test_file.write_text("test content") assert test_file.read_text() == "test content" def test_configuration_watcher_real_functionality(self): """Test that configuration watchers receive real updates using actual callback mechanism""" # Use real temporary directory test_base_path = self.temp_dir / "watcher_test" test_base_path.mkdir(exist_ok=True) # Set real environment variable os.environ["BASE_PATH"] = str(test_base_path) try: config_manager = ConfigurationManager() # Track real watcher calls watcher_calls = [] def test_watcher(config: SystemConfig): watcher_calls.append({ 'environment': config.environment, 'base_path': str(config.base_path), 'api_port': config.api_port }) # Register real watcher config_manager.watch_config_changes(test_watcher) # Load initial configuration (triggers real file system operations) initial_config = config_manager.load_config() # Property: Watcher should be called with initial config assert len(watcher_calls) == 1 assert watcher_calls[0]['environment'] == initial_config.environment assert Path(watcher_calls[0]['base_path']) == initial_config.base_path # Apply new configuration with real validation and directory creation new_config = SystemConfig( base_path=test_base_path, environment="staging", api_port=8001, dashboard_port=5002 ) config_manager.apply_config(new_config) # Property: Watcher should be called with new config assert len(watcher_calls) == 2 assert watcher_calls[1]['environment'] == "staging" assert watcher_calls[1]['api_port'] == 8001 # Verify real directories were created for new config assert new_config.data_path.exists() assert new_config.logs_path.exists() finally: # Cleanup handled by teardown_method pass def test_real_environment_variable_loading(self): """Test loading configuration from real environment variables""" # Set real environment variables test_env = { "BASE_PATH": str(self.temp_dir), "ENVIRONMENT": "staging", "API_PORT": "8080", "DASHBOARD_PORT": "5050", "WORKER_THREADS": "8", "CLIP_MODEL": "ViT-L-14", "FAISS_DIMENSIONS": "768" } try: for key, value in test_env.items(): os.environ[key] = value # Load configuration using real environment variable parsing config_manager = ConfigurationManager() config = config_manager.load_config() # Verify real environment variables were loaded correctly assert config.environment == "staging" assert config.api_port == 8080 assert config.dashboard_port == 5050 assert config.worker_threads == 8 assert config.clip_model == "ViT-L-14" assert config.faiss_dimensions == 768 assert config.base_path == self.temp_dir # Verify real directories were created assert config.data_path.exists() assert config.logs_path.exists() finally: # Cleanup environment variables for key in test_env: os.environ.pop(key, None) class ConfigurationStateMachine(RuleBasedStateMachine): """Stateful property testing for Configuration Manager using real functionality""" def __init__(self): super().__init__() self.temp_dir = Path(tempfile.mkdtemp()) self.config_manager = ConfigurationManager() self.applied_configs = [] self.watcher_calls = [] self.original_env = {} # Backup original environment for var in ["BASE_PATH", "ENVIRONMENT", "API_PORT", "DASHBOARD_PORT"]: self.original_env[var] = os.environ.get(var) # Set base path to our temporary directory os.environ["BASE_PATH"] = str(self.temp_dir) def teardown(self): """Cleanup real resources""" # Restore environment for key, value in self.original_env.items(): if value is None: os.environ.pop(key, None) else: os.environ[key] = value # Cleanup real directory if self.temp_dir.exists(): shutil.rmtree(self.temp_dir) @initialize() def setup(self): """Initialize with real configuration manager and watcher""" def track_watcher(config: SystemConfig): self.watcher_calls.append({ 'environment': config.environment, 'timestamp': len(self.watcher_calls) }) self.config_manager.watch_config_changes(track_watcher) @rule( environment=st.sampled_from(["development", "staging"]), api_port=st.integers(min_value=8000, max_value=8010), dashboard_port=st.integers(min_value=5000, max_value=5010) ) def apply_valid_config(self, environment, api_port, dashboard_port): """Apply a valid configuration using real SystemConfig and file operations""" assume(api_port != dashboard_port) # Create real configuration config = SystemConfig( base_path=self.temp_dir, environment=environment, api_port=api_port, dashboard_port=dashboard_port ) # Apply using real configuration manager self.config_manager.apply_config(config) self.applied_configs.append(config) # Verify real directories were created assert config.data_path.exists() assert config.logs_path.exists() @rule() def reload_config(self): """Reload configuration from real environment""" reloaded = self.config_manager.reload_config() # Property: Reloaded config should be valid using real validation errors = self.config_manager.validate_config(reloaded) error_count = sum(1 for error in errors if error.severity == "error") assert error_count == 0 # Verify real directories exist assert reloaded.data_path.exists() @invariant() def config_always_valid(self): """Configuration should always be valid using real validation logic""" current_config = self.config_manager.get_config() errors = self.config_manager.validate_config(current_config) error_count = sum(1 for error in errors if error.severity == "error") assert error_count == 0 @invariant() def watchers_called_for_changes(self): """Watchers should be called for each real configuration change""" # Number of watcher calls should match number of config changes + initial load expected_calls = len(self.applied_configs) + 1 # +1 for initial load assert len(self.watcher_calls) >= expected_calls @invariant() def directories_always_exist(self): """Real directories should always exist for current configuration""" current_config = self.config_manager.get_config() assert current_config.data_path.exists() assert current_config.logs_path.exists() # Test the stateful machine TestConfigurationStateMachine = ConfigurationStateMachine.TestCase if __name__ == "__main__": # Run a quick property test with real functionality test_instance = TestConfigurationProperties() test_instance.setup_method() try: print("Running Property 1: Configuration Consistency with real file system...") test_instance.test_property_1_configuration_consistency( environment="development", api_port=8000, dashboard_port=5001, worker_threads=4, health_check_interval=30 ) print("✅ Property 1 passed") print("Running Property 10: Configuration Validation with real validation...") test_instance.test_property_10_configuration_validation_completeness( secret_key="test_key_12345", encryption_password="test_password_12345", invalid_port=80, invalid_threads=-1, invalid_interval=0 ) print("✅ Property 10 passed") print("Running real environment variable loading test...") test_instance.test_real_environment_variable_loading() print("✅ Environment variable loading test passed") print("✅ All configuration properties validated with real functionality") finally: test_instance.teardown_method()