""" Executa invocações de Lambda a partir de um arquivo de configuração JSON/JSONC. Recursos suportados: - Placeholders de ambiente no estilo ${VAR} em toda a configuração - Comentários na configuração (JSONC: // e /* */) - Carregamento de conteúdo XML de arquivo externo - Validação opcional por cenário (substring regex no resultado) """ from __future__ import annotations import argparse import json import re from dataclasses import dataclass from pathlib import Path from string import Template from typing import Any, Dict, List, Optional from colorama import Fore from common import print_message from src.datalake.commons import lambda_utils from src.datalake.commons.ci_utils import BLUE, RESET from src.datalake.commons.data_env import load_runtime_env @dataclass class Validation(object): enabled: bool = False success_regex: Optional[str] = None failure_message: Optional[str] = None @dataclass class Scenario(object): name: str execution_type: str lambda_function: str payload: Dict[str, Any] xml_file: str validation: Validation def _strip_jsonc(content: str) -> str: # Remove comentários de bloco e de linha sem depender de libs externas. no_block = re.sub(r"/\*.*?\*/", "", content, flags=re.DOTALL) no_line = re.sub(r"(^|\s)//.*$", "", no_block, flags=re.MULTILINE) return no_line def _expand_env(env: Dict[str, str], value: Any) -> Any: if isinstance(value, str): return Template(value).safe_substitute(env) if isinstance(value, list): return [_expand_env(env, item) for item in value] if isinstance(value, dict): return {k: _expand_env(env, v) for k, v in value.items()} return value def load_config(env: Dict[str, str], path: Path) -> Dict[str, Any]: raw = path.read_text(encoding="utf-8") cfg = json.loads(_strip_jsonc(raw)) return _expand_env(env, cfg) def parse_scenarios(config: Dict[str, Any]) -> List[Scenario]: scenarios: List[Scenario] = [] for item in config.get("scenarios", []): validation_cfg = item.get("validation", {}) scenarios.append( Scenario( name=item["name"], execution_type=item.get("execution_type", "default"), lambda_function=item.get("lambda_function", "fileProcessorFunctionId"), payload=item["payload"], xml_file=item.get("xml_file", ""), validation=Validation( enabled=bool(validation_cfg.get("enabled", False)), success_regex=validation_cfg.get("success_regex"), failure_message=validation_cfg.get("failure_message"), ), ) ) return scenarios def build_payload(base_payload: Dict[str, Any], xml_file_path: Path) -> Dict[str, Any]: payload = dict(base_payload) payload["object_content"] = xml_file_path.read_text(encoding="utf-8") return payload def validate_scenario( output: dict[str, Any], validation: Validation, scenario_name: str ) -> None: """ Valida a saída de um cenário de teste. Args: output (dict[str, Any]): Saída da função Lambda. validation (Validation): Configuração de validação. scenario_name (str): Nome do cenário. Raises: AssertionError: Se a validação falhar. """ if not validation.enabled: print_message(f"[Dados] validação desabilitada para cenário: {scenario_name}") return error_message = None if validation.success_regex: if re.search( validation.success_regex, str(output), flags=re.IGNORECASE | re.MULTILINE ): print_message( f"[Dados] validação por regex OK para cenário: '{scenario_name}'" ) else: error_message = validation.failure_message or ( f"Validação falhou no cenário '{scenario_name}'. Regex não encontrada: {validation.success_regex}" ) raise AssertionError(error_message) def validate_errors_list(errors_list: List[str]) -> int: if errors_list: print_message( f"\n{Fore.RED}[Dados] execução finalizada com erros, verifique os logs acima para mais detalhes" ) for error in errors_list: print_message(f"[Dados] erro em: {error} - ") return 1 else: print_message(f"\n{BLUE}[Dados] execução finalizada com sucesso{RESET}") return 0 def load_payload(config_path: Path, scenario: Scenario) -> dict[str, Any]: if scenario.xml_file: xml_file = (config_path.parent.parent / scenario.xml_file).resolve() if not xml_file.exists(): raise FileNotFoundError(f"Arquivo XML não encontrado: {xml_file}") payload = build_payload(scenario.payload, xml_file) else: payload = scenario.payload print_message(f"[Dados] payload gerado em: {payload}") return payload def invoke_lambda_and_validate( env: dict[str, str], errors_list: list[str], scenario: Scenario, payload: dict[str, Any], ) -> None: try: response_json = lambda_utils.invoke_lambda( name=scenario.lambda_function, payload=json.dumps(payload), env=env, validate=True, expect_success=False, ) validate_scenario(response_json, scenario.validation, scenario.name) except (AssertionError, ValueError) as e: print_message( f"[Dados] erro ao processar cenário: '{scenario.name}' - erro: {type(e).__name__}:{str(e)} " f"- payload: {payload if payload else 'N/A'} " ) errors_list.append(f"'{scenario.name}' - {type(e).__name__}:{str(e)}") def run_scenarios(config_path: Path, only: Optional[List[str]]) -> int: env = load_runtime_env(apply_set_all=True, configure_aws=True) config = load_config(env, config_path) scenarios = parse_scenarios(config) if only: allowed = set(only) scenarios = [s for s in scenarios if s.name in allowed] if scenarios: errors_list = [] for scenario in scenarios: print_message( f"\n{Fore.YELLOW}[Dados] lambda invoke scenario - {scenario.name}" ) print_message(f"[Dados] tipo de execução: {scenario.execution_type}") payload = load_payload(config_path, scenario) invoke_lambda_and_validate(env, errors_list, scenario, payload) return validate_errors_list(errors_list) else: print_message("Nenhum cenário para executar.") return -1 def main() -> int: parser = argparse.ArgumentParser( description="Invoca lambdas do xml processor por configuração JSON" ) parser.add_argument( "--config", default=str(Path(__file__).with_name("invoke_lambdas.config.json")), help="Caminho do arquivo de configuração JSON/JSONC", ) parser.add_argument( "--only", nargs="*", help="Executa apenas os cenários informados pelo nome", ) args = parser.parse_args() try: return run_scenarios(config_path=Path(args.config).resolve(), only=args.only) except Exception as exc: print_message(f"[ERRO] {exc}") return 1 if __name__ == "__main__": raise SystemExit(main())