MK8S-184 - Use modern Python library pathlib.Path instead of os.path

Author: ChengYanJin

salt/metalk8s/addons/prometheus-operator/deployed/files/restart-on-ca-change.py

@@ -3,26 +3,25 @@
 import os
 import sys
 from datetime import datetime, timezone
+from pathlib import Path
 
 import requests
 
 HASH_FILE_NAME = ".ca-hash-previous"
 
-SA_TOKEN = "/var/run/secrets/kubernetes.io/serviceaccount/token"
-SA_CA = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
+SA_TOKEN = Path("/var/run/secrets/kubernetes.io/serviceaccount/token")
+SA_CA = Path("/var/run/secrets/kubernetes.io/serviceaccount/ca.crt")
 K8S_API = "https://kubernetes.default.svc"
 
 
-def hash_file(file_path: str) -> str:
+def hash_file(file_path: Path) -> str:
     h = hashlib.sha256()
-    with open(file_path, "rb") as f:
-        h.update(f.read())
+    h.update(file_path.read_bytes())
     return h.hexdigest()
 
 
 def trigger_restart(namespace: str, deployment: str) -> None:
-    with open(SA_TOKEN) as f:
-        token = f.read()
+    token = SA_TOKEN.read_text()
     timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
     body = {
         "spec": {
@@ -47,24 +46,22 @@ def trigger_restart(namespace: str, deployment: str) -> None:
 
 
 def main() -> None:
-    ca_dir = os.environ["CA_DIR"]
-    ca_file = os.path.join(ca_dir, os.environ["CA_FILE_NAME"])
-    hash_file_path = os.path.join(ca_dir, HASH_FILE_NAME)
+    ca_dir = Path(os.environ["CA_DIR"])
+    ca_file = ca_dir / os.environ["CA_FILE_NAME"]
+    hash_file_path = ca_dir / HASH_FILE_NAME
 
-    if not os.path.exists(ca_file):
+    if not ca_file.exists():
         print(f"CA file {ca_file} does not exist, skipping")
         return
 
     current_hash = hash_file(ca_file)
 
-    if not os.path.exists(hash_file_path):
-        with open(hash_file_path, "w") as f:
-            f.write(current_hash)
+    if not hash_file_path.exists():
+        hash_file_path.write_text(current_hash)
         print("Initial CA load, skipping restart")
         return
 
-    with open(hash_file_path) as f:
-        previous_hash = f.read().strip()
+    previous_hash = hash_file_path.read_text().strip()
 
     if current_hash == previous_hash:
         return
@@ -82,8 +79,7 @@ def main() -> None:
         sys.exit(1)
 
     # Persist hash only after successful restart
-    with open(hash_file_path, "w") as f:
-        f.write(current_hash)
+    hash_file_path.write_text(current_hash)
     print(f"Rolling restart triggered for {deployment}")
 
 

salt/tests/unit/scripts/test_restart_on_ca_change.py

@@ -2,10 +2,12 @@
 
 import importlib.util
 import os
+import tempfile
+from pathlib import Path
 
 import requests
 from unittest import TestCase
-from unittest.mock import mock_open, patch
+from unittest.mock import patch
 
 # The script has a hyphenated filename, so we need importlib to load it
 _SCRIPT_PATH = os.path.join(
@@ -37,28 +39,36 @@ class TestHashFile(TestCase):
 
     def test_returns_sha256_hex(self):
         """hash_file returns a 64-char hex string."""
-        with patch("builtins.open", mock_open(read_data=b"cert-data")):
-            result = restart_on_ca_change.hash_file("/tmp/secrets/ca.crt")
+        with tempfile.NamedTemporaryFile() as f:
+            f.write(b"cert-data")
+            f.flush()
+            result = restart_on_ca_change.hash_file(Path(f.name))
         self.assertIsInstance(result, str)
         self.assertEqual(len(result), 64)
 
     def test_different_content_different_hash(self):
         """hash_file returns different hashes for different file contents."""
-        with patch("builtins.open", mock_open(read_data=b"cert-v1")):
-            hash_v1 = restart_on_ca_change.hash_file("/tmp/secrets/ca.crt")
+        with tempfile.NamedTemporaryFile() as f1, tempfile.NamedTemporaryFile() as f2:
+            f1.write(b"cert-v1")
+            f1.flush()
+            hash_v1 = restart_on_ca_change.hash_file(Path(f1.name))
 
-        with patch("builtins.open", mock_open(read_data=b"cert-v2")):
-            hash_v2 = restart_on_ca_change.hash_file("/tmp/secrets/ca.crt")
+            f2.write(b"cert-v2")
+            f2.flush()
+            hash_v2 = restart_on_ca_change.hash_file(Path(f2.name))
 
         self.assertNotEqual(hash_v1, hash_v2)
 
     def test_same_content_same_hash(self):
         """hash_file returns the same hash for the same content."""
-        with patch("builtins.open", mock_open(read_data=b"cert-data")):
-            hash_1 = restart_on_ca_change.hash_file("/tmp/secrets/ca.crt")
+        with tempfile.NamedTemporaryFile() as f1, tempfile.NamedTemporaryFile() as f2:
+            f1.write(b"cert-data")
+            f1.flush()
+            hash_1 = restart_on_ca_change.hash_file(Path(f1.name))
 
-        with patch("builtins.open", mock_open(read_data=b"cert-data")):
-            hash_2 = restart_on_ca_change.hash_file("/tmp/secrets/ca.crt")
+            f2.write(b"cert-data")
+            f2.flush()
+            hash_2 = restart_on_ca_change.hash_file(Path(f2.name))
 
         self.assertEqual(hash_1, hash_2)
 
@@ -67,49 +77,54 @@ def test_same_content_same_hash(self):
 class TestMain(TestCase):
     """Tests for main function."""
 
-    @patch("os.path.exists", return_value=False)
-    def test_ca_file_missing_skips(self, _mock_exists):
+    def test_ca_file_missing_skips(self):
         """main skips when CA file does not exist."""
-        with patch("builtins.print") as mock_print:
-            restart_on_ca_change.main()
-        mock_print.assert_called_once_with(
-            "CA file /tmp/secrets/ca.crt does not exist, skipping"
-        )
+        with tempfile.TemporaryDirectory() as tmpdir:
+            env = {**ENV_VARS, "CA_DIR": tmpdir}
+            with patch.dict(os.environ, env), patch("builtins.print") as mock_print:
+                restart_on_ca_change.main()
+            mock_print.assert_called_once_with(
+                f"CA file {Path(tmpdir) / 'ca.crt'} does not exist, skipping"
+            )
 
     def test_initial_load_skips_restart(self):
         """main writes hash and skips restart on initial load."""
-        # ca_file exists, hash_file_path does not
-        with patch(
-            "os.path.exists", side_effect=lambda p: p == "/tmp/secrets/ca.crt"
-        ), patch.object(
-            restart_on_ca_change, "hash_file", return_value="abc123"
-        ), patch(
-            "builtins.open", mock_open()
-        ), patch(
-            "builtins.print"
-        ) as mock_print:
-            restart_on_ca_change.main()
-        mock_print.assert_called_once_with("Initial CA load, skipping restart")
+        with tempfile.TemporaryDirectory() as tmpdir:
+            ca_file = Path(tmpdir) / "ca.crt"
+            ca_file.write_bytes(b"cert-data")
+            env = {**ENV_VARS, "CA_DIR": tmpdir}
+            with patch.dict(os.environ, env), patch("builtins.print") as mock_print:
+                restart_on_ca_change.main()
+            mock_print.assert_called_once_with("Initial CA load, skipping restart")
+            # Hash file should have been created
+            hash_file = Path(tmpdir) / ".ca-hash-previous"
+            self.assertTrue(hash_file.exists())
 
     def test_unchanged_hash_no_restart(self):
         """main does nothing when hash has not changed."""
-        with patch("os.path.exists", return_value=True), patch.object(
-            restart_on_ca_change, "hash_file", return_value="same-hash"
-        ), patch("builtins.open", mock_open(read_data="same-hash")), patch(
-            "builtins.print"
-        ) as mock_print:
-            restart_on_ca_change.main()
-        mock_print.assert_not_called()
+        with tempfile.TemporaryDirectory() as tmpdir:
+            ca_file = Path(tmpdir) / "ca.crt"
+            ca_file.write_bytes(b"cert-data")
+            # Pre-compute and write the hash
+            current_hash = restart_on_ca_change.hash_file(ca_file)
+            hash_file = Path(tmpdir) / ".ca-hash-previous"
+            hash_file.write_text(current_hash)
+            env = {**ENV_VARS, "CA_DIR": tmpdir}
+            with patch.dict(os.environ, env), patch("builtins.print") as mock_print:
+                restart_on_ca_change.main()
+            mock_print.assert_not_called()
 
     @patch.object(restart_on_ca_change, "trigger_restart")
     def test_changed_hash_triggers_restart(self, mock_restart):
         """main triggers restart when hash has changed."""
-        with patch("os.path.exists", return_value=True), patch.object(
-            restart_on_ca_change, "hash_file", return_value="new-hash"
-        ), patch("builtins.open", mock_open(read_data="old-hash")), patch(
-            "builtins.print"
-        ) as mock_print:
-            restart_on_ca_change.main()
+        with tempfile.TemporaryDirectory() as tmpdir:
+            ca_file = Path(tmpdir) / "ca.crt"
+            ca_file.write_bytes(b"cert-data")
+            hash_file = Path(tmpdir) / ".ca-hash-previous"
+            hash_file.write_text("old-hash")
+            env = {**ENV_VARS, "CA_DIR": tmpdir}
+            with patch.dict(os.environ, env), patch("builtins.print") as mock_print:
+                restart_on_ca_change.main()
 
         mock_restart.assert_called_once_with(
             "metalk8s-monitoring", "oauth2-proxy-prometheus"
@@ -119,62 +134,63 @@ def test_changed_hash_triggers_restart(self, mock_restart):
             mock_print.call_args[0][0],
         )
 
-    @patch("builtins.print")
     @patch.object(
         restart_on_ca_change,
         "trigger_restart",
         side_effect=requests.RequestException("refused"),
     )
-    @patch.object(restart_on_ca_change, "hash_file", return_value="new-hash")
-    @patch("os.path.exists", return_value=True)
-    def test_api_failure_exits_with_error(
-        self, _mock_exists, _mock_hash, _mock_restart, _mock_print
-    ):
+    def test_api_failure_exits_with_error(self, _mock_restart):
         """main exits with code 1 when the API call fails."""
-        with patch("builtins.open", mock_open(read_data="old-hash")):
-            with self.assertRaises(SystemExit) as ctx:
-                restart_on_ca_change.main()
+        with tempfile.TemporaryDirectory() as tmpdir:
+            ca_file = Path(tmpdir) / "ca.crt"
+            ca_file.write_bytes(b"cert-data")
+            hash_file = Path(tmpdir) / ".ca-hash-previous"
+            hash_file.write_text("old-hash")
+            env = {**ENV_VARS, "CA_DIR": tmpdir}
+            with patch.dict(os.environ, env), patch("builtins.print"):
+                with self.assertRaises(SystemExit) as ctx:
+                    restart_on_ca_change.main()
 
         self.assertEqual(ctx.exception.code, 1)
 
-    @patch("builtins.print")
     @patch.object(
         restart_on_ca_change,
         "trigger_restart",
         side_effect=requests.RequestException("refused"),
     )
-    @patch.object(restart_on_ca_change, "hash_file", return_value="new-hash")
-    @patch("os.path.exists", return_value=True)
-    def test_hash_not_persisted_on_api_failure(
-        self, _mock_exists, _mock_hash, _mock_restart, _mock_print
-    ):
+    def test_hash_not_persisted_on_api_failure(self, _mock_restart):
         """Hash file is not updated when the API call fails."""
-        mock_file = mock_open(read_data="old-hash")
-
-        with patch("builtins.open", mock_file):
-            try:
-                restart_on_ca_change.main()
-            except SystemExit:
-                pass
-
-        # Only the hash file read should have happened, no write
-        write_calls = mock_file().write.call_args_list
-        self.assertEqual(len(write_calls), 0)
+        with tempfile.TemporaryDirectory() as tmpdir:
+            ca_file = Path(tmpdir) / "ca.crt"
+            ca_file.write_bytes(b"cert-data")
+            hash_file = Path(tmpdir) / ".ca-hash-previous"
+            hash_file.write_text("old-hash")
+            env = {**ENV_VARS, "CA_DIR": tmpdir}
+            with patch.dict(os.environ, env), patch("builtins.print"):
+                try:
+                    restart_on_ca_change.main()
+                except SystemExit:
+                    pass
+
+            # Hash file should still contain the old hash
+            self.assertEqual(hash_file.read_text(), "old-hash")
 
 
 class TestTriggerRestart(TestCase):
     """Tests for trigger_restart function."""
 
     @patch.object(restart_on_ca_change.requests, "patch")
-    @patch(
-        "builtins.open",
-        mock_open(read_data="fake-token"),
-    )
     def test_sends_patch_request(self, mock_patch):
         """trigger_restart sends a PATCH to the K8s API."""
-        restart_on_ca_change.trigger_restart(
-            "metalk8s-monitoring", "oauth2-proxy-prometheus"
-        )
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".token") as f:
+            f.write("fake-token")
+            f.flush()
+            with patch.object(
+                restart_on_ca_change, "SA_TOKEN", Path(f.name)
+            ):
+                restart_on_ca_change.trigger_restart(
+                    "metalk8s-monitoring", "oauth2-proxy-prometheus"
+                )
 
         mock_patch.assert_called_once()
         args, kwargs = mock_patch.call_args