from pathlib import Path from colorama import Fore from common import ( cd_path, exit_message, get_env_variable_required, print_message, repository_is_flex, ) from common.artifactory_helpers import check_maven_artifact_exists from common.file_helpers import ( get_java_version, get_xml_namespace_and_root_element, load_sdl_properties, ) from common.sonar_helper import SonarHelper, SonarScannerType from src.interface.maven_project_with_docker import MavenProjectWithDockerInterface class MavenSDL(MavenProjectWithDockerInterface): def __init__(self) -> None: super().__init__() self.properties = load_sdl_properties() if not self.properties: exit_message( "O projeto está usando o script de SDL porém não possui " "nenhuma propriedade no/não possui o arquivo sdl.properties." ) self.generator_java_output = self._get_java_output() self.output_pom_exists = Path(f"{self.generator_java_output}/pom.xml").is_file() self.sources_generated = False def set_version(self): # Prints the sed command to update the version in the `sdl.properties` file for logging purposes, # allowing the user to execute it on their local machine. print_message( f'sed -i "s/generator.app.version=.*/generator.app.version={self.version}/" sdl.properties', Fore.LIGHTGREEN_EX, ) self._replace_line_in_file( "sdl.properties", "generator.app.version=", f"generator.app.version={self.version}", ) super().set_version() if self.output_pom_exists or self._custom_parent_file(): self._execute_mvn_command( f"versions:set -DgenerateBackupPoms=false -DnewVersion={self.version} " f"-DprocessAllModules -N versions:update-child-modules --file {self.generator_java_output}" ) else: self._run_generate_sources() def generate(self): super().generate() if self.output_pom_exists: self._run_generate_sources() def compile(self): if self.buildable_project.skip_build(): return self._execute_mvn_command( f"install -DskipTests=true -Ddockerfile.skip=true --file {self.generator_java_output}" ) def unit_test(self): if self.buildable_project.skip_build(): return self._docker_login() self._execute_mvn_command( f"install -Ddockerfile.skip=true --file {self.generator_java_output}" ) def sonar_scanner(self): sonar_helper = SonarHelper( sonar_scanner_type=SonarScannerType.SDL, skip_build=self.buildable_project.skip_build(), ) sonar_helper.scanner_analyze() def package(self): self._validate_artifacts_before_publish() java_version = get_java_version() if java_version >= "17": project_root = Path(get_env_variable_required("CI_PROJECT_DIR")) self._execute_mvn_command( f"deploy -DskipTests=true -Ddockerfile.skip=true --file {self.generator_java_output}" ) cd_path(f"{self.generator_java_output}/impl/target/docker") self._push_image() if repository_is_flex(): cd_path(str(project_root)) cd_path(f"{self.generator_java_output}/http-impl/target/docker") self.image_name = f"{self.image_name}-http" self._push_image() else: self._docker_login() self._create_repository() self._execute_mvn_command( f"deploy -DskipTests=true --file {self.generator_java_output}" ) def build_docker_image(self): self._docker_login() java_version = get_java_version() if java_version >= "17": project_root = Path(get_env_variable_required("CI_PROJECT_DIR")) cd_path(f"{self.generator_java_output}/impl/target/docker") self._build_image() self._save_image() if repository_is_flex(): cd_path(str(project_root)) cd_path(f"{self.generator_java_output}/http-impl/target/docker") self.image_name = f"{self.image_name}-http" self._build_image() self._save_image() else: self._execute_mvn_command( f"install -DskipTests=true --file {self.generator_java_output}" ) self._save_image() def trivy_scan_image(self, custom_args=""): for tar_file in ( Path(self.ci_project_dir) .joinpath(self.ci_project_dir, "docker_image") .glob("*.tar") ): self.image_name = tar_file.stem super().trivy_scan_image(custom_args="--skip-dirs app") def _get_java_output(self) -> str: try: output = self.properties["generator.java.output"] except KeyError: output = get_env_variable_required("GENERATOR_JAVA_OUTPUT") return output @staticmethod def _replace_line_in_file(file_path, line_to_find, new_line): file_path = Path(file_path) if not file_path.is_file(): exit_message(f"File {file_path} not found") # Open the file in read mode and read lines with open(file_path, "r", encoding="utf-8") as file: lines = file.readlines() # Replace the line for i, line in enumerate(lines): if line.startswith(line_to_find): lines[i] = new_line + "\n" break # Open the file in write mode and write the modified lines with open(file_path, "w", encoding="utf-8") as file: file.writelines(lines) def _run_generate_sources(self): if not self.sources_generated: self._execute_mvn_command("generate-sources") self.sources_generated = True def _custom_parent_file(self): if "generator.java.parent.pom.custom" in self.properties: self._run_generate_sources() return True return False def _check_mutations_xml(self, xml_dir="target/pit-reports/"): return super()._check_mutations_xml("java/impl/target/pit-reports/") def _get_parent_group_id(self, root, namespace, mvn_group_id): """Extract group_id from root or parent element.""" group_id = root.find(mvn_group_id, namespace) if group_id is None: parent = root.find("mvn:parent", namespace) if parent is not None: group_id = parent.find(mvn_group_id, namespace) return group_id def _add_parent_module(self, modules, group_id, artifact_id): """Add parent module to modules list.""" if group_id is not None and artifact_id is not None: modules.append( { "group_id": group_id.text, "artifact_id": artifact_id.text, } ) def _process_child_module( self, modules, module_element, pom_path, group_id, namespace, mvn_group_id ): """Process a single child module.""" module_path = pom_path.parent / module_element.text / "pom.xml" if not module_path.exists(): return mod_namespace, mod_root = get_xml_namespace_and_root_element(str(module_path)) mod_group_id = mod_root.find(mvn_group_id, mod_namespace) if mod_group_id is None: mod_group_id = group_id mod_artifact_id = mod_root.find("mvn:artifactId", mod_namespace) if mod_artifact_id is None: return modules.append( { "group_id": ( mod_group_id.text if mod_group_id is not None else group_id.text ), "artifact_id": mod_artifact_id.text, } ) def _get_maven_modules(self, pom_path: Path) -> list: """ Detecta todos os módulos Maven do projeto (parent + children). Returns: Lista de dicts com {group_id, artifact_id} """ MVN_GROUP_ID = "mvn:groupId" modules = [] namespace, root = get_xml_namespace_and_root_element(str(pom_path)) group_id = self._get_parent_group_id(root, namespace, MVN_GROUP_ID) artifact_id = root.find("mvn:artifactId", namespace) self._add_parent_module(modules, group_id, artifact_id) modules_element = root.find("mvn:modules", namespace) if modules_element is not None: for module in modules_element.findall("mvn:module", namespace): self._process_child_module( modules, module, pom_path, group_id, namespace, MVN_GROUP_ID ) return modules def _check_maven_artifacts(self, group_id: str, artifact_id: str) -> bool: """ Verifica se um artefato Maven específico existe no repositório. Args: group_id: ID do grupo Maven artifact_id: ID do artefato Returns: True se o artefato existe, False caso contrário """ try: return check_maven_artifact_exists(group_id, artifact_id, self.version) except Exception as e: print_message( f"⚠️ Aviso: Não foi possível verificar artefato Maven {artifact_id}: {e}", Fore.YELLOW, ) return False def _validate_maven_artifacts(self) -> list: """ Valida se artefatos Maven já existem. Returns: Lista de artifact_ids que já existem """ if not self.output_pom_exists: exit_message("Arquivo pom.xml não encontrado para validação") pom_path = Path(f"{self.generator_java_output}/pom.xml") modules = self._get_maven_modules(pom_path) maven_artifacts_found = [] for module in modules: exists = self._check_maven_artifacts( module["group_id"], module["artifact_id"] ) if exists: maven_artifacts_found.append(module["artifact_id"]) return maven_artifacts_found def _build_maven_artifacts_list(self, maven_artifacts: list) -> str: """ Constrói lista formatada de artefatos Maven encontrados. Args: maven_artifacts: Lista de artifact_ids que existem Returns: String formatada com a lista de artefatos """ if not maven_artifacts: return " (nenhum)" artifacts_formatted = "\n".join( [f" • {artifact}" for artifact in maven_artifacts] ) return artifacts_formatted def _validate_artifacts_before_publish(self): """ Valida se os artefatos Maven já existem antes de publicar. Implementa fail-fast para evitar sobrescrita de versões. """ maven_artifacts_found = self._validate_maven_artifacts() if not maven_artifacts_found: return print_message( f""" ╔══════════════════════════════════════════════════════════════════════════╗ ⚠️ ERRO: Artefatos Maven já publicados para a versão {self.version} ╚══════════════════════════════════════════════════════════════════════════╝ 🚫 MÓDULOS MAVEN EXISTENTES: {self._build_maven_artifacts_list(maven_artifacts_found)} ❌ POR QUE NÃO É PERMITIDO SOBRESCREVER: Sobrescrever versões publicadas viola o princípio de imutabilidade de artefatos e pode causar inconsistências em ambientes que já utilizam esta versão. ✅ AÇÕES NECESSÁRIAS: Para republicar esta versão: • Utilize o mecanismo de deleção de artefatos pelo Bastion • Aguarde a confirmação da deleção de TODOS os artefatos Maven listados acima • Execute o pipeline novamente 📊 ARTEFATOS ENCONTRADOS: {len(maven_artifacts_found)} módulo(s) """, Fore.RED, ) exit_message("Falha na validação pré-publicação. Veja instruções acima.")