refactor(bump): cleanup related to update_version_file

Author: bearomorphism

commitizen/bump.py

@@ -3,13 +3,13 @@
 import os
 import re
 from collections import OrderedDict
-from collections.abc import Iterable
+from collections.abc import Generator, Iterable
 from glob import iglob
 from logging import getLogger
 from string import Template
 from typing import cast
 
-from commitizen.defaults import BUMP_MESSAGE, ENCODING, MAJOR, MINOR, PATCH
+from commitizen.defaults import BUMP_MESSAGE, MAJOR, MINOR, PATCH
 from commitizen.exceptions import CurrentVersionNotFoundError
 from commitizen.git import GitCommit, smart_open
 from commitizen.version_schemes import Increment, Version
@@ -64,8 +64,8 @@ def update_version_in_files(
     new_version: str,
     files: Iterable[str],
     *,
-    check_consistency: bool = False,
-    encoding: str = ENCODING,
+    check_consistency: bool,
+    encoding: str,
 ) -> list[str]:
     """Change old version to the new one in every file given.
 
@@ -75,16 +75,22 @@ def update_version_in_files(
 
     Returns the list of updated files.
     """
-    # TODO: separate check step and write step
-    updated = []
-    for path, regex in _files_and_regexes(files, current_version):
-        current_version_found, version_file = _bump_with_regex(
-            path,
-            current_version,
-            new_version,
-            regex,
-            encoding=encoding,
-        )
+    updated_files = []
+
+    for path, pattern in _resolve_files_and_regexes(files, current_version):
+        current_version_found = False
+        bumped_lines = []
+
+        with open(path, encoding=encoding) as version_file:
+            for line in version_file:
+                bumped_line = (
+                    line.replace(current_version, new_version)
+                    if pattern.search(line)
+                    else line
+                )
+
+                current_version_found = current_version_found or bumped_line != line
+                bumped_lines.append(bumped_line)
 
         if check_consistency and not current_version_found:
             raise CurrentVersionNotFoundError(
@@ -93,53 +99,32 @@ def update_version_in_files(
                 "version_files are possibly inconsistent."
             )
 
+        bumped_version_file_content = "".join(bumped_lines)
+
         # Write the file out again
         with smart_open(path, "w", encoding=encoding) as file:
-            file.write(version_file)
-        updated.append(path)
-    return updated
+            file.write(bumped_version_file_content)
+        updated_files.append(path)
+
+    return updated_files
 
 
-def _files_and_regexes(patterns: Iterable[str], version: str) -> list[tuple[str, str]]:
+def _resolve_files_and_regexes(
+    patterns: Iterable[str], version: str
+) -> Generator[tuple[str, re.Pattern], None, None]:
     """
     Resolve all distinct files with their regexp from a list of glob patterns with optional regexp
     """
-    out: set[tuple[str, str]] = set()
+    filepath_set: set[tuple[str, str]] = set()
     for pattern in patterns:
         drive, tail = os.path.splitdrive(pattern)
         path, _, regex = tail.partition(":")
         filepath = drive + path
-        if not regex:
-            regex = re.escape(version)
+        regex = regex or re.escape(version)
 
-        for file in iglob(filepath):
-            out.add((file, regex))
+        filepath_set.update((path, regex) for path in iglob(filepath))
 
-    return sorted(out)
-
-
-def _bump_with_regex(
-    version_filepath: str,
-    current_version: str,
-    new_version: str,
-    regex: str,
-    encoding: str = ENCODING,
-) -> tuple[bool, str]:
-    current_version_found = False
-    lines = []
-    pattern = re.compile(regex)
-    with open(version_filepath, encoding=encoding) as f:
-        for line in f:
-            if not pattern.search(line):
-                lines.append(line)
-                continue
-
-            bumped_line = line.replace(current_version, new_version)
-            if bumped_line != line:
-                current_version_found = True
-            lines.append(bumped_line)
-
-    return current_version_found, "".join(lines)
+    return ((path, re.compile(regex)) for path, regex in sorted(filepath_set))
 
 
 def create_commit_message(

tests/test_bump_update_version_in_files.py

@@ -103,7 +103,11 @@ def test_update_version_in_files(version_files, file_regression):
     old_version = "1.2.3"
     new_version = "2.0.0"
     bump.update_version_in_files(
-        old_version, new_version, version_files, encoding="utf-8"
+        old_version,
+        new_version,
+        version_files,
+        check_consistency=False,
+        encoding="utf-8",
     )
 
     file_contents = ""
@@ -119,7 +123,9 @@ def test_partial_update_of_file(version_repeated_file, file_regression):
     regex = "version"
     location = f"{version_repeated_file}:{regex}"
 
-    bump.update_version_in_files(old_version, new_version, [location], encoding="utf-8")
+    bump.update_version_in_files(
+        old_version, new_version, [location], check_consistency=False, encoding="utf-8"
+    )
     with open(version_repeated_file, encoding="utf-8") as f:
         file_regression.check(f.read(), extension=".json")
 
@@ -129,7 +135,9 @@ def test_random_location(random_location_version_file, file_regression):
     new_version = "2.0.0"
     location = f"{random_location_version_file}:version.+Commitizen"
 
-    bump.update_version_in_files(old_version, new_version, [location], encoding="utf-8")
+    bump.update_version_in_files(
+        old_version, new_version, [location], check_consistency=False, encoding="utf-8"
+    )
     with open(random_location_version_file, encoding="utf-8") as f:
         file_regression.check(f.read(), extension=".lock")
 
@@ -141,7 +149,9 @@ def test_duplicates_are_change_with_no_regex(
     new_version = "2.0.0"
     location = f"{random_location_version_file}:version"
 
-    bump.update_version_in_files(old_version, new_version, [location], encoding="utf-8")
+    bump.update_version_in_files(
+        old_version, new_version, [location], check_consistency=False, encoding="utf-8"
+    )
     with open(random_location_version_file, encoding="utf-8") as f:
         file_regression.check(f.read(), extension=".lock")
 
@@ -153,7 +163,9 @@ def test_version_bump_increase_string_length(
     new_version = "1.2.10"
     location = f"{multiple_versions_increase_string}:version"
 
-    bump.update_version_in_files(old_version, new_version, [location], encoding="utf-8")
+    bump.update_version_in_files(
+        old_version, new_version, [location], check_consistency=False, encoding="utf-8"
+    )
     with open(multiple_versions_increase_string, encoding="utf-8") as f:
         file_regression.check(f.read(), extension=".txt")
 
@@ -165,7 +177,9 @@ def test_version_bump_reduce_string_length(
     new_version = "2.0.0"
     location = f"{multiple_versions_reduce_string}:version"
 
-    bump.update_version_in_files(old_version, new_version, [location], encoding="utf-8")
+    bump.update_version_in_files(
+        old_version, new_version, [location], check_consistency=False, encoding="utf-8"
+    )
     with open(multiple_versions_reduce_string, encoding="utf-8") as f:
         file_regression.check(f.read(), extension=".txt")
 
@@ -204,7 +218,9 @@ def test_multiple_versions_to_bump(
     new_version = "1.2.10"
     location = f"{multiple_versions_to_update_poetry_lock}:version"
 
-    bump.update_version_in_files(old_version, new_version, [location], encoding="utf-8")
+    bump.update_version_in_files(
+        old_version, new_version, [location], check_consistency=False, encoding="utf-8"
+    )
     with open(multiple_versions_to_update_poetry_lock, encoding="utf-8") as f:
         file_regression.check(f.read(), extension=".toml")
 
@@ -220,8 +236,178 @@ def test_update_version_in_globbed_files(commitizen_config_file, file_regression
     version_files = [commitizen_config_file.dirpath("*.toml")]
 
     bump.update_version_in_files(
-        old_version, new_version, version_files, encoding="utf-8"
+        old_version,
+        new_version,
+        version_files,
+        check_consistency=False,
+        encoding="utf-8",
     )
 
     for file in commitizen_config_file, other:
         file_regression.check(file.read_text("utf-8"), extension=".toml")
+
+
+def test_update_version_in_files_with_check_consistency_true(version_files):
+    """Test update_version_in_files with check_consistency=True (success case)."""
+    old_version = "1.2.3"
+    new_version = "2.0.0"
+
+    # This should succeed because all files contain the current version
+    updated_files = bump.update_version_in_files(
+        old_version,
+        new_version,
+        version_files,
+        check_consistency=True,
+        encoding="utf-8",
+    )
+
+    # Verify that all files were updated
+    assert len(updated_files) == len(version_files)
+    for file_path in updated_files:
+        assert file_path in version_files
+
+
+def test_update_version_in_files_with_check_consistency_true_failure(
+    commitizen_config_file, inconsistent_python_version_file
+):
+    """Test update_version_in_files with check_consistency=True (failure case)."""
+    old_version = "1.2.3"
+    new_version = "2.0.0"
+    version_files = [commitizen_config_file, inconsistent_python_version_file]
+
+    # This should fail because inconsistent_python_version_file doesn't contain the current version
+    with pytest.raises(CurrentVersionNotFoundError) as excinfo:
+        bump.update_version_in_files(
+            old_version,
+            new_version,
+            version_files,
+            check_consistency=True,
+            encoding="utf-8",
+        )
+
+    expected_msg = (
+        f"Current version {old_version} is not found in {inconsistent_python_version_file}.\n"
+        "The version defined in commitizen configuration and the ones in "
+        "version_files are possibly inconsistent."
+    )
+    assert expected_msg in str(excinfo.value)
+
+
+def test_update_version_in_files_with_latin1_encoding(tmp_path):
+    """Test update_version_in_files with latin-1 encoding."""
+    # Create a test file with latin-1 encoding
+    test_file = tmp_path / "test_latin1.txt"
+    content = 'version = "1.2.3"\n# This is a test file with latin-1 encoding\n'
+    test_file.write_text(content, encoding="latin-1")
+
+    old_version = "1.2.3"
+    new_version = "2.0.0"
+
+    updated_files = bump.update_version_in_files(
+        old_version,
+        new_version,
+        [str(test_file)],
+        check_consistency=True,
+        encoding="latin-1",
+    )
+
+    # Verify the file was updated
+    assert len(updated_files) == 1
+    assert str(test_file) in updated_files
+
+    # Verify the content was updated correctly
+    updated_content = test_file.read_text(encoding="latin-1")
+    assert f'version = "{new_version}"' in updated_content
+    assert f'version = "{old_version}"' not in updated_content
+
+
+def test_update_version_in_files_with_utf16_encoding(tmp_path):
+    """Test update_version_in_files with utf-16 encoding."""
+    # Create a test file with utf-16 encoding
+    test_file = tmp_path / "test_utf16.txt"
+    content = 'version = "1.2.3"\n# This is a test file with utf-16 encoding\n'
+    test_file.write_text(content, encoding="utf-16")
+
+    old_version = "1.2.3"
+    new_version = "2.0.0"
+
+    updated_files = bump.update_version_in_files(
+        old_version,
+        new_version,
+        [str(test_file)],
+        check_consistency=True,
+        encoding="utf-16",
+    )
+
+    # Verify the file was updated
+    assert len(updated_files) == 1
+    assert str(test_file) in updated_files
+
+    # Verify the content was updated correctly
+    updated_content = test_file.read_text(encoding="utf-16")
+    assert f'version = "{new_version}"' in updated_content
+    assert f'version = "{old_version}"' not in updated_content
+
+
+def test_update_version_in_files_return_value(version_files):
+    """Test that update_version_in_files returns the correct list of updated files."""
+    old_version = "1.2.3"
+    new_version = "2.0.0"
+
+    updated_files = bump.update_version_in_files(
+        old_version,
+        new_version,
+        version_files,
+        check_consistency=False,
+        encoding="utf-8",
+    )
+
+    # Verify return value is a list
+    assert isinstance(updated_files, list)
+
+    # Verify all files in the input are in the returned list
+    assert len(updated_files) == len(version_files)
+    for file_path in version_files:
+        assert file_path in updated_files
+
+    # Verify the returned paths are strings
+    for file_path in updated_files:
+        assert isinstance(file_path, str)
+
+
+def test_update_version_in_files_return_value_partial_update(tmp_path):
+    """Test return value when only some files are updated."""
+    # Create two test files
+    file1 = tmp_path / "file1.txt"
+    file2 = tmp_path / "file2.txt"
+
+    # File1 contains the version to update
+    file1.write_text('version = "1.2.3"\n')
+
+    # File2 doesn't contain the version
+    file2.write_text("some other content\n")
+
+    old_version = "1.2.3"
+    new_version = "2.0.0"
+
+    updated_files = bump.update_version_in_files(
+        old_version,
+        new_version,
+        [str(file1), str(file2)],
+        check_consistency=False,
+        encoding="utf-8",
+    )
+
+    # Verify return value
+    assert isinstance(updated_files, list)
+    assert len(updated_files) == 2  # Both files should be in the list
+    assert str(file1) in updated_files
+    assert str(file2) in updated_files
+
+    # Verify file1 was actually updated
+    content1 = file1.read_text(encoding="utf-8")
+    assert f'version = "{new_version}"' in content1
+
+    # Verify file2 was not changed
+    content2 = file2.read_text(encoding="utf-8")
+    assert content2 == "some other content\n"