import os import json import logging import typing as t import jsonschema from jsonschema import validate import psycopg2 import redis import qdrant_client import requests # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') class ConfigValidator: """ A class to validate configuration files and environment variables for the Genesis system. """ def __init__(self, config_files: t.List[str], config_schemas: t.Dict[str, dict]): """ Initializes the ConfigValidator with a list of configuration files and corresponding schemas. Args: config_files: A list of paths to the configuration files. config_schemas: A dictionary mapping config file names to their JSON schemas. """ self.config_files = config_files self.config_schemas = config_schemas self.validation_report: t.Dict[str, t.List[str]] = {} # Store validation messages def load_config(self, config_file: str) -> t.Dict: """ Loads a configuration file from the given path. Args: config_file: The path to the configuration file. Returns: A dictionary containing the configuration data. Raises: FileNotFoundError: If the configuration file does not exist. json.JSONDecodeError: If the configuration file is not valid JSON. """ try: with open(config_file, 'r') as f: config_data = json.load(f) return config_data except FileNotFoundError as e: logging.error(f"Config file not found: {config_file} - {e}") raise except json.JSONDecodeError as e: logging.error(f"Invalid JSON in config file: {config_file} - {e}") raise def validate_config(self, config_data: t.Dict, schema: dict, config_file: str) -> None: """ Validates the given configuration data against the provided JSON schema. Args: config_data: The configuration data to validate. schema: The JSON schema to validate against. config_file: The name of the config file being validated. """ try: validate(instance=config_data, schema=schema) logging.info(f"Config file '{config_file}' validated successfully against schema.") if config_file in self.validation_report: self.validation_report[config_file].append("Schema validation successful.") else: self.validation_report[config_file] = ["Schema validation successful."] except jsonschema.exceptions.ValidationError as e: error_message = f"Config file '{config_file}' failed schema validation: {e}" logging.error(error_message) if config_file in self.validation_report: self.validation_report[config_file].append(error_message) else: self.validation_report[config_file] = [error_message] def validate_env_vars(self, config_data: t.Dict, config_file: str) -> None: """ Validates that the required environment variables are set. Args: config_data: The configuration data (dictionary) to check for environment variables. config_file: The name of the config file being validated. """ env_var_errors = [] for key, value in config_data.items(): if isinstance(value, str) and value.startswith("${") and value.endswith("}"): env_var_name = value[2:-1] if env_var_name not in os.environ: error_message = f"Environment variable '{env_var_name}' not set." env_var_errors.append(error_message) logging.error(error_message) if env_var_errors: if config_file in self.validation_report: self.validation_report[config_file].extend(env_var_errors) else: self.validation_report[config_file] = env_var_errors else: if config_file in self.validation_report: self.validation_report[config_file].append("All environment variables are set.") else: self.validation_report[config_file] = ["All environment variables are set."] def validate_connections(self, config_data: t.Dict, config_file: str) -> None: """ Validates database, Redis, Qdrant, and AIVA Ollama connections based on config data. Args: config_data: The configuration data (dictionary) containing connection details. config_file: The name of the config file being validated. """ connection_errors = [] # PostgreSQL Connection Validation if 'postgres' in config_data: postgres_config = config_data['postgres'] try: conn = psycopg2.connect( host=postgres_config['host'], port=postgres_config['port'], database=postgres_config['database'], user=postgres_config['user'], password=postgres_config['password'] ) conn.close() logging.info("PostgreSQL connection successful.") if config_file in self.validation_report: self.validation_report[config_file].append("PostgreSQL connection successful.") else: self.validation_report[config_file] = ["PostgreSQL connection successful."] except Exception as e: error_message = f"PostgreSQL connection failed: {e}" connection_errors.append(error_message) logging.error(error_message) # Redis Connection Validation if 'redis' in config_data: redis_config = config_data['redis'] try: r = redis.Redis(host=redis_config['host'], port=redis_config['port']) r.ping() logging.info("Redis connection successful.") if config_file in self.validation_report: self.validation_report[config_file].append("Redis connection successful.") else: self.validation_report[config_file] = ["Redis connection successful."] except Exception as e: error_message = f"Redis connection failed: {e}" connection_errors.append(error_message) logging.error(error_message) # Qdrant Connection Validation if 'qdrant' in config_data: qdrant_config = config_data['qdrant'] try: client = qdrant_client.QdrantClient(host=qdrant_config['host'], port=qdrant_config['port']) client.get_telemetry_data() # Simple call to check connection logging.info("Qdrant connection successful.") if config_file in self.validation_report: self.validation_report[config_file].append("Qdrant connection successful.") else: self.validation_report[config_file] = ["Qdrant connection successful."] except Exception as e: error_message = f"Qdrant connection failed: {e}" connection_errors.append(error_message) logging.error(error_message) # AIVA Ollama Connection Validation if 'aiva_ollama' in config_data: aiva_ollama_config = config_data['aiva_ollama'] try: response = requests.get(f"http://{aiva_ollama_config['host']}:{aiva_ollama_config['port']}/") # Replace with a meaningful Ollama API endpoint response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx) logging.info("AIVA Ollama connection successful.") if config_file in self.validation_report: self.validation_report[config_file].append("AIVA Ollama connection successful.") else: self.validation_report[config_file] = ["AIVA Ollama connection successful."] except requests.exceptions.RequestException as e: error_message = f"AIVA Ollama connection failed: {e}" connection_errors.append(error_message) logging.error(error_message) if connection_errors: if config_file in self.validation_report: self.validation_report[config_file].extend(connection_errors) else: self.validation_report[config_file] = connection_errors def run_validation(self) -> t.Dict[str, t.List[str]]: """ Runs the validation process for all configuration files. Returns: A dictionary containing the validation report. The keys are the config file names, and the values are lists of validation messages (errors or successes). """ for config_file in self.config_files: try: config_data = self.load_config(config_file) schema = self.config_schemas.get(config_file) if schema: self.validate_config(config_data, schema, config_file) else: logging.warning(f"No schema found for config file: {config_file}. Skipping schema validation.") if config_file not in self.validation_report: self.validation_report[config_file] = ["No schema found, skipping schema validation."] self.validate_env_vars(config_data, config_file) self.validate_connections(config_data, config_file) except Exception as e: error_message = f"An error occurred while validating config file '{config_file}': {e}" logging.exception(error_message) if config_file in self.validation_report: self.validation_report[config_file].append(error_message) else: self.validation_report[config_file] = [error_message] return self.validation_report def generate_report(self, report_file: str = "validation_report.txt") -> None: """ Generates a human-readable report of the validation results. Args: report_file: The file to write the report to. Defaults to "validation_report.txt". """ with open(report_file, "w") as f: for config_file, messages in self.validation_report.items(): f.write(f"Validation report for: {config_file}\n") for message in messages: f.write(f"- {message}\n") f.write("\n") logging.info(f"Validation report generated: {report_file}") if __name__ == '__main__': # Example Usage config_files = [ '/mnt/e/genesis-system/config/database.json', '/mnt/e/genesis-system/config/redis.json', '/mnt/e/genesis-system/config/qdrant.json', '/mnt/e/genesis-system/config/aiva.json' ] config_schemas = { '/mnt/e/genesis-system/config/database.json': { "type": "object", "properties": { "host": {"type": "string"}, "port": {"type": "integer"}, "database": {"type": "string"}, "user": {"type": "string"}, "password": {"type": "string"} }, "required": ["host", "port", "database", "user", "password"] }, '/mnt/e/genesis-system/config/redis.json': { "type": "object", "properties": { "host": {"type": "string"}, "port": {"type": "integer"} }, "required": ["host", "port"] }, '/mnt/e/genesis-system/config/qdrant.json': { "type": "object", "properties": { "host": {"type": "string"}, "port": {"type": "integer"} }, "required": ["host", "port"] }, '/mnt/e/genesis-system/config/aiva.json': { "type": "object", "properties": { "host": {"type": "string"}, "port": {"type": "integer"} }, "required": ["host", "port"] } } # Create dummy config files os.makedirs('/mnt/e/genesis-system/config', exist_ok=True) with open('/mnt/e/genesis-system/config/database.json', 'w') as f: json.dump({"host": "postgresql-genesis-u50607.vm.elestio.app", "port": 25432, "database": "genesis", "user": "genesis", "password": "password"}, f) with open('/mnt/e/genesis-system/config/redis.json', 'w') as f: json.dump({"host": "redis-genesis-u50607.vm.elestio.app", "port": 26379}, f) with open('/mnt/e/genesis-system/config/qdrant.json', 'w') as f: json.dump({"host": "qdrant-b3knu-u50607.vm.elestio.app", "port": 6333}, f) with open('/mnt/e/genesis-system/config/aiva.json', 'w') as f: json.dump({"host": "localhost", "port": 23405}, f) validator = ConfigValidator(config_files, config_schemas) report = validator.run_validation() print(report) validator.generate_report()