diff --git a/demisto_sdk/commands/common/git_util.py b/demisto_sdk/commands/common/git_util.py index f5ce611d10..28353b09d5 100644 --- a/demisto_sdk/commands/common/git_util.py +++ b/demisto_sdk/commands/common/git_util.py @@ -21,7 +21,11 @@ DEMISTO_GIT_UPSTREAM, ISO_TIMESTAMP_FORMAT, PACKS_FOLDER, + PRIVATE_REPO_STATUS_FILE_CONFIGURATION, + PRIVATE_REPO_STATUS_FILE_PRIVATE, + PRIVATE_REPO_STATUS_FILE_TEST_CONF, ) +from demisto_sdk.commands.common.handlers import DEFAULT_JSON_HANDLER from demisto_sdk.commands.common.logger import logger @@ -696,11 +700,82 @@ def renamed_files( return all_renamed_files def get_all_changed_pack_ids(self, prev_ver: str) -> Set[str]: - return { + """Get all pack IDs that have been changed. + + This includes packs with modified, added, or renamed files from both + the content repository and private repositories (if status files exist). + For renamed files, both the source and destination packs are included + to ensure proper graph updates when items are moved between packs. + + Args: + prev_ver: The base branch/commit to compare against + + Returns: + Set of pack IDs that have changes + """ + # Get pack IDs from modified and added files + changed_pack_ids = { file.parts[1] for file in self._get_all_changed_files(prev_ver) | self._get_staged_files() if file.parts[0] == PACKS_FOLDER } + + # Get renamed files with both old and new paths from content repo + renamed_files: Set[Tuple[Path, Path]] = self.renamed_files( # type: ignore[assignment] + prev_ver=prev_ver, + committed_only=False, + staged_only=False, + get_only_current_file_names=False, + ) + + # Add pack IDs from both old and new paths of renamed files + for renamed_tuple in renamed_files: + if isinstance(renamed_tuple, tuple) and len(renamed_tuple) == 2: + old_path, new_path = renamed_tuple + if old_path.parts[0] == PACKS_FOLDER: + changed_pack_ids.add(old_path.parts[1]) + if new_path.parts[0] == PACKS_FOLDER: + changed_pack_ids.add(new_path.parts[1]) + + # Process private repository status files if they exist + artifacts_folder = os.getenv("ARTIFACTS_FOLDER", "") + logs_dir = Path(artifacts_folder) / "logs" if artifacts_folder else Path("logs") + + status_files = [ + logs_dir / PRIVATE_REPO_STATUS_FILE_PRIVATE, + logs_dir / PRIVATE_REPO_STATUS_FILE_TEST_CONF, + logs_dir / PRIVATE_REPO_STATUS_FILE_CONFIGURATION, + ] + + for status_file in status_files: + try: + if status_file.exists(): + with open(status_file, "r") as f: + file_statuses = DEFAULT_JSON_HANDLER.load(f) + + for file_path_str, status_info in file_statuses.items(): + if not file_path_str: + continue + + # Only process renamed files + if status_info.get("status") == "renamed": + old_path_str = status_info.get("old_path") + if old_path_str: + old_path = Path(old_path_str) + new_path = Path(file_path_str) + + # Add pack IDs from both old and new paths + if old_path.parts and old_path.parts[0] == PACKS_FOLDER and len(old_path.parts) > 1: + changed_pack_ids.add(old_path.parts[1]) + if new_path.parts and new_path.parts[0] == PACKS_FOLDER and len(new_path.parts) > 1: + changed_pack_ids.add(new_path.parts[1]) + + except DEFAULT_JSON_HANDLER.JSONDecodeError as e: + logger.debug(f"Invalid JSON format in {status_file}: {str(e)}") + except Exception as e: + logger.debug(f"Error processing {status_file}: {str(e)}") + + return changed_pack_ids def _get_untracked_files(self, requested_status: str) -> set: """return all untracked files of the given requested status.