Merge pull request #13 from plesk/mass-rpmnew-handle-function

Make it possible to handle all *.rpmnew files in a directory

pleskdistup/common/src/rpm.py

@@ -144,6 +144,16 @@ def handle_rpmnew(original_path: str) -> bool:
     return True
 
 
+def handle_all_rpmnew_files(directory: str) -> typing.List[str]:
+    fixed_list = []
+    for file in files.find_files_case_insensitive(directory, ["*.rpmnew"]):
+        original_file = file[:-len(".rpmnew")]
+        if handle_rpmnew(original_file):
+            fixed_list.append(original_file)
+
+    return fixed_list
+
+
 def find_related_repofiles(repository_file: str) -> typing.List[str]:
     return files.find_files_case_insensitive("/etc/yum.repos.d", repository_file)
 

pleskdistup/common/tests/rpmtests.py

@@ -1,6 +1,7 @@
 # Copyright 2023-2024. WebPros International GmbH. All rights reserved.
 import unittest
 import os
+import shutil
 
 import src.rpm as rpm
 
@@ -232,12 +233,17 @@ def test_write_exsisted_repodata(self):
 
 
 class HandleRpmnewFilesTests(unittest.TestCase):
+    test_dir: str = "rpm_test_dir"
+
     def tearDown(self):
         tests_related_files = ["test.txt", "test.txt.rpmnew", "test.txt.rpmsave"]
         for file in tests_related_files:
             if os.path.exists(file):
                 os.remove(file)
 
+        if os.path.exists(self.test_dir):
+            shutil.rmtree(self.test_dir)
+
     def test_no_rpmnew(self):
         with open("test.txt", "w") as f:
             f.write("test")
@@ -267,3 +273,41 @@ def test_missing_original(self):
         self.assertEqual(open("test.txt").read(), "2")
 
         self.assertFalse(os.path.exists("test.txt.rpmsave"))
+
+    def test_handle_whole_directory(self):
+        os.mkdir(self.test_dir)
+
+        original_files = {
+            "test1.txt": "1",
+            "test1.txt.rpmnew": "2",
+            "test2.txt": "3",
+            "test2.txt.rpmnew": "4",
+            "test3.txt": "5",
+            "test4.txt.rpmnew": "6"
+        }
+
+        expected_files = {
+            "test1.txt": "2",
+            "test2.txt": "4",
+            "test3.txt": "5",
+            "test4.txt": "6"
+        }
+
+        for file, content in original_files.items():
+            with open(os.path.join(self.test_dir, file), "w") as f:
+                f.write(content)
+
+        result = rpm.handle_all_rpmnew_files(self.test_dir)
+
+        for file, content in expected_files.items():
+            full_filepath = os.path.join(self.test_dir, file)
+            # since test3.txt was not substituted, it should not be in the result
+            if file != "test3.txt":
+                self.assertTrue(full_filepath in result)
+            else:
+                self.assertFalse(full_filepath in result)
+
+            self.assertTrue(os.path.exists(full_filepath))
+            self.assertEqual(open(full_filepath).read(), content)
+
+        shutil.rmtree(self.test_dir)