fix: add necessary logic to make minimize_diff work in pipelines

AGENTS.md

@@ -540,7 +540,7 @@ convention).
 
 Per-directory readmes (`/docs/features/README.md`, `/docs/use-cases/README.md`, etc.) are
 meta-docs that explain what kind of content the directory holds. They are not navigation
-indexes and do not need a per-doc entry.
+indices and do not need a per-doc entry.
 
 **Why:** GitHub's link-checker catches dead links but does not warn when a new doc is missing
 from the index. Readers discover docs through the index readmes, not by browsing directories.

build_effective_set_generator/effective-set-generator/src/main/java/org/qubership/cloud/devops/cli/repository/implementation/FileDataConverterImpl.java

@@ -48,6 +48,8 @@
 @Slf4j
 public class FileDataConverterImpl implements FileDataConverter {
     public static final String CLEANUPER = "cleanuper";
+    /** Match SOPS decrypt output so re-encrypt with minimize_diff stays byte-stable. */
+    private static final int YAML_INDENT = 4;
     private final ObjectMapper objectMapper;
     private final FileSystemUtils fileSystemUtils;
 
@@ -124,6 +126,7 @@ private static Yaml getYamlObject(boolean expand) {
         DumperOptions options = new DumperOptions();
         options.setDefaultFlowStyle(DumperOptions.FlowStyle.BLOCK);
         options.setDefaultScalarStyle(DumperOptions.ScalarStyle.PLAIN);
+        options.setIndent(YAML_INDENT);
         options.setPrettyFlow(false);
         if (expand) {            
             options.setDereferenceAliases(true);

python/envgene/envgenehelper/crypt.py

@@ -1,5 +1,8 @@
 import os
 import re
+import shutil
+import tempfile
+from hashlib import sha256
 from os import getenv, path
 from typing import Callable
 
@@ -33,6 +36,44 @@
     'Fernet': extract_value_Fernet
 }
 
+def _path_safe_backup_name(file_path: str) -> str:
+    rel = path.relpath(file_path, BASE_DIR)
+    return rel.replace(path.sep, '__') + '.bak'
+
+
+def _cred_backup_dir() -> str:
+    backup_key_parts = [
+        f"CI_JOB_ID={getenv('CI_JOB_ID', '')}",
+        f"BASE_DIR={path.abspath(BASE_DIR)}", # for running locally
+    ]
+    backup_key = sha256("\0".join(backup_key_parts).encode("utf-8")).hexdigest()[:16]
+    return path.join(tempfile.gettempdir(), f"envgene_cred_backup_{backup_key}")
+
+
+def _cred_backup_path(file_path: str) -> str:
+    return path.join(_cred_backup_dir(), _path_safe_backup_name(file_path))
+
+
+def _backup_encrypted_cred_file(file_path: str) -> None:
+    if not is_encrypted(file_path):
+        return
+    backup_path = _cred_backup_path(file_path)
+    os.makedirs(path.dirname(backup_path), exist_ok=True)
+    shutil.copy2(file_path, backup_path)
+
+
+def _get_cred_backup_path(file_path: str) -> str | None:
+    backup_path = _cred_backup_path(file_path)
+    if path.exists(backup_path):
+        return backup_path
+    return None
+
+
+def _cleanup_cred_backups() -> None:
+    backup_dir = _cred_backup_dir()
+    if backup_dir and path.isdir(backup_dir):
+        shutil.rmtree(backup_dir, ignore_errors=True)
+
 
 def get_configured_encryption_type():
     return get_crypt_backend(), get_crypt()
@@ -71,7 +112,7 @@ def encrypt_file(file_path, *, secret_key=None, in_place=True, public_key=None,
         if not check_file_exists(old_file_path):
             minimize_diff = False
             logger.warning(f"Cred file at {old_file_path} doesn't exist, minimize_diff parameter is ignored")
-        if not is_encrypted(old_file_path, crypt_backend):
+        elif not is_encrypted(old_file_path, crypt_backend):
             minimize_diff = False
             logger.warning(f"Cred file at {old_file_path} is not encrypted, minimize_diff parameter is ignored")
     res = _handle_missing_file(file_path, default_yaml, allow_default)
@@ -155,23 +196,29 @@ def check_for_encrypted_files(files):
 
 
 def decrypt_all_cred_files_for_env(**kwargs):
-    IS_CRYPT = get_crypt()
+    _cleanup_cred_backups()
     files = get_all_necessary_cred_files()
-    if not IS_CRYPT:
+    if not get_crypt():
         check_for_encrypted_files(files)
-    else:
-        for f in files:
-            decrypt_file(f, **kwargs)
-        logger.debug("Decrypted next cred files:")
-        logger.debug(files)
+        return
+
+    for f in files:
+        _backup_encrypted_cred_file(f)
+        decrypt_file(f, **kwargs)
+    logger.debug("Decrypted next cred files:")
+    logger.debug(files)
 
 
 def encrypt_all_cred_files_for_env(**kwargs):
-    files = get_all_necessary_cred_files()
-    logger.debug("Attempting to encrypt(if crypt is true) next files:")
-    logger.debug(files)
-    for f in files:
-        encrypt_file(f, **kwargs)
+    try:
+        files = get_all_necessary_cred_files()
+        logger.debug("Attempting to encrypt(if crypt is true) next files:")
+        logger.debug(files)
+        for f in files:
+            backup = _get_cred_backup_path(f)
+            encrypt_file(f, minimize_diff=backup is not None, old_file_path=backup, **kwargs)
+    finally:
+        _cleanup_cred_backups()
 
 
 def get_crypt():

python/envgene/envgenehelper/crypt_backends/sops_handler.py

@@ -1,10 +1,11 @@
+import base64
 import os
 import subprocess
 import tempfile
 import shutil
 
 from ..business_helper import getenv_with_error
-from ..yaml_helper import openYaml, readYaml, get_or_create_nested_yaml_attribute, writeYamlToFile, dumpYamlToStr
+from ..yaml_helper import openYaml, readYaml, get_or_create_nested_yaml_attribute
 from ..logger import logger
 
 from .constants import *
@@ -23,51 +24,47 @@ def _run_SOPS(arg_str, return_codes_to_ignore=None):
         raise subprocess.SubprocessError()
     return result
 
-def _create_replace_content_sh(content):
-    delimiter = 'ENVGENE_SOPS_EDIT_CUSTOM_EOF'
-
-    script_content = f"""#!/bin/sh
-if [ -z "$1" ]; then
-    echo "No target file specified."
-    exit 1
-fi
-cat > "$1" << '{delimiter}'
-{content}
-{delimiter}
+def _create_replace_content_sh(content_bytes):
+    """Build an executable SOPS EDITOR script that writes plaintext bytes exactly."""
+    payload = base64.b64encode(content_bytes).decode('ascii')
+    script_content = f"""#!/usr/bin/env python3
+import base64
+import sys
+
+if len(sys.argv) < 2:
+    raise SystemExit("No target file specified.")
+with open(sys.argv[1], "wb") as out:
+    out.write(base64.b64decode({payload!r}))
 """
-    script = tempfile.NamedTemporaryFile(delete=False, suffix=".sh")
-
-    script.write(script_content.encode('utf-8'))
+    script = tempfile.NamedTemporaryFile(delete=False, suffix=".py", mode='w', encoding='utf-8')
+    script.write(script_content)
     script.close()
 
     return script.name
 
-def _sops_edit(file_path, new_content, public_key):
+def _sops_edit(encrypted_path, plaintext_path, public_key):
     # expects that SOPS age key is set in environment variables
-    new_content_str = dumpYamlToStr(new_content)
-    editor_path = _create_replace_content_sh(new_content_str)
+    with open(plaintext_path, 'rb') as f:
+        plaintext_bytes = f.read()
+    editor_path = _create_replace_content_sh(plaintext_bytes)
     try:
         os.chmod(editor_path, 0o777)
         os.environ['EDITOR'] = editor_path
-        sops_args = f'edit --age {public_key} {file_path}'
+        sops_args = f'edit --age {public_key} {encrypted_path}'
         _run_SOPS(sops_args, [200]) # 200 is FileHasNotBeenModified error code
     finally:
         if os.path.exists(editor_path):
             os.remove(editor_path)
 
 def _get_minimized_diff(file_path, old_file_path, public_key):
-    new_content = openYaml(file_path)
-
     tmp_file_obj = tempfile.NamedTemporaryFile(delete=False, suffix=".yml")
     tmp_file_obj.close()
+    tmp_path = tmp_file_obj.name
 
-    shutil.copy(old_file_path, tmp_file_obj.name)
-
-    _sops_edit(tmp_file_obj.name, new_content, public_key)
-    content_with_minimized_diff = openYaml(tmp_file_obj.name)
-    os.remove(tmp_file_obj.name)
+    shutil.copy(old_file_path, tmp_path)
 
-    return content_with_minimized_diff
+    _sops_edit(tmp_path, file_path, public_key)
+    return tmp_path
 
 def crypt_SOPS(file_path, secret_key, in_place, public_key, mode, minimize_diff=False, old_file_path=None, *args, **kwargs):
     if not secret_key:
@@ -90,9 +87,19 @@ def crypt_SOPS(file_path, secret_key, in_place, public_key, mode, minimize_diff=
         return openYaml(file_path)
 
     if minimize_diff and mode != "decrypt":
-        result = _get_minimized_diff(file_path, old_file_path, public_key)
-        if in_place:
-            writeYamlToFile(file_path, result)
+        tmp_path = _get_minimized_diff(file_path, old_file_path, public_key)
+        try:
+            if in_place:
+                shutil.copy2(tmp_path, file_path)
+            else:
+                result = openYaml(tmp_path)
+        finally:
+            if os.path.exists(tmp_path):
+                os.remove(tmp_path)
+        logger.info(f'The file has been {mode}ed. Path: {file_path}')
+        if not in_place:
+            return result
+        return openYaml(file_path)
     else:
         sops_args = f' --{SOPS_MODES[mode]} '
         if mode != "decrypt":

python/envgene/envgenehelper/test_crypt.py

@@ -1,13 +1,22 @@
 import os
 import copy
+import shutil
+import tempfile
 import pytest
 from subprocess import SubprocessError
 
 from ruyaml import CommentedMap
 
 from .collections_helper import compare_dicts
 
-from .crypt import decrypt_file, encrypt_file, is_encrypted
+from . import crypt as crypt_module
+from .crypt import (
+    decrypt_file,
+    encrypt_file,
+    decrypt_all_cred_files_for_env,
+    encrypt_all_cred_files_for_env,
+    is_encrypted,
+)
 from .file_helper import check_file_exists, writeToFile
 from .yaml_helper import openYaml, set_nested_yaml_attribute, writeYamlToFile
 
@@ -128,9 +137,9 @@ def test_with_file_missing(crypt_kwargs, crypt_func):
         new_yaml = crypt_func(**crypt_kwargs)
 
     new_yaml = crypt_func(**crypt_kwargs, allow_default=True)
-    assert type(new_yaml) == CommentedMap
+    assert type(new_yaml) is CommentedMap
     new_yaml = crypt_func(**crypt_kwargs, allow_default=True, default_yaml=dict)
-    assert type(new_yaml) == dict
+    assert type(new_yaml) is dict
 
     assert not check_file_exists(cred_file)
 
@@ -144,12 +153,6 @@ def test_is_encrypted(crypt_kwargs):
     decrypt_file(**crypt_kwargs)
     assert not is_encrypted(cred_file, crypt_kwargs['crypt_backend'])
 
-def compare_encrypted_files(source, target):
-    sops_metadata_to_ignore = [['sops', 'lastmodified'],['sops','mac']]
-    diff_paths, removed_paths = compare_dicts(source, target)
-    diff_paths = [item for item in diff_paths if item not in sops_metadata_to_ignore]
-    return diff_paths, removed_paths
-
 def test_minimize_diff(crypt_kwargs):
     cred_file = crypt_kwargs['file_path']
 
@@ -163,7 +166,7 @@ def test_minimize_diff(crypt_kwargs):
     decrypt_file(**crypt_kwargs)
     encrypt_file(**crypt_kwargs, minimize_diff=True, old_file_path=old_cred_file)
 
-    diff_paths, removed_paths = compare_encrypted_files(initial_enc_content, openYaml(cred_file))
+    diff_paths, removed_paths = compare_dicts(initial_enc_content, openYaml(cred_file))
     assert len(removed_paths) == 0 and len(diff_paths) == 0
 
     # test with one change
@@ -172,9 +175,72 @@ def test_minimize_diff(crypt_kwargs):
     writeYamlToFile(cred_file, new_content)
     new_enc_content = encrypt_file(**crypt_kwargs, minimize_diff=True, old_file_path=old_cred_file)
 
-    diff_paths, removed_paths = compare_encrypted_files(initial_enc_content, new_enc_content)
-    assert len(removed_paths) == 0 and len(diff_paths) == 1
+    diff_paths, removed_paths = compare_dicts(initial_enc_content, new_enc_content)
+    assert len(removed_paths) == 0
+    assert ['first_cred', 'data', 'secret'] in diff_paths
+    if crypt_kwargs.get('crypt_backend') == 'SOPS':
+        assert ['sops', 'mac'] in diff_paths
+    else:
+        assert len(diff_paths) == 1
 
     # test wrong parameter combination
     with pytest.raises(ValueError):
         encrypt_file(**crypt_kwargs, minimize_diff=True)
+
+
+@pytest.mark.parametrize("crypt_backend", ["SOPS", "Fernet"])
+def test_encrypt_all_cred_files_minimize_diff(monkeypatch, crypt_backend):
+    if crypt_backend == "SOPS":
+        secret_key = crypt_test_data[0]["secret_key"]
+        public_key = crypt_test_data[0]["public_key"]
+        monkeypatch.setenv("ENVGENE_AGE_PRIVATE_KEY", secret_key)
+        monkeypatch.setenv("PUBLIC_AGE_KEYS", public_key)
+    else:
+        secret_key = crypt_test_data[1]["secret_key"]
+        public_key = None
+        monkeypatch.setenv("SECRET_KEY", secret_key)
+
+    project_dir = tempfile.mkdtemp()
+    try:
+        cred_dir = os.path.join(project_dir, "configuration", "credentials")
+        os.makedirs(cred_dir)
+        cred_file = os.path.join(cred_dir, "credentials.yml")
+        shutil.copy(TEST_FILE, cred_file)
+
+        config_path = os.path.join(project_dir, "configuration", "config.yml")
+        with open(config_path, "w", encoding="utf-8") as config_file:
+            config_file.write(f"crypt: true\ncrypt_backend: {crypt_backend}\n")
+
+        monkeypatch.setenv("CI_PROJECT_DIR", project_dir)
+        monkeypatch.delenv("ENV_NAMES", raising=False)
+        monkeypatch.setattr(crypt_module, "BASE_DIR", project_dir)
+
+        crypt_kwargs = {
+            "file_path": cred_file,
+            "crypt_backend": crypt_backend,
+            "secret_key": secret_key,
+            "public_key": public_key,
+            "ignore_is_crypt": True,
+            "is_crypt": True,
+        }
+        bulk_kwargs = {key: value for key, value in crypt_kwargs.items() if key != "file_path"}
+
+        initial_enc_content = encrypt_file(**crypt_kwargs)
+        decrypt_all_cred_files_for_env(**bulk_kwargs)
+
+        backup_dir = crypt_module._cred_backup_dir()
+        backup_path = crypt_module._cred_backup_path(cred_file)
+        assert backup_dir.startswith(tempfile.gettempdir())
+        assert not backup_dir.startswith(project_dir)
+        assert os.path.exists(backup_path)
+
+        encrypt_all_cred_files_for_env(**bulk_kwargs)
+        assert not os.path.exists(backup_dir)
+
+        diff_paths, removed_paths = compare_dicts(
+            initial_enc_content, openYaml(cred_file)
+        )
+        assert len(removed_paths) == 0 and len(diff_paths) == 0
+    finally:
+        crypt_module._cleanup_cred_backups()
+        shutil.rmtree(project_dir, ignore_errors=True)