from github.env import get_required_env_var, exit_on_error_and_write_summary from typing import Optional from actions_logging.app_logging import logger import json import os from env_files.utils import get_all_env_files_names from env_files.constants import ( DEFAULT_ENV_FILES_PATH, GLOBAL_FILE, ENV_FILES_SUBFOLDERS, ) class ValidationError(Exception): """Exception raised for validation errors in env files.""" pass def extract_error_messages(errors: list[ValidationError]) -> list[str]: """ Extract clean error messages from ValidationError exceptions with type information :param errors: List of ValidationError exceptions :return: List of error message strings with exception type prefix """ messages = [] logger.debug("parsing thrown validation errors") for exc in errors: exc_type = type(exc).__name__ if hasattr(exc, 'args') and exc.args: # Get the first argument which is typically the error message arg = exc.args[0] if isinstance(arg, list): logger.debug(f"encountered nested list of errors") for nested_err in arg: if isinstance(nested_err, ValidationError): nested_messages = extract_error_messages([nested_err]) messages.extend(nested_messages) else: # Add type information for non-ValidationError errors in the list nested_type = type(nested_err).__name__ if not isinstance(nested_err, str) else "Error" messages.append(f"{nested_type}: {str(nested_err)}") elif isinstance(arg, ValidationError): logger.debug("encountered validation error") nested_messages = extract_error_messages([arg]) messages.extend(nested_messages) elif isinstance(arg, Exception): logger.debug("encountered general exception") arg_type = type(arg).__name__ messages.append(f"{arg_type}: {str(arg)}") else: messages.append(f"{exc_type}: {str(arg)}") else: messages.append(f"{exc_type}: {str(exc)}") return messages def check_if_file_exists(file_path): if not os.path.exists(file_path): raise ValidationError(f"file {file_path} does not exist") logger.info(f"file {file_path} exists") def compare_files(base_file_path, env_file_path): """ Compare the keys in the base file to the keys in an environment file. Ensures that all keys in the environment file exist in the base file. Raises ValidationError if the base file is empty or if the environment file contains keys not present in the base file. Args: base_file_path (str): Path to the base JSON file (typically global.json) env_file_path (str): Path to the environment JSON file to validate Raises: ValidationError: If validation fails due to empty base file or extra keys in env file Exception: If files cannot be loaded or other errors occur """ errors = [] try: logger.debug(f"loading {base_file_path}") with open(base_file_path, 'r') as bf: base_file = json.load(bf) except Exception as e: errors.append(f"couldn't load {base_file_path}: {e}") try: logger.debug(f"loading {env_file_path}") with open(env_file_path, 'r') as ef: env_file = json.load(ef) except Exception as e: errors.append(f"couldn't load {env_file_path}: {e}") if errors: raise ValidationError(errors) try: base_set = set(base_file.keys()) if len(base_set) == 0: # if the base file is empty, then the env file should be empty as well so no .env file is created. # remove the empty files and don't create empty files! logger.debug(f"no data in {base_file_path} will add error and raise") errors.append(ValidationError(f"NOT ALLOWED - keys in global.json {base_file_path} are empty")) raise ValidationError(errors) env_set = set(env_file.keys()) if env_set - base_set: logger.debug(f"env set: {env_set}") logger.debug(f"globals set: {base_set}") errors.append(ValidationError(f"NOT ALLOWED - keys in env file {env_file_path} that are not in {base_file_path}: {env_set - base_set}")) if errors: logger.debug(f"errors found in env file {env_file_path} that are not in {base_file_path}") raise ValidationError(errors) logger.info(f"there is no keys in env file {env_file_path} that are not in {base_file_path}") except ValidationError: raise except Exception as e: exit_on_error_and_write_summary(f"error in compare_files - {e}") def validate_env_vars(global_file_path=None, env_file_path=None): """ Validate environment variables by comparing keys in global file to keys in env file. If global_file_path is not provided, it will be retrieved from GLOBAL_FILE_PATH environment variable. If env_file_path is not provided, it will be retrieved from ENV_FILE_PATH environment variable. If the env file does not exist, validation is skipped. Args: global_file_path (str, optional): Path to the global JSON file. Defaults to None. env_file_path (str, optional): Path to the environment JSON file. Defaults to None. Raises: ValidationError: If validation fails Exception: For other errors during validation """ try: if not global_file_path: global_file_path = get_required_env_var('GLOBAL_FILE_PATH') check_if_file_exists(global_file_path) if not env_file_path: env_file_path = os.getenv('ENV_FILE_PATH') if not os.path.exists(env_file_path): logger.info_yellow(f"{global_file_path} exists and file {env_file_path} does not exist, skipping comparison") return logger.debug(f"will try to validate {global_file_path} {env_file_path}") compare_files(global_file_path, env_file_path) except ValidationError as ve: logger.error(f"{global_file_path} {env_file_path} pair is invalid") raise ve except Exception as e: exit_on_error_and_write_summary(f"error in compare_global_to_env - {e}") def validate_global_file_rule(env_files_path, pr_changed_files, service: Optional[str] = None) -> list[Exception]: """ Validate environment files against the global file based on PR changes. This function follows these rules: 1. If the global file has changed, validate all environment files 2. If the global file has not changed, only validate the environment files that were changed in the PR Args: env_files_path (str): Base path for environment files (single service) or "base_path/service_name" (multi-service) pr_changed_files (list): List of files changed in the pull request service (Optional[str]): Service name for multi-service repositories. Defaults to None. Returns: list[ValidationError]: List of validation errors found Raises: RuntimeError: If an unexpected error occurs during validation """ errors: list[ValidationError] = [] try: global_file_path = os.path.join(env_files_path, GLOBAL_FILE) extra_log_message = "" if service: extra_log_message = f" for service {service}" if global_file_path in pr_changed_files: logger.info(f"global file was changed{extra_log_message}, will validate EVERY env file") env_folders = os.listdir(env_files_path) if not env_folders: logger.info(f"no env files found in {env_files_path}, exiting") exit(0) env_files_exist = False for env_folder in env_folders: logger.debug(f"check if the sub dir of {env_files_path} is in {ENV_FILES_SUBFOLDERS}:") if env_folder in ENV_FILES_SUBFOLDERS: env_files = get_all_env_files_names(str(os.path.join(env_files_path, env_folder))) for env_file in env_files: env_files_exist = True try: validate_env_vars(global_file_path, env_file) except ValidationError as e: errors.append(e) if not env_files_exist: logger.info(f"no env files found in {env_files_path}, exiting") exit(0) else: logger.info( f"global file was not changed {extra_log_message}, will validate only the env files that were changed") for env_file in pr_changed_files: if env_file.startswith(env_files_path) and env_file.endswith(".json"): try: logger.debug(f"going to validate {env_file}") validate_env_vars(global_file_path, env_file) except ValidationError as e: logger.debug(f"errors found in {env_file}") errors.append(e) return errors except Exception as e: raise RuntimeError(f"An error occurred in validate_global_file_rule as {e}") def get_services_their_env_files_changed(pr_changed_files, env_files_path) -> set: """this function walk over the list of paths that changed in the pr and take the dir after the env_files_path and add it to the set """ try: services_their_env_files_changed = set() for file in pr_changed_files: if file.startswith(env_files_path): relative_path = file.split(env_files_path)[1].strip(os.path.sep) path_components = relative_path.split(os.path.sep) if path_components: services_their_env_files_changed.add(path_components[0]) return services_their_env_files_changed except Exception as e: raise RuntimeError(f"An error occurred in get_services_their_env_files_changed as {e}") def validate_env_vars_are_in_global(is_multi_service_repo: bool, env_files_path: str, pr_changed_files: list): """ Validate that all keys present in environment files exist in their corresponding global files. Handles both multi-service and single-service repository structures differently. For multi-service repositories, it validates each service that has changed files. For single-service repositories, it validates at the repository level. Args: is_multi_service_repo (bool): Whether this is a multi-service repository env_files_path (str): Base path to the environment files pr_changed_files (list): List of files changed in the pull request Raises: RuntimeError: If an unexpected error occurs during validation SystemExit: If validation errors are found, exits with error summary """ all_errors: list[ValidationError] = [] try: if is_multi_service_repo: logger.info("this is a multi service repo, will get each changed service") services_their_env_files_changed = get_services_their_env_files_changed(pr_changed_files, env_files_path) logger.info(f"the services that their env files were changed: {services_their_env_files_changed}") for service in services_their_env_files_changed: logger.info(f"validating global file rule for service {service}") service_env_files_path = os.path.join(env_files_path, service) errors = validate_global_file_rule(service_env_files_path, pr_changed_files, service) if errors: logger.debug(f"errors encountered during validation for service {service}") all_errors.extend(errors) logger.debug(f"Finished collecting env files validation info for all services {services_their_env_files_changed}") else: # this is a single service repo logger.info("validating global file rule") errors = validate_global_file_rule(env_files_path, pr_changed_files) all_errors.extend(errors) logger.debug("Finished collecting env files validation info") if all_errors: logger.debug(f"errors encountered during validation") error_messages = extract_error_messages(all_errors) error_summary = "\n".join(error_messages) exit_on_error_and_write_summary(f"Found validation errors:\n{error_summary}") except Exception as e: raise RuntimeError(f"An error occurred in iterate_over_env_files_and_validate_global_rule as {e}") def validate_all_env_files_for_path(base_path) -> list[Exception]: """ Validate all environment files in a given path against its global file. Examines all environment files located in the standard environment subfolders (defined in ENV_FILES_SUBFOLDERS) and verifies that all keys in each environment file are present in the global.json file located at the base path. Args: base_path (str): Base directory path containing the global.json file and environment folders Returns: list[ValidationError]: Collection of validation errors found during the validation process Raises: ValidationError: If any environment file contains keys not in the global file """ logger.debug(f"Will validate all env files for path {base_path}") errors: list[ValidationError] = [] global_file_path = os.path.join(base_path, "global.json") for env_folder in ENV_FILES_SUBFOLDERS: if env_folder in ENV_FILES_SUBFOLDERS: env_files = get_all_env_files_names(str(os.path.join(base_path, env_folder))) for env_file in env_files: logger.debug(f"Will validate env file {env_file}") try: validate_env_vars(global_file_path, env_file) except ValidationError as e: errors.append(e) logger.debug(f"Finished collecting env file validation info for {env_file}") return errors def validate_all_env_files(is_multi_service_repo: bool = False): """ Validate all environment files in the repository against their global files. This is an entry point function that coordinates the validation process for the entire repository. Depending on the repository structure: - For multi-service repositories: Validates each service's environment files separately against that service's global.json - For single-service repositories: Validates all environment files against the repository's single global.json file Collects and processes validation errors, and exits with an error summary if any are found. Args: is_multi_service_repo (bool, optional): Whether this is a multi-service repository. Defaults to False. Raises: RuntimeError: If an unexpected error occurs during validation SystemExit: If validation errors are found, exits with error summary """ errors: list[ValidationError] = [] try: logger.info_yellow("Will validate all env files") if is_multi_service_repo: services = os.listdir(DEFAULT_ENV_FILES_PATH) logger.info(f"Will validate all env files for services: {services}") for svc in services: logger.debug(f"starting to validate all env files for service: {svc}") errors.extend(validate_all_env_files_for_path(os.path.join(DEFAULT_ENV_FILES_PATH, svc))) logger.debug(f"Validated all env files for service: {svc}") logger.info(f"Validated all env files for services: {services}") else: errors = validate_all_env_files_for_path(DEFAULT_ENV_FILES_PATH) if errors: logger.debug(f"errors encountered during validation") error_messages = extract_error_messages(errors) error_summary = "\n".join(error_messages) exit_on_error_and_write_summary(f"Found validation errors:\n{error_summary}") logger.info_green("validated all env files successfully") except Exception as e: raise RuntimeError(f"An error occurred in validate_all_env_files as {e}") if __name__ == '__main__': is_multi_service_repo = os.getenv('IS_MULTI_SERVICE_REPO') == 'true' validate_all_env_files(is_multi_service_repo)